Coverage for python/lsst/images/fits/_output_archive.py: 72%
229 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-16 15:24 -0700
1# This file is part of lsst-images.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT file at the top-level directory of this distribution
7# for details of code ownership.
8#
9# Use of this source code is governed by a 3-clause BSD-style
10# license that can be found in the LICENSE file.
12from __future__ import annotations
14__all__ = ("FitsOutputArchive", "write")
16import dataclasses
17from collections.abc import Callable, Hashable, Iterator, Mapping
18from contextlib import contextmanager
19from pathlib import Path
20from typing import Any, Self
22import astropy.io.fits
23import astropy.table
24import astropy.time
25import numpy as np
26import pydantic
28from .._transforms import FrameSet
29from ..serialization import (
30 ArchiveTree,
31 ArrayReferenceModel,
32 ButlerInfo,
33 MetadataValue,
34 NestedOutputArchive,
35 NumberType,
36 OutputArchive,
37 TableColumnModel,
38 TableModel,
39 no_header_updates,
40)
41from ._common import (
42 JSON_COLUMN,
43 JSON_EXTNAME,
44 ExtensionHDU,
45 ExtensionKey,
46 FitsCompressionOptions,
47 FitsOpaqueMetadata,
48 PointerModel,
49 suppress_fits_card_warnings,
50)
52_FITS_FORMAT_VERSION = 1
53"""Container layout version for files written by `FitsOutputArchive`.
55Bumps when the on-disk FITS layout (HDU placement, INDX/JSON keyword schema)
56changes. Independent of any data-model ``SCHEMA_VERSION``.
57"""
60def _set_creation_date(header: astropy.io.fits.Header) -> None:
61 """Record the current UTC time in the ``DATE`` card of a FITS header.
63 Parameters
64 ----------
65 header
66 Header to modify in place.
68 Notes
69 -----
70 The FITS standard requires every HDU to record the UTC date and time its
71 header was created via a ``DATE`` card in ``YYYY-MM-DDThh:mm:ss[.sss]``
72 form. Any pre-existing ``DATE`` card (such as one preserved from an input
73 archive) is overwritten in place so the value reflects this write.
74 """
75 header.set("DATE", astropy.time.Time.now().fits, "UTC date this HDU was written.")
78def write(
79 obj: Any,
80 path: Path | str,
81 compression_options: Mapping[str, FitsCompressionOptions | None] | None = None,
82 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates,
83 compression_seed: int | None = None,
84 metadata: dict[str, MetadataValue] | None = None,
85 butler_info: ButlerInfo | None = None,
86) -> Any:
87 """Write an object with a ``serialize`` method to a FITS file.
89 Parameters
90 ----------
91 obj
92 Object with a ``serialize`` method to write.
93 path
94 Name of the file to write to. Must not already exist.
95 compression_options
96 Options for how to compress the FITS file, keyed by the name of
97 the attribute (with JSON pointer ``/`` separators for nested
98 attributes).
99 update_header
100 A callback that will be given the primary HDU FITS header and an
101 opportunity to modify it.
102 compression_seed
103 A FITS tile compression seed to use whenever the configured
104 compression seed is `None` or (for backwards compatibility) ``0``.
105 This value is then incremented every time it is used.
106 metadata
107 Additional metadata to save with the object. This will override any
108 flexible metadata carried by the object itself with the same keys.
109 butler_info
110 Butler information to store in the file.
112 Returns
113 -------
114 `.serialization.ArchiveTree`
115 The serialized representation of the object.
116 """
117 opaque_metadata = getattr(obj, "_opaque_metadata", None)
118 name = getattr(obj, "_archive_default_name", None)
119 with FitsOutputArchive.open(
120 path,
121 compression_options=compression_options,
122 opaque_metadata=opaque_metadata,
123 update_header=update_header,
124 compression_seed=compression_seed,
125 ) as archive:
126 tree = archive.serialize_direct(name, obj.serialize) if name is not None else obj.serialize(archive)
127 if metadata is not None: 127 ↛ 128line 127 didn't jump to line 128 because the condition on line 127 was never true
128 tree.metadata.update(metadata)
129 if butler_info is not None:
130 tree.butler_info = butler_info
131 archive.add_tree(tree)
132 return tree
135class FitsOutputArchive(OutputArchive[PointerModel]):
136 """An implementation of the `.serialization.OutputArchive` interface that
137 writes to FITS files.
139 Instances of this class should only be constructed via the `open`
140 context manager.
142 Parameters
143 ----------
144 hdu_list
145 HDU list that the archive writes its HDUs into.
146 compression_options
147 Options for how to compress the FITS file, keyed by the name of
148 the attribute (with JSON pointer ``/`` separators for nested
149 attributes).
150 opaque_metadata
151 Opaque metadata to carry through from the object being written.
152 compression_seed
153 A FITS tile compression seed to use whenever the configured
154 compression seed is `None` or (for backwards compatibility)
155 ``0``.
156 """
158 def __init__(
159 self,
160 hdu_list: astropy.io.fits.HDUList,
161 compression_options: Mapping[str, FitsCompressionOptions | None] | None = None,
162 opaque_metadata: Any = None,
163 compression_seed: int | None = None,
164 ) -> None:
165 super().__init__()
166 # JSON blobs for objects we've saved as pointers:
167 self._pointer_targets: list[bytes] = []
168 # Mapping from user provided key (e.g. id(some object)) to a table
169 # pointer to where we actually saved it:
170 self._pointers_by_key: dict[Hashable, PointerModel] = {}
171 self._hdu_list = hdu_list
172 self._primary_hdu = astropy.io.fits.PrimaryHDU()
173 self._primary_hdu.header.set("FMTVER", _FITS_FORMAT_VERSION, "FITS container layout version.")
174 self._primary_hdu.header.set("INDXADDR", 0, "Offset in bytes to the HDU index.")
175 self._primary_hdu.header.set("INDXSIZE", 0, "Size of the HDU index.")
176 self._primary_hdu.header.set("JSONADDR", 0, "Offset in bytes to the JSON tree HDU.")
177 self._primary_hdu.header.set("JSONSIZE", 0, "Size of the JSON tree HDU.")
178 self._compression_options = dict(compression_options) if compression_options is not None else {}
179 self._compression_seed = compression_seed
180 self._opaque_metadata = (
181 opaque_metadata if isinstance(opaque_metadata, FitsOpaqueMetadata) else FitsOpaqueMetadata()
182 )
183 if (opaque_primary_header := self._opaque_metadata.headers.get(ExtensionKey())) is not None:
184 self._primary_hdu.header.extend(opaque_primary_header)
185 self._hdu_list.append(self._primary_hdu)
186 self._json_hdu_added: bool = False
187 self._frame_sets: list[tuple[FrameSet, PointerModel]] = []
189 @classmethod
190 @contextmanager
191 def open(
192 cls,
193 filename: Path | str,
194 compression_options: Mapping[str, FitsCompressionOptions | None] | None = None,
195 opaque_metadata: Any = None,
196 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates,
197 compression_seed: int | None = None,
198 ) -> Iterator[Self]:
199 """Create an output archive that writes to the given file.
201 Parameters
202 ----------
203 filename
204 Name of the file to write to. Must not already exist.
205 compression_options
206 Options for how to compress the FITS file, keyed by the name of
207 the attribute (with JSON pointer ``/`` separators for nested
208 attributes).
209 opaque_metadata
210 Metadata read from an input archive along with the object being
211 written now. Ignored if the metadata is not from a FITS archive.
212 update_header
213 A callback that will be given the primary HDU FITS header and an
214 opportunity to modify it.
215 compression_seed
216 A FITS tile compression seed to use whenever the configured
217 compression seed is `None` or (for backwards compatibility) ``0``.
218 This value is then incremented every time it is used.
220 Returns
221 -------
222 `contextlib.AbstractContextManager` [`FitsOutputArchive`]
223 A context manager that returns a `FitsOutputArchive` when entered.
224 """
225 # Astropy auto-fixes over-long keywords (HIERARCH) and comments
226 # (truncation) as cards are created and written, but doesn't honor its
227 # own output_verify option for these warnings, so we filter them out
228 # across the write of the main HDUs.
229 with suppress_fits_card_warnings(), astropy.io.fits.open(filename, mode="append") as hdu_list:
230 if hdu_list: 230 ↛ 231line 230 didn't jump to line 231 because the condition on line 230 was never true
231 raise OSError(f"File {filename!r} already exists.")
232 archive = cls(hdu_list, compression_options, opaque_metadata, compression_seed=compression_seed)
233 update_header(hdu_list[0].header)
234 # Set the creation date after update_header so a caller's callback
235 # cannot leave a stale DATE in the primary header; this card must
236 # always record the time of this write.
237 _set_creation_date(hdu_list[0].header)
238 yield archive
239 if not archive._json_hdu_added: 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 raise RuntimeError("Write context exited without 'add_tree' being called.")
241 hdu_list.flush()
242 # This multi-open dance is necessary to get Astropy to tell us the
243 # byte addresses of the HDUs. Hopefully we can get an upstream change
244 # make this unnecessary at some point.
245 with astropy.io.fits.open(filename, mode="readonly", disable_image_compression=True) as hdu_list:
246 index_hdu = cls._make_index_table(hdu_list)
247 with astropy.io.fits.open(filename, mode="append") as hdu_list:
248 hdu_list.append(index_hdu)
249 json_bytes = _HDUBytes.from_index_row(index_hdu.data[-1])
250 index_bytes = _HDUBytes.from_write_hdu(index_hdu)
251 # Update the primary HDU with the address and size of the index and
252 # JSON HDUs, and rewrite just that. We do this write manually, since
253 # astropy's docs on its 'update' mode are scarce and it's not obvious
254 # whether we can guarantee it won't rewrite the whole file if we edit
255 # the primary header.
256 archive._primary_hdu.header["INDXADDR"] = index_bytes.header_address
257 archive._primary_hdu.header["INDXSIZE"] = index_bytes.size
258 archive._primary_hdu.header["JSONADDR"] = json_bytes.header_address
259 archive._primary_hdu.header["JSONSIZE"] = json_bytes.size
260 with open(filename, "r+b") as stream:
261 stream.write(archive._primary_hdu.header.tostring().encode())
263 def serialize_direct[T: pydantic.BaseModel | None](
264 self, name: str, serializer: Callable[[OutputArchive[PointerModel]], T]
265 ) -> T:
266 nested = NestedOutputArchive[PointerModel](name, self)
267 return serializer(nested)
269 def serialize_pointer[T: ArchiveTree](
270 self, name: str, serializer: Callable[[OutputArchive[PointerModel]], T], key: Hashable
271 ) -> PointerModel:
272 if (pointer := self._pointers_by_key.get(key)) is not None: 272 ↛ 273line 272 didn't jump to line 273 because the condition on line 272 was never true
273 return pointer
274 # Rooting the nested archive at "" follows the NestedOutputArchive
275 # convention that root paths are absolute, but serialize_direct roots
276 # at the bare name, so an array written by a pointer target would get
277 # a "/"-prefixed EXTNAME (e.g. "/DATA"). No serializer currently
278 # writes arrays inside pointer targets; if one ever does, consider
279 # rooting at ``name`` instead, as the NDF backend does.
280 model = self.serialize_direct("", serializer)
281 json_bytes = model.model_dump_json().encode()
282 self._pointer_targets.append(json_bytes)
283 pointer = PointerModel(
284 column=TableColumnModel(
285 name=JSON_COLUMN,
286 data=ArrayReferenceModel(
287 source=f"fits:{JSON_EXTNAME}[1]",
288 shape=[len(json_bytes)],
289 datatype=NumberType.uint8,
290 ),
291 ),
292 row=len(self._pointer_targets),
293 )
294 self._pointers_by_key[key] = pointer
295 return pointer
297 def serialize_frame_set[T: ArchiveTree](
298 self, name: str, frame_set: FrameSet, serializer: Callable[[OutputArchive], T], key: Hashable
299 ) -> PointerModel:
300 # Docstring inherited.
301 pointer = self.serialize_pointer(name, serializer, key)
302 self._frame_sets.append((frame_set, pointer))
303 return pointer
305 def iter_frame_sets(self) -> Iterator[tuple[FrameSet, PointerModel]]:
306 return iter(self._frame_sets)
308 def add_array(
309 self,
310 array: np.ndarray,
311 *,
312 name: str | None = None,
313 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates,
314 tile_shape: tuple[int, ...] | None = None,
315 options_name: str | None = None,
316 ) -> ArrayReferenceModel:
317 if name is None: 317 ↛ 318line 317 didn't jump to line 318 because the condition on line 317 was never true
318 raise RuntimeError("Cannot save array with name=None unless it is nested.")
319 name, version = self._register_name(name)
320 extname = name.upper()
321 hdu = self._opaque_metadata.maybe_use_precompressed(extname)
322 if hdu is None: 322 ↛ 329line 322 didn't jump to line 329 because the condition on line 322 was always true
323 if options_name is None: 323 ↛ 325line 323 didn't jump to line 325 because the condition on line 323 was always true
324 options_name = name
325 if (compression_options := self._get_compression_options(options_name, tile_shape)) is not None: 325 ↛ 328line 325 didn't jump to line 328 because the condition on line 325 was always true
326 hdu = compression_options.make_hdu(array, name=extname)
327 else:
328 hdu = astropy.io.fits.ImageHDU(array, name=extname)
329 key = self._add_hdu(hdu, version, update_header)
330 return ArrayReferenceModel(
331 source=str(key),
332 shape=list(array.shape),
333 datatype=NumberType.from_numpy(array.dtype),
334 )
336 def add_table(
337 self,
338 table: astropy.table.Table,
339 *,
340 name: str | None = None,
341 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates,
342 ) -> TableModel:
343 if name is None:
344 raise RuntimeError("Cannot save table with name=None unless it is nested.")
345 name, version = self._register_name(name)
346 extname = name.upper()
347 hdu: astropy.io.fits.BinTableHDU = astropy.io.fits.table_to_hdu(table, name=extname)
348 # Extract column information directly from the input array, not the
349 # data in the binary table HDU, because we want to assume as little as
350 # possible about where Astropy does uint -> TZERO stuff.
351 columns = TableColumnModel.from_table(table)
352 key = self._add_hdu(hdu, version, update_header)
353 for n, c in enumerate(columns, start=1):
354 assert isinstance(c.data, ArrayReferenceModel)
355 c.data.source = f"{key}[{n}]"
356 return TableModel(columns=columns, meta=table.meta)
358 def add_structured_array(
359 self,
360 array: np.ndarray,
361 *,
362 name: str | None = None,
363 units: Mapping[str, astropy.units.Unit] | None = None,
364 descriptions: Mapping[str, str] | None = None,
365 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates,
366 ) -> TableModel:
367 if name is None:
368 raise RuntimeError("Cannot save structured array with name=None unless it is nested.")
369 name, version = self._register_name(name)
370 extname = name.upper()
371 # Extract column information directly from the input array, not the
372 # data in the binary table HDU, because we want to assume as little as
373 # possible about where Astropy does uint -> TZERO stuff.
374 columns = TableColumnModel.from_record_dtype(array.dtype)
375 hdu = astropy.io.fits.BinTableHDU(array, name=extname)
376 if units is not None:
377 for c in columns:
378 c.unit = units.get(c.name)
379 if descriptions is not None:
380 for c in columns:
381 c.description = descriptions.get(c.name, "")
382 key = self._add_hdu(hdu, version, update_header)
383 for n, c in enumerate(columns, start=1):
384 assert isinstance(c.data, ArrayReferenceModel)
385 c.data.source = f"{key}[{n}]"
386 return TableModel(columns=columns)
388 def _add_hdu(
389 self,
390 hdu: astropy.io.fits.ImageHDU | astropy.io.fits.CompImageHDU | astropy.io.fits.BinTableHDU,
391 version: int,
392 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates,
393 ) -> ExtensionKey:
394 key = ExtensionKey(hdu.name, version)
395 key.check()
396 if version > 1:
397 hdu.header["EXTVER"] = version
398 update_header(hdu.header)
399 if (opaque_headers := self._opaque_metadata.headers.get(key)) is not None:
400 hdu.header.extend(opaque_headers)
401 _set_creation_date(hdu.header)
402 self._hdu_list.append(hdu)
403 return key
405 def add_tree(self, tree: ArchiveTree) -> None:
406 """Write the JSON tree to the archive.
408 This method must be called exactly once, just before the `open` context
409 is exited.
411 Parameters
412 ----------
413 tree
414 Pydantic model that represents the tree.
415 """
416 self._primary_hdu.header.set("DATAMODL", tree.schema_url, "Schema URL.")
417 json_hdu = astropy.io.fits.BinTableHDU.from_columns(
418 [astropy.io.fits.Column(JSON_COLUMN, "PB")],
419 nrows=len(self._pointer_targets) + 1,
420 name=JSON_EXTNAME,
421 )
422 json_hdu.data[0][JSON_COLUMN] = np.frombuffer(tree.model_dump_json().encode(), dtype=np.byte)
423 for n, json_target_data in enumerate(self._pointer_targets):
424 json_hdu.data[n + 1][JSON_COLUMN] = np.frombuffer(json_target_data, dtype=np.byte)
425 _set_creation_date(json_hdu.header)
426 self._hdu_list.append(json_hdu)
427 self._json_hdu_added = True
429 def _get_compression_options(
430 self, name: str, tile_shape: tuple[int, ...] | None
431 ) -> FitsCompressionOptions | None:
432 result = self._compression_options.get(name, FitsCompressionOptions.DEFAULT)
433 if result is None: 433 ↛ 434line 433 didn't jump to line 434 because the condition on line 433 was never true
434 return result
435 if tile_shape is not None and result.tile_shape is None: 435 ↛ 436line 435 didn't jump to line 436 because the condition on line 435 was never true
436 result = result.model_copy(update={"tile_shape": tile_shape})
437 if result.quantization is None:
438 return result
439 if self._compression_seed is not None and not result.quantization.seed: 439 ↛ 450line 439 didn't jump to line 450 because the condition on line 439 was always true
440 result = result.model_copy(
441 update={
442 "quantization": result.quantization.model_copy(update={"seed": self._compression_seed})
443 }
444 )
445 self._compression_seed += 1
446 if self._compression_seed > 10000: 446 ↛ 447line 446 didn't jump to line 447 because the condition on line 446 was never true
447 self._compression_seed = 1
448 # MyPy can tell that result.quantization is not None in the 'if', but
449 # forgets that by this 'else':
450 elif result.quantization.seed is None: # type: ignore[union-attr]
451 raise RuntimeError("No quantization seed provided.")
452 return result
454 @staticmethod
455 def _make_index_table(hdu_list: astropy.io.fits.HDUList) -> astropy.io.fits.BinTableHDU:
456 # We use a fixed-length string for the EXTNAME column; it might be
457 # better to use a variable-length array, but I have not been able to
458 # figure out how to get astropy to accept a string for the the
459 # character (TFORM='A') variant of that. And that's only better if the
460 # EXTNAMEs get super long, which is not likely (but maybe something to
461 # guard against).
462 max_name_size = max(len(hdu.header.get("EXTNAME", "")) for hdu in hdu_list)
463 # Attach the ZIMAGE values as a boolean array on the Column itself.
464 # Astropy only fills a logical column's raw bytes with 'F' as the
465 # default (rather than NULL, 0x00) when the bool array is present at
466 # construction; assigning False element-wise after construction leaves
467 # the byte as NULL under Astropy 8.0.0, which warns on every read.
468 # Should be fixed in v8.0.1.
469 # See https://github.com/astropy/astropy/pull/19939
470 # but pre-allocating the entire column works in v7 and v8.
471 zimage = np.array([hdu.header.get("ZIMAGE", False) for hdu in hdu_list], dtype=bool)
472 index_hdu = astropy.io.fits.BinTableHDU.from_columns(
473 [
474 astropy.io.fits.Column("EXTNAME", f"A{max_name_size}"),
475 astropy.io.fits.Column("EXTVER", "J"),
476 astropy.io.fits.Column("XTENSION", "A8"),
477 astropy.io.fits.Column("ZIMAGE", "L", array=zimage),
478 ]
479 + _HDUBytes.get_index_hdu_columns(),
480 nrows=len(hdu_list),
481 name="INDEX",
482 )
483 hdu: ExtensionHDU | astropy.io.fits.PrimaryHDU
484 for n, hdu in enumerate(hdu_list):
485 index_hdu.data[n]["EXTNAME"] = hdu.header.get("EXTNAME", "")
486 index_hdu.data[n]["EXTVER"] = hdu.header.get("EXTVER", 1)
487 index_hdu.data[n]["XTENSION"] = hdu.header.get("XTENSION", "IMAGE")
488 bytes = _HDUBytes.from_read_hdu(hdu)
489 bytes.update_index_row(index_hdu.data[n])
490 _set_creation_date(index_hdu.header)
491 return index_hdu
494@dataclasses.dataclass
495class _HDUBytes:
496 """A struct that records the byte offsets into a FITS HDU."""
498 @classmethod
499 def from_write_hdu(cls, hdu: astropy.io.fits.PrimaryHDU | ExtensionHDU) -> Self:
500 """Construct from an Astropy HDU instance that has just been written.
502 Parameters
503 ----------
504 hdu
505 An Astropy HDU object.
507 Returns
508 -------
509 hdu_bytes
510 Struct with byte offsets.
512 Notes
513 -----
514 This method relies on internal Astropy attributes and does not work on
515 CompImageHDU objects.
516 """
517 # This is implemented by accessing private Astropy attributes because
518 # it turns out that's much more reliable than the public fileinfo()
519 # method, which seems to always return a dict with `None` entries or
520 # raise; it looks buggy, but docs are scarce enough that it's not clear
521 # what the right behavior is supposed to be.
522 if (header_address := getattr(hdu, "_header_offset", None)) is None: 522 ↛ 523line 522 didn't jump to line 523 because the condition on line 522 was never true
523 raise RuntimeError("Failed to get Astropy's _header_offset.")
524 if (data_address := getattr(hdu, "_data_offset", None)) is None: 524 ↛ 525line 524 didn't jump to line 525 because the condition on line 524 was never true
525 raise RuntimeError("Failed to get Astropy's _data_offset.")
526 if (data_size := getattr(hdu, "_data_size", None)) is None: 526 ↛ 527line 526 didn't jump to line 527 because the condition on line 526 was never true
527 raise RuntimeError("Failed to get Astropy's _data_size.")
528 return cls(header_address, data_address, data_size)
530 @classmethod
531 def from_read_hdu(cls, hdu: astropy.io.fits.PrimaryHDU | ExtensionHDU) -> Self:
532 """Construct from an Astropy HDU instance that has just been read.
534 Parameters
535 ----------
536 hdu
537 An Astropy HDU object.
539 Returns
540 -------
541 hdu_bytes
542 Struct with byte offsets.
543 """
544 info = hdu.fileinfo()
545 header_address = info["hdrLoc"]
546 data_address = info["datLoc"]
547 data_size = info["datSpan"]
548 return cls(header_address, data_address, data_size)
550 @classmethod
551 def from_index_row(cls, row: np.void) -> Self:
552 """Construct from a row of the index HDU.
554 Parameters
555 ----------
556 row
557 A Numpy struct-like scalar.
559 Returns
560 -------
561 hdu_bytes
562 Struct with byte offsets.
563 """
564 return cls(
565 header_address=int(row["HDRADDR"]),
566 data_address=int(row["DATADDR"]),
567 data_size=int(row["DATSIZE"]),
568 )
570 @staticmethod
571 def get_index_hdu_columns() -> list[astropy.io.fits.Column]:
572 """Return the definitions of the columns this class gets and sets
573 from the index HDU.
575 Returns
576 -------
577 columns
578 A `list` of `astropy.io.fits.Column` objects that represent the
579 header address, data address, and data size.
580 """
581 return [
582 astropy.io.fits.Column("HDRADDR", "K"),
583 astropy.io.fits.Column("DATADDR", "K"),
584 astropy.io.fits.Column("DATSIZE", "K"),
585 ]
587 header_address: int
588 """Offset from the beginning of the start of the file to the header of this
589 HDU, in bytes.
590 """
592 data_address: int
593 """Offset from the beginning of the start of the file to the data section
594 of this HDU, in bytes.
595 """
597 data_size: int
598 """Size of the data section in bytes."""
600 @property
601 def header_size(self) -> int:
602 """Size of the header in bytes."""
603 return self.data_address - self.header_address
605 @property
606 def end_address(self) -> int:
607 """Offset in bytes from the start of the file to the end of the HDU."""
608 return self.data_address + self.data_size
610 @property
611 def size(self) -> int:
612 """Total size of this HDU in bytes."""
613 return self.data_size + self.data_address - self.header_address
615 def update_index_row(self, row: np.void) -> None:
616 """Set the values of a row of the index HDU from this strut.
618 Parameters
619 ----------
620 row
621 A Numpy struct-like scalar to modify in place.
622 """
623 row["HDRADDR"] = self.header_address
624 row["DATADDR"] = self.data_address
625 row["DATSIZE"] = self.data_size