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

224 statements  

« prev     ^ index     » next       coverage.py v7.14.1, created at 2026-06-10 02:26 -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. 

11 

12from __future__ import annotations 

13 

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

15 

16import dataclasses 

17import warnings 

18from collections import Counter 

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

20from contextlib import contextmanager 

21from typing import Any, Self 

22 

23import astropy.io.fits 

24import astropy.table 

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) 

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 write( 

60 obj: Any, 

61 path: str, 

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

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

64 compression_seed: int | None = None, 

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

66 butler_info: ButlerInfo | None = None, 

67) -> Any: 

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

69 

70 Parameters 

71 ---------- 

72 path 

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

74 compression_options 

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

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

77 attributes). 

78 update_header 

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

80 opportunity to modify it. 

81 compression_seed 

82 A FITS tile compression seed to use whenever the configured 

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

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

85 metadata 

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

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

88 butler_info 

89 Butler information to store in the file. 

90 

91 Returns 

92 ------- 

93 `.serialization.ArchiveTree` 

94 The serialized representation of the object. 

95 """ 

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

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

98 with FitsOutputArchive.open( 

99 path, 

100 compression_options=compression_options, 

101 opaque_metadata=opaque_metadata, 

102 update_header=update_header, 

103 compression_seed=compression_seed, 

104 ) as archive: 

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

106 if metadata is not None: 

107 tree.metadata.update(metadata) 

108 if butler_info is not None: 

109 tree.butler_info = butler_info 

110 archive.add_tree(tree) 

111 return tree 

112 

113 

114class FitsOutputArchive(OutputArchive[PointerModel]): 

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

116 writes to FITS files. 

117 

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

119 context manager. 

120 """ 

121 

122 def __init__( 

123 self, 

124 hdu_list: astropy.io.fits.HDUList, 

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

126 opaque_metadata: Any = None, 

127 compression_seed: int | None = None, 

128 ): 

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

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

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

132 # pointer to where we actually saved it: 

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

134 self._hdu_list = hdu_list 

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

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

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

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

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

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

141 self._hdus_by_name = Counter[str]() 

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

143 self._compression_seed = compression_seed 

144 self._opaque_metadata = ( 

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

146 ) 

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

148 self._primary_hdu.header.extend(opaque_primary_header) 

149 self._hdu_list.append(self._primary_hdu) 

150 self._json_hdu_added: bool = False 

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

152 

153 @classmethod 

154 @contextmanager 

155 def open( 

156 cls, 

157 filename: str, 

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

159 opaque_metadata: Any = None, 

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

161 compression_seed: int | None = None, 

162 ) -> Iterator[Self]: 

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

164 

165 Parameters 

166 ---------- 

167 filename 

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

169 compression_options 

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

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

172 attributes). 

173 opaque_metadata 

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

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

176 update_header 

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

178 opportunity to modify it. 

179 compression_seed 

180 A FITS tile compression seed to use whenever the configured 

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

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

183 

184 Returns 

185 ------- 

186 `contextlib.AbstractContextManager` [`FitsOutputArchive`] 

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

188 """ 

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

190 if hdu_list: 

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

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

193 update_header(hdu_list[0].header) 

194 yield archive 

195 if not archive._json_hdu_added: 

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

197 # If a header card has a long string that required CONTINUE, 

198 # Astropy will truncate the comment and warn without reporting 

199 # what the offending card is. But it doesn't look at its own 

200 # output_verify kwarg when doing that, and it doesn't actually 

201 # trigger if you try to format the header cards one at a time! 

202 # So we have no choice but to silence the warnings manually, until 

203 # we can get Astropy fixed. 

204 with warnings.catch_warnings(): 

205 warnings.filterwarnings( 

206 "ignore", 

207 message="comment will be truncated", 

208 category=astropy.io.fits.verify.VerifyWarning, 

209 ) 

210 hdu_list.flush() 

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

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

213 # make this unnecessary at some point. 

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

215 index_hdu = cls._make_index_table(hdu_list) 

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

217 hdu_list.append(index_hdu) 

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

219 index_bytes = _HDUBytes.from_write_hdu(index_hdu) 

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

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

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

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

224 # the primary header. 

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

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

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

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

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

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

231 

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

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

234 ) -> T: 

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

236 return serializer(nested) 

237 

238 def serialize_pointer[T: ArchiveTree]( 

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

240 ) -> PointerModel: 

241 if (pointer := self._pointers_by_key.get(key)) is not None: 

242 return pointer 

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

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

245 self._pointer_targets.append(json_bytes) 

246 pointer = PointerModel( 

247 column=TableColumnModel( 

248 name=JSON_COLUMN, 

249 data=ArrayReferenceModel( 

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

251 shape=[len(json_bytes)], 

252 datatype=NumberType.uint8, 

253 ), 

254 ), 

255 row=len(self._pointer_targets), 

256 ) 

257 self._pointers_by_key[key] = pointer 

258 return pointer 

259 

260 def serialize_frame_set[T: ArchiveTree]( 

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

262 ) -> PointerModel: 

263 # Docstring inherited. 

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

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

266 return pointer 

267 

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

269 return iter(self._frame_sets) 

270 

271 def add_array( 

272 self, 

273 array: np.ndarray, 

274 *, 

275 name: str | None = None, 

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

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

278 options_name: str | None = None, 

279 ) -> ArrayReferenceModel: 

280 if name is None: 

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

282 extname = name.upper() 

283 hdu = self._opaque_metadata.maybe_use_precompressed(extname) 

284 if hdu is None: 

285 if options_name is None: 

286 options_name = name 

287 if (compression_options := self._get_compression_options(options_name, tile_shape)) is not None: 

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

289 else: 

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

291 key = self._add_hdu(hdu, update_header) 

292 return ArrayReferenceModel( 

293 source=str(key), 

294 shape=list(array.shape), 

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

296 ) 

297 

298 def add_table( 

299 self, 

300 table: astropy.table.Table, 

301 *, 

302 name: str | None = None, 

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

304 ) -> TableModel: 

305 if name is None: 

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

307 extname = name.upper() 

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

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

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

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

312 columns = TableColumnModel.from_table(table) 

313 key = self._add_hdu(hdu, update_header) 

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

315 assert isinstance(c.data, ArrayReferenceModel) 

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

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

318 

319 def add_structured_array( 

320 self, 

321 array: np.ndarray, 

322 *, 

323 name: str | None = None, 

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

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

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

327 ) -> TableModel: 

328 if name is None: 

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

330 extname = name.upper() 

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

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

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

334 columns = TableColumnModel.from_record_dtype(array.dtype) 

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

336 if units is not None: 

337 for c in columns: 

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

339 if descriptions is not None: 

340 for c in columns: 

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

342 key = self._add_hdu(hdu, update_header) 

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

344 assert isinstance(c.data, ArrayReferenceModel) 

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

346 return TableModel(columns=columns) 

347 

348 def _add_hdu( 

349 self, 

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

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

352 ) -> ExtensionKey: 

353 n_hdus = self._hdus_by_name.get(hdu.name, 0) 

354 key = ExtensionKey(hdu.name, n_hdus + 1) 

355 key.check() 

356 if n_hdus: 

357 hdu.header["EXTVER"] = key.ver 

358 self._hdus_by_name[hdu.name] += 1 

359 update_header(hdu.header) 

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

361 hdu.header.extend(opaque_headers) 

362 self._hdu_list.append(hdu) 

363 return key 

364 

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

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

367 

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

369 is exited. 

370 

371 Parameters 

372 ---------- 

373 tree 

374 Pydantic model that represents the tree. 

375 """ 

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

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

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

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

