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

228 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-06-27 08:32 +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 typing import Any, Self 

20 

21import astropy.io.fits 

22import astropy.table 

23import astropy.time 

24import numpy as np 

25import pydantic 

26 

27from .._transforms import FrameSet 

28from ..serialization import ( 

29 ArchiveTree, 

30 ArrayReferenceModel, 

31 ButlerInfo, 

32 MetadataValue, 

33 NestedOutputArchive, 

34 NumberType, 

35 OutputArchive, 

36 TableColumnModel, 

37 TableModel, 

38 no_header_updates, 

39) 

40from ._common import ( 

41 JSON_COLUMN, 

42 JSON_EXTNAME, 

43 ExtensionHDU, 

44 ExtensionKey, 

45 FitsCompressionOptions, 

46 FitsOpaqueMetadata, 

47 PointerModel, 

48 suppress_fits_card_warnings, 

49) 

50 

51_FITS_FORMAT_VERSION = 1 

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

53 

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

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

56""" 

57 

58 

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

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

61 

62 Parameters 

63 ---------- 

64 header 

65 Header to modify in place. 

66 

67 Notes 

68 ----- 

69 The FITS standard requires every HDU to record the UTC date and time its 

70 header was created via a ``DATE`` card in ``YYYY-MM-DDThh:mm:ss[.sss]`` 

71 form. Any pre-existing ``DATE`` card (such as one preserved from an input 

72 archive) is overwritten in place so the value reflects this write. 

73 """ 

74 header.set("DATE", astropy.time.Time.now().fits, "UTC date this HDU was written.") 

75 

76 

77def write( 

78 obj: Any, 

79 path: str, 

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

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

82 compression_seed: int | None = None, 

83 metadata: dict[str, MetadataValue] | None = None, 

84 butler_info: ButlerInfo | None = None, 

85) -> Any: 

86 """Write an object with a ``serialize`` method to a FITS file. 

87 

88 Parameters 

89 ---------- 

90 obj 

91 Object with a ``serialize`` method to write. 

92 path 

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

94 compression_options 

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

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

97 attributes). 

98 update_header 

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

100 opportunity to modify it. 

101 compression_seed 

102 A FITS tile compression seed to use whenever the configured 

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

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

105 metadata 

106 Additional metadata to save with the object. This will override any 

107 flexible metadata carried by the object itself with the same keys. 

108 butler_info 

109 Butler information to store in the file. 

110 

111 Returns 

112 ------- 

113 `.serialization.ArchiveTree` 

114 The serialized representation of the object. 

115 """ 

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

117 name = getattr(obj, "_archive_default_name", 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_direct(name, obj.serialize) if name is not None else obj.serialize(archive) 

126 if metadata is not None: 126 ↛ 127line 126 didn't jump to line 127 because the condition on line 126 was never true

127 tree.metadata.update(metadata) 

128 if butler_info is not None: 

129 tree.butler_info = butler_info 

130 archive.add_tree(tree) 

131 return tree 

132 

133 

134class FitsOutputArchive(OutputArchive[PointerModel]): 

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

136 writes to FITS files. 

137 

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

139 context manager. 

140 

141 Parameters 

142 ---------- 

143 hdu_list 

144 HDU list that the archive writes its HDUs into. 

145 compression_options 

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

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

148 attributes). 

149 opaque_metadata 

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

151 compression_seed 

152 A FITS tile compression seed to use whenever the configured 

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

154 ``0``. 

