Coverage for python/lsst/images/fits/_output_archive.py: 51%

224 statements  

« prev     ^ index     » next       coverage.py v7.15.2, created at 2026-07-24 02:46 +0000

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. 

11 

12from __future__ import annotations 

13 

14__all__ = ("FitsOutputArchive", "write") 

15 

16import dataclasses 

17from collections.abc import Callable, Hashable, Iterator, Mapping 

18from contextlib import contextmanager 

19from pathlib import Path 

20from typing import Any, Self 

21 

22import astropy.io.fits 

23import astropy.table 

24import astropy.time 

25import numpy as np 

26import pydantic 

27 

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) 

51 

52_FITS_FORMAT_VERSION = 1 

53"""Container layout version for files written by `FitsOutputArchive`. 

54 

55Bumps when the on-disk FITS layout (HDU placement, INDX/JSON keyword schema) 

56changes. Independent of any data-model ``SCHEMA_VERSION``. 

57""" 

58 

59 

60def _set_creation_date(header: astropy.io.fits.Header) -> None: 

61 """Record the current UTC time in the ``DATE`` card of a FITS header. 

62 

63 Parameters 

64 ---------- 

65 header 

66 Header to modify in place. 

67 

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.") 

76 

77 

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. 

88 

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. 

111 

112 Returns 

113 ------- 

114 `.serialization.ArchiveTree` 

115 The serialized representation of the object. 

116 """ 

117 opaque_metadata = getattr(obj, "_opaque_metadata", None) 

118 with FitsOutputArchive.open( 

119 path, 

120 compression_options=compression_options, 

121 opaque_metadata=opaque_metadata, 

122 update_header=update_header, 

123 compression_seed=compression_seed, 

124 ) as archive: 

125 tree = archive.serialize_root(obj, metadata, butler_info) 

126 archive.add_tree(tree) 

127 return tree 

128 

129 

130class FitsOutputArchive(OutputArchive[PointerModel]): 

131 """An implementation of the `.serialization.OutputArchive` interface that 

132 writes to FITS files. 

133 

134 Instances of this class should only be constructed via the `open` 

135 context manager. 

136 

137 Parameters 

138 ---------- 

139 hdu_list 

140 HDU list that the archive writes its HDUs into. 

141 compression_options 

142 Options for how to compress the FITS file, keyed by the name of 

143 the attribute (with JSON pointer ``/`` separators for nested 

144 attributes). 

145 opaque_metadata 

146 Opaque metadata to carry through from the object being written. 

147 compression_seed 

148 A FITS tile compression seed to use whenever the configured 

149 compression seed is `None` or (for backwards compatibility) 

150 ``0``. 

151 """ 

152 

153 def __init__( 

154 self, 

155 hdu_list: astropy.io.fits.HDUList, 

156 compression_options: Mapping[str, FitsCompressionOptions | None] | None = None, 

157 opaque_metadata: Any = None, 

158 compression_seed: int | None = None, 

159 ) -> None: 

160 super().__init__() 

161 # JSON blobs for objects we've saved as pointers: 

162 self._pointer_targets: list[bytes] = [] 

163 # Mapping from user provided key (e.g. id(some object)) to a table 

164 # pointer to where we actually saved it: 

165 self._pointers_by_key: dict[Hashable, PointerModel] = {} 

166 self._hdu_list = hdu_list 

167 self._primary_hdu = astropy.io.fits.PrimaryHDU() 

168 self._primary_hdu.header.set("FMTVER", _FITS_FORMAT_VERSION, "FITS container layout version.") 

169 self._primary_hdu.header.set("INDXADDR", 0, "Offset in bytes to the HDU index.") 

170 self._primary_hdu.header.set("INDXSIZE", 0, "Size of the HDU index.") 

171 self._primary_hdu.header.set("JSONADDR", 0, "Offset in bytes to the JSON tree HDU.") 

172 self._primary_hdu.header.set("JSONSIZE", 0, "Size of the JSON tree HDU.") 

173 self._compression_options = dict(compression_options) if compression_options is not None else {} 

174 self._compression_seed = compression_seed 

175 self._opaque_metadata = ( 

176 opaque_metadata if isinstance(opaque_metadata, FitsOpaqueMetadata) else FitsOpaqueMetadata() 

177 ) 

