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

228 statements  

« prev     ^ index     » next       coverage.py v7.14.3, created at 2026-06-25 08:24 +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 path 

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

92 compression_options 

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

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

95 attributes). 

96 update_header 

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

98 opportunity to modify it. 

99 compression_seed 

100 A FITS tile compression seed to use whenever the configured 

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

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

103 metadata 

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

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

106 butler_info 

107 Butler information to store in the file. 

108 

109 Returns 

110 ------- 

111 `.serialization.ArchiveTree` 

112 The serialized representation of the object. 

113 """ 

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

115 name = getattr(obj, "_archive_default_name", None) 

116 with FitsOutputArchive.open( 

117 path, 

118 compression_options=compression_options, 

119 opaque_metadata=opaque_metadata, 

120 update_header=update_header, 

121 compression_seed=compression_seed, 

122 ) as archive: 

123 tree = archive.serialize_direct(name, obj.serialize) if name is not None else obj.serialize(archive) 

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

125 tree.metadata.update(metadata) 

126 if butler_info is not None: 

127 tree.butler_info = butler_info 

128 archive.add_tree(tree) 

129 return tree 

130 

131 

132class FitsOutputArchive(OutputArchive[PointerModel]): 

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

134 writes to FITS files. 

135 

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

137 context manager. 

138 """ 

139 

140 def __init__( 

141 self, 

142 hdu_list: astropy.io.fits.HDUList, 

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

144 opaque_metadata: Any = None, 

145 compression_seed: int | None = None, 

146 ) -> None: 

147 super().__init__() 

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

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

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

151 # pointer to where we actually saved it: 

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

153 self._hdu_list = hdu_list 

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

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

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

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

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

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

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

161 self._compression_seed = compression_seed 

162 self._opaque_metadata = ( 

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

164 ) 

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

166 self._primary_hdu.header.extend(opaque_primary_header) 

167 self._hdu_list.append(self._primary_hdu) 

168 self._json_hdu_added: bool = False 

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

170 

171 @classmethod 

172 @contextmanager 

173 def open( 

174 cls, 

175 filename: str, 

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

177 opaque_metadata: Any = None, 

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

179 compression_seed: int | None = None, 

180 ) -> Iterator[Self]: 

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

182 

183 Parameters 

184 ---------- 

185 filename 

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

187 compression_options 

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

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

190 attributes). 

191 opaque_metadata 

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

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

194 update_header 

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

196 opportunity to modify it. 

197 compression_seed 

198 A FITS tile compression seed to use whenever the configured 

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

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

201 

202 Returns 

203 ------- 

204 `contextlib.AbstractContextManager` [`FitsOutputArchive`] 

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

206 """ 

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

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

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

210 # across the write of the main HDUs. 

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

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

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

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

215 update_header(hdu_list[0].header) 

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

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

218 # always record the time of this write. 

219 _set_creation_date(hdu_list[0].header) 

220 yield archive 

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

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

223 hdu_list.flush() 

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

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

226 # make this unnecessary at some point. 

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

228 index_hdu = cls._make_index_table(hdu_list) 

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

230 hdu_list.append(index_hdu) 

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

232 index_bytes = _HDUBytes.from_write_hdu(index_hdu) 

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

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

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

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

237 # the primary header. 

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

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

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

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

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

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

244 

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

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

247 ) -> T: 

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

249 return serializer(nested) 

250 

251 def serialize_pointer[T: ArchiveTree]( 

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

253 ) -> PointerModel: 

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

255 return pointer 

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

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

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

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

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

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

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

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

264 self._pointer_targets.append(json_bytes) 

265 pointer = PointerModel( 

266 column=TableColumnModel( 

267 name=JSON_COLUMN, 

268 data=ArrayReferenceModel( 

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

270 shape=[len(json_bytes)], 

271 datatype=NumberType.uint8, 

272 ), 

273 ), 

274 row=len(self._pointer_targets), 

275 ) 

276 self._pointers_by_key[key] = pointer 

277 return pointer 

278 

279 def serialize_frame_set[T: ArchiveTree]( 

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

281 ) -> PointerModel: 

282 # Docstring inherited. 

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

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

285 return pointer 

286 

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

288 return iter(self._frame_sets) 

289 

290 def add_array( 

291 self, 

292 array: np.ndarray, 

293 *, 

294 name: str | None = None, 

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

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

297 options_name: str | None = None, 

298 ) -> ArrayReferenceModel: 

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

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

301 name, version = self._register_name(name) 

302 extname = name.upper() 

303 hdu = self._opaque_metadata.maybe_use_precompressed(extname) 

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

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

306 options_name = name 

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

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

309 else: 

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

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

312 return ArrayReferenceModel( 

313 source=str(key), 

314 shape=list(array.shape), 

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

316 ) 

317 

318 def add_table( 

319 self, 

320 table: astropy.table.Table, 

321 *, 

322 name: str | None = None, 

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

324 ) -> TableModel: 

325 if name is None: 

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

327 name, version = self._register_name(name) 

328 extname = name.upper() 

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

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

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

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

333 columns = TableColumnModel.from_table(table) 

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

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

336 assert isinstance(c.data, ArrayReferenceModel) 

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

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

339 

340 def add_structured_array( 

341 self, 

342 array: np.ndarray, 

343 *, 

344 name: str | None = None, 

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

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

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

348 ) -> TableModel: 

349 if name is None: 

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

351 name, version = self._register_name(name) 

352 extname = name.upper() 

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

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

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

356 columns = TableColumnModel.from_record_dtype(array.dtype) 

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

358 if units is not None: 

359 for c in columns: 

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

361 if descriptions is not None: 

362 for c in columns: 

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

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

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

366 assert isinstance(c.data, ArrayReferenceModel) 

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

368 return TableModel(columns=columns) 

369 

370 def _add_hdu( 

371 self, 

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

373 version: int, 

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

375 ) -> ExtensionKey: 

376 key = ExtensionKey(hdu.name, version) 

377 key.check() 

378 if version > 1: 

379 hdu.header["EXTVER"] = version 

380 update_header(hdu.header) 

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

382 hdu.header.extend(opaque_headers) 

383 _set_creation_date(hdu.header) 

384 self._hdu_list.append(hdu) 

385 return key 

386 

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

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

389 

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

391 is exited. 

392 

393 Parameters 

394 ---------- 

395 tree 

396 Pydantic model that represents the tree. 

397 """ 

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

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

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

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