155 """ 

156 

157 def __init__( 

158 self, 

159 hdu_list: astropy.io.fits.HDUList, 

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

161 opaque_metadata: Any = None, 

162 compression_seed: int | None = None, 

163 ) -> None: 

164 super().__init__() 

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

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

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

168 # pointer to where we actually saved it: 

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

170 self._hdu_list = hdu_list 

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

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

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

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

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

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

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

178 self._compression_seed = compression_seed 

179 self._opaque_metadata = ( 

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

181 ) 

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

183 self._primary_hdu.header.extend(opaque_primary_header) 

184 self._hdu_list.append(self._primary_hdu) 

185 self._json_hdu_added: bool = False 

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

187 

188 @classmethod 

189 @contextmanager 

190 def open( 

191 cls, 

192 filename: str, 

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

194 opaque_metadata: Any = None, 

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

196 compression_seed: int | None = None, 

197 ) -> Iterator[Self]: 

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

199 

200 Parameters 

201 ---------- 

202 filename 

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

204 compression_options 

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

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

207 attributes). 

208 opaque_metadata 

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

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

211 update_header 

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

213 opportunity to modify it. 

214 compression_seed 

215 A FITS tile compression seed to use whenever the configured 

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

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

218 

219 Returns 

220 ------- 

221 `contextlib.AbstractContextManager` [`FitsOutputArchive`] 

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

223 """ 

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

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

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

227 # across the write of the main HDUs. 

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

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

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

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

232 update_header(hdu_list[0].header) 

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

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

235 # always record the time of this write. 

236 _set_creation_date(hdu_list[0].header) 

237 yield archive 

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

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

240 hdu_list.flush() 

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

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

243 # make this unnecessary at some point. 

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

245 index_hdu = cls._make_index_table(hdu_list) 

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

247 hdu_list.append(index_hdu) 

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

249 index_bytes = _HDUBytes.from_write_hdu(index_hdu) 

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

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

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

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

254 # the primary header. 

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

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

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

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

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

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

261 

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

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

264 ) -> T: 

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

266 return serializer(nested) 

267 

268 def serialize_pointer[T: ArchiveTree]( 

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

270 ) -> PointerModel: 

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

272 return pointer 

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

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

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

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

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

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

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

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

281 self._pointer_targets.append(json_bytes) 

282 pointer = PointerModel( 

283 column=TableColumnModel( 

284 name=JSON_COLUMN, 

285 data=ArrayReferenceModel( 

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

287 shape=[len(json_bytes)], 

288 datatype=NumberType.uint8, 

289 ), 

290 ), 

291 row=len(self._pointer_targets), 

292 ) 

293 self._pointers_by_key[key] = pointer 

294 return pointer 

295 

296 def serialize_frame_set[T: ArchiveTree]( 

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

298 ) -> PointerModel: 

299 # Docstring inherited. 

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

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

302 return pointer 

303 

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

305 return iter(self._frame_sets) 

306 

307 def add_array( 

308 self, 

309 array: np.ndarray, 

310 *, 

311 name: str | None = None, 

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

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

314 options_name: str | None = None, 

315 ) -> ArrayReferenceModel: 

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

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

318 name, version = self._register_name(name) 

319 extname = name.upper() 

320 hdu = self._opaque_metadata.maybe_use_precompressed(extname) 

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

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

323 options_name = name 

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

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

326 else: 

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

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

329 return ArrayReferenceModel( 

330 source=str(key), 

331 shape=list(array.shape), 

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

333 ) 

334 

335 def add_table( 

336 self, 

337 table: astropy.table.Table, 

338 *, 

339 name: str | None = None, 

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

341 ) -> TableModel: 

342 if name is None: 

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

344 name, version = self._register_name(name) 

345 extname = name.upper() 

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

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

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

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

350 columns = TableColumnModel.from_table(table) 

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

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

353 assert isinstance(c.data, ArrayReferenceModel) 

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

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

356 

357 def add_structured_array( 

358 self, 

359 array: np.ndarray, 

360 *, 

361 name: str | None = None, 

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

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

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

365 ) -> TableModel: 

366 if name is None: 

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

368 name, version = self._register_name(name) 

369 extname = name.upper() 

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

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

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

373 columns = TableColumnModel.from_record_dtype(array.dtype) 

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

375 if units is not None: 

376 for c in columns: 

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

378 if descriptions is not None: 

379 for c in columns: 

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

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

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

383 assert isinstance(c.data, ArrayReferenceModel) 

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

385 return TableModel(columns=columns) 

386 

387 def _add_hdu( 

388 self, 

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

390 version: int, 

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

392 ) -> ExtensionKey: 

393 key = ExtensionKey(hdu.name, version) 

394 key.check() 

395 if version > 1: 