178 if (opaque_primary_header := self._opaque_metadata.headers.get(ExtensionKey())) is not None: 

179 self._primary_hdu.header.extend(opaque_primary_header) 

180 self._hdu_list.append(self._primary_hdu) 

181 self._json_hdu_added: bool = False 

182 self._frame_sets: list[tuple[FrameSet, PointerModel]] = [] 

183 

184 @classmethod 

185 @contextmanager 

186 def open( 

187 cls, 

188 filename: Path | str, 

189 compression_options: Mapping[str, FitsCompressionOptions | None] | None = None, 

190 opaque_metadata: Any = None, 

191 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

192 compression_seed: int | None = None, 

193 ) -> Iterator[Self]: 

194 """Create an output archive that writes to the given file. 

195 

196 Parameters 

197 ---------- 

198 filename 

199 Name of the file to write to. Must not already exist. 

200 compression_options 

201 Options for how to compress the FITS file, keyed by the name of 

202 the attribute (with JSON pointer ``/`` separators for nested 

203 attributes). 

204 opaque_metadata 

205 Metadata read from an input archive along with the object being 

206 written now. Ignored if the metadata is not from a FITS archive. 

207 update_header 

208 A callback that will be given the primary HDU FITS header and an 

209 opportunity to modify it. 

210 compression_seed 

211 A FITS tile compression seed to use whenever the configured 

212 compression seed is `None` or (for backwards compatibility) ``0``. 

213 This value is then incremented every time it is used. 

214 

215 Returns 

216 ------- 

217 `contextlib.AbstractContextManager` [`FitsOutputArchive`] 

218 A context manager that returns a `FitsOutputArchive` when entered. 

219 """ 

220 # Astropy auto-fixes over-long keywords (HIERARCH) and comments 

221 # (truncation) as cards are created and written, but doesn't honor its 

222 # own output_verify option for these warnings, so we filter them out 

223 # across the write of the main HDUs. 

224 with suppress_fits_card_warnings(), astropy.io.fits.open(filename, mode="append") as hdu_list: 

225 if hdu_list: 225 ↛ 226line 225 didn't jump to line 226 because the condition on line 225 was never true

226 raise OSError(f"File {filename!r} already exists.") 

227 archive = cls(hdu_list, compression_options, opaque_metadata, compression_seed=compression_seed) 

228 update_header(hdu_list[0].header) 

229 # Set the creation date after update_header so a caller's callback 

230 # cannot leave a stale DATE in the primary header; this card must 

231 # always record the time of this write. 

232 _set_creation_date(hdu_list[0].header) 

233 yield archive 

234 if not archive._json_hdu_added: 234 ↛ 235line 234 didn't jump to line 235 because the condition on line 234 was never true

235 raise RuntimeError("Write context exited without 'add_tree' being called.") 

236 hdu_list.flush() 

237 # This multi-open dance is necessary to get Astropy to tell us the 

238 # byte addresses of the HDUs. Hopefully we can get an upstream change 

239 # make this unnecessary at some point. 

240 with astropy.io.fits.open(filename, mode="readonly", disable_image_compression=True) as hdu_list: 

241 index_hdu = cls._make_index_table(hdu_list) 

242 with astropy.io.fits.open(filename, mode="append") as hdu_list: 

243 hdu_list.append(index_hdu) 

244 json_bytes = _HDUBytes.from_index_row(index_hdu.data[-1]) 

245 index_bytes = _HDUBytes.from_write_hdu(index_hdu) 

246 # Update the primary HDU with the address and size of the index and 

247 # JSON HDUs, and rewrite just that. We do this write manually, since 

248 # astropy's docs on its 'update' mode are scarce and it's not obvious 

249 # whether we can guarantee it won't rewrite the whole file if we edit 

250 # the primary header. 

251 archive._primary_hdu.header["INDXADDR"] = index_bytes.header_address 

252 archive._primary_hdu.header["INDXSIZE"] = index_bytes.size 

253 archive._primary_hdu.header["JSONADDR"] = json_bytes.header_address 

254 archive._primary_hdu.header["JSONSIZE"] = json_bytes.size 