402 name=JSON_EXTNAME, 

403 ) 

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

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

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

407 _set_creation_date(json_hdu.header) 

408 self._hdu_list.append(json_hdu) 

409 self._json_hdu_added = True 

410 

411 def _get_compression_options( 

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

413 ) -> FitsCompressionOptions | None: 

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

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

416 return result 

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

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

419 if result.quantization is None: 

420 return result 

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

422 result = result.model_copy( 

423 update={ 

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

425 } 

426 ) 

427 self._compression_seed += 1 

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

429 self._compression_seed = 1 

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

431 # forgets that by this 'else': 

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

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

434 return result 

435 

436 @staticmethod 

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

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

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

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

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

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

443 # guard against). 

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

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

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

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

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

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

450 # Should be fixed in v8.0.1. 

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

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

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

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

455 [ 

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

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

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

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

460 ] 

461 + _HDUBytes.get_index_hdu_columns(), 

462 nrows=len(hdu_list), 

463 name="INDEX", 

464 ) 

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

466 for n, hdu in enumerate(hdu_list): 

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

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

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

470 bytes = _HDUBytes.from_read_hdu(hdu) 

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

472 _set_creation_date(index_hdu.header) 

473 return index_hdu 

474 

475 

476@dataclasses.dataclass 

477class _HDUBytes: 

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

479 

480 @classmethod 

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

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

483 

484 Parameters 

485 ---------- 

486 hdu 

487 An Astropy HDU object. 

488 

489 Returns 

490 ------- 

491 hdu_bytes 

492 Struct with byte offsets. 

493 

494 Notes 

495 ----- 

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

497 CompImageHDU objects. 

498 """ 

499 # This is implemented by accessing private Astropy attributes because 

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

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

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

503 # what the right behavior is supposed to be. 

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

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

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

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

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

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

510 return cls(header_address, data_address, data_size) 

511 

512 @classmethod 

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

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

515 

516 Parameters 

517 ---------- 

518 hdu 

519 An Astropy HDU object. 

520 

521 Returns 

522 ------- 

523 hdu_bytes 

524 Struct with byte offsets. 

525 """ 

526 info = hdu.fileinfo() 

527 header_address = info["hdrLoc"] 

528 data_address = info["datLoc"] 

529 data_size = info["datSpan"] 

530 return cls(header_address, data_address, data_size) 

531 

532 @classmethod 

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

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

535 

536 Parameters 

537 ---------- 

538 row 

539 A Numpy struct-like scalar. 

540 

541 Returns 

542 ------- 

543 hdu_bytes 

544 Struct with byte offsets. 

545 """ 

546 return cls( 

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

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

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

550 ) 

551 

552 @staticmethod 

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

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

555 from the index HDU. 

556 

557 Returns 

558 ------- 

559 columns 

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

561 header address, data address, and data size. 

562 """ 

563 return [ 

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

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

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

567 ] 

568 

569 header_address: int 

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

571 HDU, in bytes. 

572 """ 

573 

574 data_address: int 

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

576 of this HDU, in bytes. 

577 """ 

578 

579 data_size: int 

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

581 

582 @property 

583 def header_size(self) -> int: 

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

585 return self.data_address - self.header_address 

586 

587 @property 

588 def end_address(self) -> int: 

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

590 return self.data_address + self.data_size 

591 

592 @property 

593 def size(self) -> int: 

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

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

596 

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

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

599 

600 Parameters 

601 ---------- 

602 row 

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

604 """ 

605 row["HDRADDR"] = self.header_address 

606 row["DATADDR"] = self.data_address 

607 row["DATSIZE"] = self.data_size