396 hdu.header["EXTVER"] = version 

397 update_header(hdu.header) 

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

399 hdu.header.extend(opaque_headers) 

400 _set_creation_date(hdu.header) 

401 self._hdu_list.append(hdu) 

402 return key 

403 

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

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

406 

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

408 is exited. 

409 

410 Parameters 

411 ---------- 

412 tree 

413 Pydantic model that represents the tree. 

414 """ 

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

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

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

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

419 name=JSON_EXTNAME, 

420 ) 

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

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

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

424 _set_creation_date(json_hdu.header) 

425 self._hdu_list.append(json_hdu) 

426 self._json_hdu_added = True 

427 

428 def _get_compression_options( 

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

430 ) -> FitsCompressionOptions | None: 

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

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

433 return result 

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

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

436 if result.quantization is None: 

437 return result 

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

439 result = result.model_copy( 

440 update={ 

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

442 } 

443 ) 

444 self._compression_seed += 1 

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

446 self._compression_seed = 1 

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

448 # forgets that by this 'else': 

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

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

451 return result 

452 

453 @staticmethod 

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

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

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

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

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

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

460 # guard against). 

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

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

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

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

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

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

467 # Should be fixed in v8.0.1. 

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

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

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

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

472 [ 

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

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

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

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

477 ] 

478 + _HDUBytes.get_index_hdu_columns(), 

479 nrows=len(hdu_list), 

480 name="INDEX", 

481 ) 

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

483 for n, hdu in enumerate(hdu_list): 

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

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

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

487 bytes = _HDUBytes.from_read_hdu(hdu) 

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

489 _set_creation_date(index_hdu.header) 

490 return index_hdu 

491 

492 

493@dataclasses.dataclass 

494class _HDUBytes: 

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

496 

497 @classmethod 

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

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

500 

501 Parameters 

502 ---------- 

503 hdu 

504 An Astropy HDU object. 

505 

506 Returns 

507 ------- 

508 hdu_bytes 

509 Struct with byte offsets. 

510 

511 Notes 

512 ----- 

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

514 CompImageHDU objects. 

515 """ 

516 # This is implemented by accessing private Astropy attributes because 

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

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

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

520 # what the right behavior is supposed to be. 

521 if (header_address := getattr(hdu, "_header_offset", 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 _header_offset.") 

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

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

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

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

527 return cls(header_address, data_address, data_size) 

528 

529 @classmethod 

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

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

532 

533 Parameters 

534 ---------- 

535 hdu 

536 An Astropy HDU object. 

537 

538 Returns 

539 ------- 

540 hdu_bytes 

541 Struct with byte offsets. 

542 """ 

543 info = hdu.fileinfo() 

544 header_address = info["hdrLoc"] 

545 data_address = info["datLoc"] 

546 data_size = info["datSpan"] 

547 return cls(header_address, data_address, data_size) 

548 

549 @classmethod 

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

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

552 

553 Parameters 

554 ---------- 

555 row 

556 A Numpy struct-like scalar. 

557 

558 Returns 

559 ------- 

560 hdu_bytes 

561 Struct with byte offsets. 

562 """ 

563 return cls( 

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

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

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

567 ) 

568 

569 @staticmethod 

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

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

572 from the index HDU. 

573 

574 Returns 

575 ------- 

576 columns 

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

578 header address, data address, and data size. 

579 """ 

580 return [ 

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

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

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

584 ] 

585 

586 header_address: int 

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

588 HDU, in bytes. 

589 """ 

590 

591 data_address: int 

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

593 of this HDU, in bytes. 

594 """ 

595 

596 data_size: int 

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

598 

599 @property 

600 def header_size(self) -> int: 

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

602 return self.data_address - self.header_address 

603 

604 @property 

605 def end_address(self) -> int: 

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

607 return self.data_address + self.data_size 

608 

609 @property 

610 def size(self) -> int: 

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

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

613 

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

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

616 

617 Parameters 

618 ---------- 

619 row 

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

621 """ 

622 row["HDRADDR"] = self.header_address 

623 row["DATADDR"] = self.data_address 

624 row["DATSIZE"] = self.data_size