255 with open(filename, "r+b") as stream: 

256 stream.write(archive._primary_hdu.header.tostring().encode()) 

257 

258 def serialize_direct[T: pydantic.BaseModel | None]( 

259 self, name: str, serializer: Callable[[OutputArchive[PointerModel]], T] 

260 ) -> T: 

261 nested = NestedOutputArchive[PointerModel](name, self) 

262 return serializer(nested) 

263 

264 def serialize_pointer[T: ArchiveTree]( 

265 self, name: str, serializer: Callable[[OutputArchive[PointerModel]], T], key: Hashable 

266 ) -> PointerModel: 

267 if (pointer := self._pointers_by_key.get(key)) is not None: 267 ↛ 268line 267 didn't jump to line 268 because the condition on line 267 was never true

268 return pointer 

269 # Rooting the nested archive at "" follows the NestedOutputArchive 

270 # convention that root paths are absolute, but serialize_direct roots 

271 # at the bare name, so an array written by a pointer target would get 

272 # a "/"-prefixed EXTNAME (e.g. "/DATA"). No serializer currently 

273 # writes arrays inside pointer targets; if one ever does, consider 

274 # rooting at ``name`` instead, as the NDF backend does. 

275 model = self.serialize_direct("", serializer) 

276 json_bytes = model.model_dump_json().encode() 

277 self._pointer_targets.append(json_bytes) 

278 pointer = PointerModel( 

279 column=TableColumnModel( 

280 name=JSON_COLUMN, 

281 data=ArrayReferenceModel( 

282 source=f"fits:{JSON_EXTNAME}[1]", 

283 shape=[len(json_bytes)], 

284 datatype=NumberType.uint8, 

285 ), 

286 ), 

287 row=len(self._pointer_targets), 

288 ) 

289 self._pointers_by_key[key] = pointer 

290 return pointer 

291 

292 def serialize_frame_set[T: ArchiveTree]( 

293 self, name: str, frame_set: FrameSet, serializer: Callable[[OutputArchive], T], key: Hashable 

294 ) -> PointerModel: 

295 # Docstring inherited. 

296 pointer = self.serialize_pointer(name, serializer, key) 

297 self._frame_sets.append((frame_set, pointer)) 

298 return pointer 

299 

300 def iter_frame_sets(self) -> Iterator[tuple[FrameSet, PointerModel]]: 

301 return iter(self._frame_sets) 

302 

303 def add_array( 

304 self, 

305 array: np.ndarray, 

306 *, 

307 name: str | None = None, 

308 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

309 tile_shape: tuple[int, ...] | None = None, 

310 options_name: str | None = None, 

311 ) -> ArrayReferenceModel: 

312 if name is None: 312 ↛ 313line 312 didn't jump to line 313 because the condition on line 312 was never true

313 raise RuntimeError("Cannot save array with name=None unless it is nested.") 

314 name, version = self._register_name(name) 

315 extname = name.upper() 

316 hdu = self._opaque_metadata.maybe_use_precompressed(extname) 

317 if hdu is None: 317 ↛ 324line 317 didn't jump to line 324 because the condition on line 317 was always true

318 if options_name is None: 318 ↛ 320line 318 didn't jump to line 320 because the condition on line 318 was always true

319 options_name = name 

320 if (compression_options := self._get_compression_options(options_name, tile_shape)) is not None: 320 ↛ 323line 320 didn't jump to line 323 because the condition on line 320 was always true

321 hdu = compression_options.make_hdu(array, name=extname) 

322 else: 

323 hdu = astropy.io.fits.ImageHDU(array, name=extname) 

324 key = self._add_hdu(hdu, version, update_header) 

325 return ArrayReferenceModel( 

326 source=str(key), 

327 shape=list(array.shape), 

328 datatype=NumberType.from_numpy(array.dtype), 

329 ) 

330 

331 def add_table( 

332 self, 

333 table: astropy.table.Table, 

334 *, 

335 name: str | None = None, 

336 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

337 ) -> TableModel: 

338 if name is None: 

339 raise RuntimeError("Cannot save table with name=None unless it is nested.") 

340 name, version = self._register_name(name) 

341 extname = name.upper() 