380 name=JSON_EXTNAME, 

381 ) 

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

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

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

385 self._hdu_list.append(json_hdu) 

386 self._json_hdu_added = True 

387 

388 def _get_compression_options( 

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

390 ) -> FitsCompressionOptions | None: 

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

392 if result is None: 

393 return result 

394 if tile_shape is not None and result.tile_shape is None: 

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

396 if result.quantization is None: 

397 return result 

398 if self._compression_seed is not None and not result.quantization.seed: 

399 result = result.model_copy( 

400 update={ 

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

402 } 

403 ) 

404 self._compression_seed += 1 

405 if self._compression_seed > 10000: 

406 self._compression_seed = 1 

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

408 # forgets that by this 'else': 

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

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

411 return result 

412 

413 @staticmethod 

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

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

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

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

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

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

420 # guard against). 

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

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

423 [ 

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

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

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

427 astropy.io.fits.Column("ZIMAGE", "L"), 

428 ] 

429 + _HDUBytes.get_index_hdu_columns(), 

430 nrows=len(hdu_list), 

431 name="INDEX", 

432 ) 

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

434 for n, hdu in enumerate(hdu_list): 

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

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

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

438 index_hdu.data[n]["ZIMAGE"] = hdu.header.get("ZIMAGE", False) 

439 bytes = _HDUBytes.from_read_hdu(hdu) 

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

441 return index_hdu 

442 

443 

444@dataclasses.dataclass 

445class _HDUBytes: 

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

447 

448 @classmethod 

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

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

451 

452 Parameters 

453 ---------- 

454 hdu 

455 An Astropy HDU object. 

456 

457 Returns 

458 ------- 

459 hdu_bytes 

460 Struct with byte offsets. 

461 

462 Notes 

463 ----- 

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

465 CompImageHDU objects. 

466 """ 

467 # This is implemented by accessing private Astropy attributes because 

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

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

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

471 # what the right behavior is supposed to be. 

472 if (header_address := getattr(hdu, "_header_offset", None)) is None: 

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

474 if (data_address := getattr(hdu, "_data_offset", None)) is None: 

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

476 if (data_size := getattr(hdu, "_data_size", None)) is None: 

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

478 return cls(header_address, data_address, data_size) 

479 

480 @classmethod 

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

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

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 info = hdu.fileinfo() 

495 header_address = info["hdrLoc"] 

496 data_address = info["datLoc"] 

497 data_size = info["datSpan"] 

498 return cls(header_address, data_address, data_size) 

499 

500 @classmethod 

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

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

503 

504 Parameters 

505 ---------- 

506 row 

507 A Numpy struct-like scalar. 

508 

509 Returns 

510 ------- 

511 hdu_bytes 

512 Struct with byte offsets. 

513 """ 

514 return cls( 

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

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

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

518 ) 

519 

520 @staticmethod 

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

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

523 from the index HDU. 

524 

525 Returns 

526 ------- 

527 columns 

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

529 header address, data address, and data size. 

530 """ 

531 return [ 

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

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

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

535 ] 

536 

537 header_address: int 

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

539 HDU, in bytes. 

540 """ 

541 

542 data_address: int 

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

544 of this HDU, in bytes. 

545 """ 

546 

547 data_size: int 

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

549 

550 @property 

551 def header_size(self) -> int: 

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

553 return self.data_address - self.header_address 

554 

555 @property 

556 def end_address(self) -> int: 

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

558 return self.data_address + self.data_size 

559 

560 @property 

561 def size(self) -> int: 

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

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

564 

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

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

567 

568 Parameters 

569 ---------- 

570 row 

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

572 """ 

573 row["HDRADDR"] = self.header_address 

574 row["DATADDR"] = self.data_address 

575 row["DATSIZE"] = self.data_size