342 hdu: astropy.io.fits.BinTableHDU = astropy.io.fits.table_to_hdu(table, name=extname) 

343 # Extract column information directly from the input array, not the 

344 # data in the binary table HDU, because we want to assume as little as 

345 # possible about where Astropy does uint -> TZERO stuff. 

346 columns = TableColumnModel.from_table(table) 

347 key = self._add_hdu(hdu, version, update_header) 

348 for n, c in enumerate(columns, start=1): 

349 assert isinstance(c.data, ArrayReferenceModel) 

350 c.data.source = f"{key}[{n}]" 

351 return TableModel(columns=columns, meta=table.meta) 

352 

353 def add_structured_array( 

354 self, 

355 array: np.ndarray, 

356 *, 

357 name: str | None = None, 

358 units: Mapping[str, astropy.units.Unit] | None = None, 

359 descriptions: Mapping[str, str] | None = None, 

360 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

361 ) -> TableModel: 

362 if name is None: 

363 raise RuntimeError("Cannot save structured array with name=None unless it is nested.") 

364 name, version = self._register_name(name) 

365 extname = name.upper() 

366 # Extract column information directly from the input array, not the 

367 # data in the binary table HDU, because we want to assume as little as 

368 # possible about where Astropy does uint -> TZERO stuff. 

369 columns = TableColumnModel.from_record_dtype(array.dtype) 

370 hdu = astropy.io.fits.BinTableHDU(array, name=extname) 

371 if units is not None: 

372 for c in columns: 

373 c.unit = units.get(c.name) 

374 if descriptions is not None: 

375 for c in columns: 

376 c.description = descriptions.get(c.name, "") 

377 key = self._add_hdu(hdu, version, update_header) 

378 for n, c in enumerate(columns, start=1): 

379 assert isinstance(c.data, ArrayReferenceModel) 

380 c.data.source = f"{key}[{n}]" 

381 return TableModel(columns=columns) 

382 

383 def _add_hdu( 

384 self, 

385 hdu: astropy.io.fits.ImageHDU | astropy.io.fits.CompImageHDU | astropy.io.fits.BinTableHDU, 

386 version: int, 

387 update_header: Callable[[astropy.io.fits.Header], None] = no_header_updates, 

388 ) -> ExtensionKey: 

389 key = ExtensionKey(hdu.name, version) 

390 key.check() 

391 if version > 1: 

392 hdu.header["EXTVER"] = version 

393 update_header(hdu.header) 

394 if (opaque_headers := self._opaque_metadata.headers.get(key)) is not None: 

395 hdu.header.extend(opaque_headers) 

396 _set_creation_date(hdu.header) 

397 self._hdu_list.append(hdu) 

398 return key 

399 

400 def add_tree(self, tree: ArchiveTree) -> None: 

401 """Write the JSON tree to the archive. 

402 

403 This method must be called exactly once, just before the `open` context 

404 is exited. 

405 

406 Parameters 

407 ---------- 

408 tree 

409 Pydantic model that represents the tree. 

410 """ 

411 self._primary_hdu.header.set("DATAMODL", tree.schema_url, "Schema URL.") 

412 json_hdu = astropy.io.fits.BinTableHDU.from_columns( 

413 [astropy.io.fits.Column(JSON_COLUMN, "PB")], 

414 nrows=len(self._pointer_targets) + 1, 

415 name=JSON_EXTNAME, 

416 ) 

417 json_hdu.data[0][JSON_COLUMN] = np.frombuffer(tree.model_dump_json().encode(), dtype=np.byte) 

418 for n, json_target_data in enumerate(self._pointer_targets): 

419 json_hdu.data[n + 1][JSON_COLUMN] = np.frombuffer(json_target_data, dtype=np.byte) 

420 _set_creation_date(json_hdu.header) 

421 self._hdu_list.append(json_hdu) 

422 self._json_hdu_added = True 

423 

424 def _get_compression_options( 

425 self, name: str, tile_shape: tuple[int, ...] | None 

426 ) -> FitsCompressionOptions | None: 

427 result = self._compression_options.get(name, FitsCompressionOptions.DEFAULT) 

428 if result is None: 428 ↛ 429line 428 didn't jump to line 429 because the condition on line 428 was never true

429 return result 

430 if tile_shape is not None and result.tile_shape is None: 430 ↛ 431line 430 didn't jump to line 431 because the condition on line 430 was never true

431 result = result.model_copy(update={"tile_shape": tile_shape}) 

432 if result.quantization is None: 

433 return result 

434 if self._compression_seed is not None and not result.quantization.seed: 434 ↛ 445line 434 didn't jump to line 445 because the condition on line 434 was always true

435 result = result.model_copy( 

436 update={ 

437 "quantization": result.quantization.model_copy(update={"seed": self._compression_seed}) 

438 } 

439 ) 

440 self._compression_seed += 1 

441 if self._compression_seed > 10000: 441 ↛ 442line 441 didn't jump to line 442 because the condition on line 441 was never true

442 self._compression_seed = 1 

443 # MyPy can tell that result.quantization is not None in the 'if', but 

444 # forgets that by this 'else': 

445 elif result.quantization.seed is None: # type: ignore[union-attr] 

446 raise RuntimeError("No quantization seed provided.") 

447 return result 

448 

449 @staticmethod 

450 def _make_index_table(hdu_list: astropy.io.fits.HDUList) -> astropy.io.fits.BinTableHDU: 

451 # We use a fixed-length string for the EXTNAME column; it might be 

452 # better to use a variable-length array, but I have not been able to 

453 # figure out how to get astropy to accept a string for the the 

454 # character (TFORM='A') variant of that. And that's only better if the 

455 # EXTNAMEs get super long, which is not likely (but maybe something to 

456 # guard against). 

457 max_name_size = max(len(hdu.header.get("EXTNAME", "")) for hdu in hdu_list) 

458 # Attach the ZIMAGE values as a boolean array on the Column itself. 

459 # Astropy only fills a logical column's raw bytes with 'F' as the 

460 # default (rather than NULL, 0x00) when the bool array is present at 

461 # construction; assigning False element-wise after construction leaves 

462 # the byte as NULL under Astropy 8.0.0, which warns on every read. 

463 # Should be fixed in v8.0.1. 

464 # See https://github.com/astropy/astropy/pull/19939 

465 # but pre-allocating the entire column works in v7 and v8. 

466 zimage = np.array([hdu.header.get("ZIMAGE", False) for hdu in hdu_list], dtype=bool) 

467 index_hdu = astropy.io.fits.BinTableHDU.from_columns( 

468 [ 

469 astropy.io.fits.Column("EXTNAME", f"A{max_name_size}"), 

470 astropy.io.fits.Column("EXTVER", "J"), 

471 astropy.io.fits.Column("XTENSION", "A8"), 

472 astropy.io.fits.Column("ZIMAGE", "L", array=zimage), 

473 ] 

474 + _HDUBytes.get_index_hdu_columns(), 

475 nrows=len(hdu_list), 

476 name="INDEX", 

477 ) 

478 hdu: ExtensionHDU | astropy.io.fits.PrimaryHDU 

479 for n, hdu in enumerate(hdu_list): 

480 index_hdu.data[n]["EXTNAME"] = hdu.header.get("EXTNAME", "") 

481 index_hdu.data[n]["EXTVER"] = hdu.header.get("EXTVER", 1) 

482 index_hdu.data[n]["XTENSION"] = hdu.header.get("XTENSION", "IMAGE") 

483 bytes = _HDUBytes.from_read_hdu(hdu) 

484 bytes.update_index_row(index_hdu.data[n]) 

485 _set_creation_date(index_hdu.header) 

486 return index_hdu 

487 

488 

489@dataclasses.dataclass 

490class _HDUBytes: 

491 """A struct that records the byte offsets into a FITS HDU.""" 

492 

493 @classmethod 

494 def from_write_hdu(cls, hdu: astropy.io.fits.PrimaryHDU | ExtensionHDU) -> Self: 

495 """Construct from an Astropy HDU instance that has just been written. 

496 

497 Parameters 

498 ---------- 

499 hdu 

500 An Astropy HDU object. 

501 

502 Returns 

503 ------- 

504 hdu_bytes 

505 Struct with byte offsets. 

506 

507 Notes 

508 ----- 

509 This method relies on internal Astropy attributes and does not work on 

510 CompImageHDU objects. 

511 """ 

512 # This is implemented by accessing private Astropy attributes because 

513 # it turns out that's much more reliable than the public fileinfo() 

514 # method, which seems to always return a dict with `None` entries or 

515 # raise; it looks buggy, but docs are scarce enough that it's not clear 

516 # what the right behavior is supposed to be. 

517 if (header_address := getattr(hdu, "_header_offset", None)) is None: 517 ↛ 518line 517 didn't jump to line 518 because the condition on line 517 was never true

518 raise RuntimeError("Failed to get Astropy's _header_offset.") 

519 if (data_address := getattr(hdu, "_data_offset", None)) is None: 519 ↛ 520line 519 didn't jump to line 520 because the condition on line 519 was never true

520 raise RuntimeError("Failed to get Astropy's _data_offset.") 

521 if (data_size := getattr(hdu, "_data_size", None)) is None: 521 ↛ 522line 521 didn't jump to line 522 because the condition on line 521 was never true

522 raise RuntimeError("Failed to get Astropy's _data_size.") 

523 return cls(header_address, data_address, data_size) 

524 

525 @classmethod 

526 def from_read_hdu(cls, hdu: astropy.io.fits.PrimaryHDU | ExtensionHDU) -> Self: 

527 """Construct from an Astropy HDU instance that has just been read. 

528 

529 Parameters 

530 ---------- 

531 hdu 

532 An Astropy HDU object. 

533 

534 Returns 

535 ------- 

536 hdu_bytes 

537 Struct with byte offsets. 

538 """ 

539 info = hdu.fileinfo() 

540 header_address = info["hdrLoc"] 

541 data_address = info["datLoc"] 

542 data_size = info["datSpan"] 

543 return cls(header_address, data_address, data_size) 

544 

545 @classmethod 

546 def from_index_row(cls, row: np.void) -> Self: 

547 """Construct from a row of the index HDU. 

548 

549 Parameters 

550 ---------- 

551 row 

552 A Numpy struct-like scalar. 

553 

554 Returns 

555 ------- 

556 hdu_bytes 

557 Struct with byte offsets. 

558 """ 

559 return cls( 

560 header_address=int(row["HDRADDR"]), 

561 data_address=int(row["DATADDR"]), 

562 data_size=int(row["DATSIZE"]), 

563 ) 

564 

565 @staticmethod 

566 def get_index_hdu_columns() -> list[astropy.io.fits.Column]: 

567 """Return the definitions of the columns this class gets and sets 

568 from the index HDU. 

569 

570 Returns 

571 ------- 

572 columns 

573 A `list` of `astropy.io.fits.Column` objects that represent the 

574 header address, data address, and data size. 

575 """ 

576 return [ 

577 astropy.io.fits.Column("HDRADDR", "K"), 

578 astropy.io.fits.Column("DATADDR", "K"), 

579 astropy.io.fits.Column("DATSIZE", "K"), 

580 ] 

581 

582 header_address: int 

583 """Offset from the beginning of the start of the file to the header of this 

584 HDU, in bytes. 

585 """ 

586 

587 data_address: int 

588 """Offset from the beginning of the start of the file to the data section 

589 of this HDU, in bytes. 

590 """ 

591 

592 data_size: int 

593 """Size of the data section in bytes.""" 

594 

595 @property 

596 def header_size(self) -> int: 

597 """Size of the header in bytes.""" 

598 return self.data_address - self.header_address 

599 

600 @property 

601 def end_address(self) -> int: 

602 """Offset in bytes from the start of the file to the end of the HDU.""" 

603 return self.data_address + self.data_size 

604 

605 @property 

606 def size(self) -> int: 

607 """Total size of this HDU in bytes.""" 

608 return self.data_size + self.data_address - self.header_address 

609 

610 def update_index_row(self, row: np.void) -> None: 

611 """Set the values of a row of the index HDU from this strut. 

612 

613 Parameters 

614 ---------- 

615 row 

616 A Numpy struct-like scalar to modify in place. 

617 """ 

618 row["HDRADDR"] = self.header_address 

619 row["DATADDR"] = self.data_address 

620 row["DATSIZE"] = self.data_size