Coverage for python/astro_metadata_translator/observationInfo.py: 26%

1# This file is part of astro_metadata_translator. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (http://www.lsst.org). 

6# See the LICENSE 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 

12"""Represent standard metadata from instrument headers.""" 

13 

14from __future__ import annotations 

15 

16__all__ = ("ObservationInfo", "makeObservationInfo") 

17 

18import copy 

19import itertools 

20import logging 

21from collections.abc import MutableMapping, Sequence 

22from typing import Any, cast, overload 

23 

24import astropy.coordinates 

25import astropy.time 

26import astropy.units 

27import numpy as np 

28from lsst.resources import ResourcePath 

29from pydantic import ( 

30 BaseModel, 

31 ConfigDict, 

32 Field, 

33 GetJsonSchemaHandler, 

34 PrivateAttr, 

35 model_serializer, 

36) 

37from pydantic.json_schema import JsonSchemaValue 

38from pydantic_core import CoreSchema 

39 

40from .headers import fix_header 

41from .properties import ( 

42 PROPERTIES, 

43 AltAzAnnotated, 

44 AngleAnnotated, 

45 EarthLocationAnnotated, 

46 ExposureTimeAnnotated, 

47 FocusZAnnotated, 

48 PressureAnnotated, 

49 PropertyDefinition, 

50 SkyCoordAnnotated, 

51 TemperatureAnnotated, 

52 TimeAnnotated, 

53 TimeDeltaAnnotated, 

54) 

55from .translator import MetadataTranslator 

56 

57log = logging.getLogger(__name__) 

58 

59 

60def _wire_doc(key: str, wire_form: str) -> str: 

61 """Append a wire-format note to a property's semantic doc. 

62 

63 Used only for ``Field(description=...)`` in the JSON schema; the 

64 underlying ``PROPERTIES[key].doc`` is intentionally left unit-agnostic 

65 because it is also surfaced as the docstring of auto-generated 

66 translator stub methods, where forcing a wire-format unit would be 

67 misleading. 

68 

69 Parameters 

70 ---------- 

71 key : `str` 

72 Property name in `PROPERTIES`. 

73 wire_form : `str` 

74 Short noun phrase describing the JSON-serialized form, e.g. 

75 ``"a float in meters"``. 

76 

77 Returns 

78 ------- 

79 description : `str` 

80 ``PROPERTIES[key].doc`` followed by ``"Serialized as <wire_form>."``. 

81 """ 

82 return f"{PROPERTIES[key].doc} Serialized as {wire_form}." 

83 

84 

85class ObservationInfo(BaseModel): 

86 """Standardized representation of an instrument header for a single 

87 exposure observation. 

88 

89 Parameters 

90 ---------- 

91 header : `dict`-like 

92 Representation of an instrument header accessible as a `dict`. 

93 May be updated with header corrections if corrections are found. 

94 filename : `str`, optional 

95 Name of the file whose header is being translated. For some 

96 datasets with missing header information this can sometimes 

97 allow for some fixups in translations. 

98 translator_class : `MetadataTranslator`-class, optional 

99 If not `None`, the class to use to translate the supplied headers 

100 into standard form. Otherwise each registered translator class will 

101 be asked in turn if it knows how to translate the supplied header. 

102 pedantic : `bool`, optional 

103 If True the translation must succeed for all properties. If False 

104 individual property translations must all be implemented but can fail 

105 and a warning will be issued. Only used if a ``header`` is specified. 

106 search_path : `~collections.abc.Iterable`, optional 

107 Override search paths to use during header fix up. Only used if a 

108 ``header`` is specified. 

109 required : `set`, optional 

110 This parameter can be used to confirm that all properties contained 

111 in the set must translate correctly and also be non-None. For the case 

112 where ``pedantic`` is `True` this will still check that the resulting 

113 value is not `None`. Only used if a ``header`` is specified. 

114 subset : `set`, optional 

115 If not `None`, controls the translations that will be performed 

116 during construction. This can be useful if the caller is only 

117 interested in a subset of the properties and knows that some of 

118 the others might be slow to compute (for example the airmass if it 

119 has to be derived). Only used if a ``header`` is specified. 

120 quiet : `bool`, optional 

121 If `True`, warning level log messages that would be issued in non 

122 pedantic mode are converted to debug messages. 

123 **kwargs : `typing.Any` 

124 Property name/value pairs for kwargs-based construction mode. This 

125 mode creates an `ObservationInfo` directly from supplied properties 

126 rather than by translating a header. If ``header`` is provided it is 

127 an error to also provide ``kwargs``. 

128 

129 Raises 

130 ------ 

131 ValueError 

132 Raised if the supplied header was not recognized by any of the 

133 registered translators. Also raised if the request property subset 

134 is not a subset of the known properties or if a header is given along 

135 with kwargs. 

136 TypeError 

137 Raised if the supplied translator class was not a MetadataTranslator. 

138 KeyError 

139 Raised if a required property cannot be calculated, or if pedantic 

140 mode is enabled and any translations fails. 

141 NotImplementedError 

142 Raised if the selected translator does not support a required 

143 property. 

144 

145 Notes 

146 ----- 

147 There is a core set of instrumental properties that are pre-defined. 

148 Additional properties may be defined, either through the 

149 `makeObservationInfo` factory function by providing the ``extensions`` 

150 definitions, or through the regular `ObservationInfo` constructor when 

151 the extensions have been defined in the `MetadataTranslator` for the 

152 instrument of interest (or in the provided ``translator_class``). 

153 

154 There are two forms of the constructor. If the ``header`` is given 

155 then a translator will be determined and the properties will be populated 

156 accordingly. No generic keyword arguments will be expected and the 

157 remaining parameters control the behavior of the translator. 

158 

159 If the header is not given it is assumed that the keyword arguments 

160 are direct specifications of observation properties. In this mode only 

161 the ``filename`` and ``translator_class`` parameters will be used. The 

162 latter is used to determine any extensions that are being provided, 

163 although when using standard serializations the special ``_translator`` 

164 key will be used instead to specify the name of the registered translator 

165 from which to extract extension definitions. 

166 

167 Headers will be corrected if correction files are located and this will 

168 modify the header provided to the constructor. Modifying the supplied 

169 header after construction will modify the internal cached header. 

170 

171 Values of the properties are read-only. 

172 """ 

173 

174 model_config = ConfigDict( 

175 extra="forbid", 

176 validate_assignment=False, 

177 populate_by_name=True, 

178 serialize_by_alias=True, # Emit the ``_translator`` alias even when nested. 

179 ser_json_inf_nan="constants", # Allow for inf and nan to round trip. 

180 ) 

181 

182 translator_name: str | None = Field( 

183 default=None, 

184 alias="_translator", 

185 description="Name of the registered metadata translator class used for these data.", 

186 ) 

187 

188 telescope: str | None = Field(default=None, description=PROPERTIES["telescope"].doc) 

189 instrument: str | None = Field(default=None, description=PROPERTIES["instrument"].doc) 

190 location: EarthLocationAnnotated | None = Field( 

191 default=None, 

192 description=_wire_doc("location", "a geocentric (x, y, z) tuple of floats in meters"), 

193 ) 

194 exposure_id: int | None = Field(default=None, description=PROPERTIES["exposure_id"].doc) 

195 visit_id: int | None = Field(default=None, description=PROPERTIES["visit_id"].doc) 

196 physical_filter: str | None = Field(default=None, description=PROPERTIES["physical_filter"].doc) 

197 datetime_begin: TimeAnnotated | None = Field( 

198 default=None, 

199 description=_wire_doc("datetime_begin", "a two-element TAI Julian Date [jd1, jd2]"), 

200 ) 

201 datetime_end: TimeAnnotated | None = Field( 

202 default=None, 

203 description=_wire_doc("datetime_end", "a two-element TAI Julian Date [jd1, jd2]"), 

204 ) 

205 exposure_time: ExposureTimeAnnotated | None = Field( 

206 default=None, description=PROPERTIES["exposure_time"].doc 

207 ) 

208 exposure_time_requested: ExposureTimeAnnotated | None = Field( 

209 default=None, description=PROPERTIES["exposure_time_requested"].doc 

210 ) 

211 dark_time: ExposureTimeAnnotated | None = Field(default=None, description=PROPERTIES["dark_time"].doc) 

212 boresight_airmass: float | None = Field(default=None, description=PROPERTIES["boresight_airmass"].doc) 

213 boresight_rotation_angle: AngleAnnotated | None = Field( 

214 default=None, 

215 description=_wire_doc("boresight_rotation_angle", "a float in degrees"), 

216 ) 

217 boresight_rotation_coord: str | None = Field( 

218 default=None, description=PROPERTIES["boresight_rotation_coord"].doc 

219 ) 

220 detector_num: int | None = Field(default=None, description=PROPERTIES["detector_num"].doc) 

221 detector_name: str | None = Field(default=None, description=PROPERTIES["detector_name"].doc) 

222 detector_unique_name: str | None = Field(default=None, description=PROPERTIES["detector_unique_name"].doc) 

223 detector_serial: str | None = Field(default=None, description=PROPERTIES["detector_serial"].doc) 

224 detector_group: str | None = Field(default=None, description=PROPERTIES["detector_group"].doc) 

225 detector_exposure_id: int | None = Field(default=None, description=PROPERTIES["detector_exposure_id"].doc) 

226 focus_z: FocusZAnnotated | None = Field( 

227 default=None, 

228 description=_wire_doc("focus_z", "a float in meters"), 

229 ) 

230 object: str | None = Field(default=None, description=PROPERTIES["object"].doc) 

231 temperature: TemperatureAnnotated | None = Field( 

232 default=None, 

233 description=_wire_doc("temperature", "a float in kelvin"), 

234 ) 

235 pressure: PressureAnnotated | None = Field( 

236 default=None, 

237 description=_wire_doc("pressure", "a float in hPa"), 

238 ) 

239 relative_humidity: float | None = Field(default=None, description=PROPERTIES["relative_humidity"].doc) 

240 tracking_radec: SkyCoordAnnotated | None = Field( 

241 default=None, 

242 description=_wire_doc("tracking_radec", "an ICRS (RA, Dec) tuple of floats in degrees"), 

243 ) 

244 altaz_begin: AltAzAnnotated | None = Field( 

245 default=None, 

246 description=_wire_doc("altaz_begin", "an (azimuth, altitude) tuple of floats in degrees"), 

247 ) 

248 altaz_end: AltAzAnnotated | None = Field( 

249 default=None, 

250 description=_wire_doc("altaz_end", "an (azimuth, altitude) tuple of floats in degrees"), 

251 ) 

252 science_program: str | None = Field(default=None, description=PROPERTIES["science_program"].doc) 

253 observation_type: str | None = Field(default=None, description=PROPERTIES["observation_type"].doc) 

254 observation_id: str | None = Field(default=None, description=PROPERTIES["observation_id"].doc) 

255 observation_reason: str | None = Field(default=None, description=PROPERTIES["observation_reason"].doc) 

256 exposure_group: str | None = Field(default=None, description=PROPERTIES["exposure_group"].doc) 

257 observing_day: int | None = Field(default=None, description=PROPERTIES["observing_day"].doc) 

258 observing_day_offset: TimeDeltaAnnotated | None = Field( 

259 default=None, 

260 description=_wire_doc("observing_day_offset", "integer seconds"), 

261 ) 

262 observation_counter: int | None = Field(default=None, description=PROPERTIES["observation_counter"].doc) 

263 has_simulated_content: bool | None = Field( 

264 default=None, description=PROPERTIES["has_simulated_content"].doc 

265 ) 

266 group_counter_start: int | None = Field(default=None, description=PROPERTIES["group_counter_start"].doc) 

267 group_counter_end: int | None = Field(default=None, description=PROPERTIES["group_counter_end"].doc) 

268 can_see_sky: bool | None = Field(default=None, description=PROPERTIES["can_see_sky"].doc) 

269 

270 # Internal runtime state. These are not part of the wire format and so are 

271 # kept as PrivateAttr to keep them out of the generated JSON Schema. 

272 _filename: str | None = PrivateAttr(default=None) 

273 _translator_class_name: str = PrivateAttr(default="<None>") 

274 _extensions: dict[str, PropertyDefinition] = PrivateAttr(default_factory=dict) 

275 _all_properties: dict[str, PropertyDefinition] = PrivateAttr(default_factory=dict) 

276 _header: MutableMapping[str, Any] = PrivateAttr(default_factory=dict) 

277 _translator: MetadataTranslator | None = PrivateAttr(default=None) 

278 _sealed: bool = PrivateAttr(default=False) 

279 

280 @property 

281 def filename(self) -> str | None: 

282 """Name of the file whose header was translated, if any.""" 

283 return self._filename 

284 

285 @filename.setter 

286 def filename(self, value: str | None) -> None: 

287 self._filename = value 

288 

289 @property 

290 def translator_class_name(self) -> str: 

291 """Name of the metadata translator class used for these data.""" 

292 return self._translator_class_name 

293 

294 @property 

295 def extensions(self) -> dict[str, PropertyDefinition]: 

296 """Definitions of the translator-specific extension properties.""" 

297 return self._extensions 

298 

299 @property 

300 def all_properties(self) -> dict[str, PropertyDefinition]: 

301 """Definitions of all known properties (core plus extensions).""" 

302 return self._all_properties 

303 

304 @overload 

305 def __init__( 305 ↛ exitline 305 didn't return from function '__init__' because

306 self, 

307 header: MutableMapping[str, Any], 

308 filename: str | ResourcePath | None = None, 

309 translator_class: type[MetadataTranslator] | None = None, 

310 pedantic: bool = False, 

311 search_path: Sequence[str] | None = None, 

312 required: set[str] | None = None, 

313 subset: set[str] | None = None, 

314 quiet: bool = False, 

315 ) -> None: ... 

316 

317 @overload 

318 def __init__( 318 ↛ exitline 318 didn't return from function '__init__' because

319 self, 

320 header: None = None, 

321 filename: str | ResourcePath | None = None, 

322 translator_class: type[MetadataTranslator] | None = None, 

323 **kwargs: Any, 

324 ) -> None: ... 

325 

326 def __init__( 

327 self, 

328 header: MutableMapping[str, Any] | None = None, 

329 filename: str | ResourcePath | None = None, 

330 translator_class: type[MetadataTranslator] | None = None, 

331 pedantic: bool = False, 

332 search_path: Sequence[str] | None = None, 

333 required: set[str] | None = None, 

334 subset: set[str] | None = None, 

335 quiet: bool = False, 

336 **kwargs: Any, 

337 ) -> None: 

338 if filename is not None: 

339 filename = str(ResourcePath(filename, forceAbsolute=True)) 

340 if header is not None: 

341 if kwargs: 

342 raise ValueError( 

343 "kwargs not allowed if constructor given a header to translate. " 

344 f"Unrecognized keys: {[k for k in kwargs]}" 

345 ) 

346 self._init_from_header( 

347 header, 

348 filename=filename, 

349 translator_class=translator_class, 

350 pedantic=pedantic, 

351 search_path=search_path, 

352 required=required, 

353 subset=subset, 

354 quiet=quiet, 

355 ) 

356 return 

357 

358 self._init_from_kwargs(filename=filename, translator_class=translator_class, **kwargs) 

359 

360 def _init_from_header( 

361 self, 

362 header: MutableMapping[str, Any], 

363 *, 

364 filename: str | None, 

365 translator_class: type[MetadataTranslator] | None, 

366 pedantic: bool, 

367 search_path: Sequence[str] | None, 

368 required: set[str] | None, 

369 subset: set[str] | None, 

370 quiet: bool, 

371 ) -> None: 

372 super().__init__() 

373 self._filename = filename 

374 self._sealed = False 

375 # Initialize the empty object 

376 self._header = {} 

377 self._translator = None 

378 failure_level = logging.DEBUG if quiet else logging.WARNING 

379 

380 # Look for translator class before header fixup. fix_header calls 

381 # determine_translator immediately on the basis that you need to know 

382 # enough of the header to work out the translator before you can fix 

383 # it up. There is no gain in asking fix_header to determine the 

384 # translator and then trying to work it out again here. 

385 if translator_class is None: 

386 translator_class = MetadataTranslator.determine_translator(header, filename=filename) 

387 elif not issubclass(translator_class, MetadataTranslator): 

388 raise TypeError(f"Translator class must be a MetadataTranslator, not {translator_class}") 

389 

390 # Fix up the header (if required) 

391 fix_header(header, translator_class=translator_class, filename=filename, search_path=search_path) 

392 

393 # Store the supplied header for later stripping 

394 self._header = header 

395 

396 # This configures both self.extensions and self.all_properties. 

397 self._declare_extensions(translator_class.extensions) 

398 

399 # Create an instance for this header 

400 translator = translator_class(header, filename=filename) 

401 

402 # Store the translator 

403 self._translator = translator 

404 self._translator_class_name = translator_class.__name__ 

405 self.translator_name = translator_class.name 

406 

407 # Form file information string in case we need an error message 

408 if filename: 

409 file_info = f" and file {filename}" 

410 else: 

411 file_info = "" 

412 

413 # Determine the properties of interest 

414 full_set = set(self.all_properties) 

415 if subset is not None: 

416 if not subset: 

417 raise ValueError("Cannot request no properties be calculated.") 

418 if not subset.issubset(full_set): 

419 raise ValueError( 

420 f"Requested subset is not a subset of known properties. Got extra: {subset - full_set}" 

421 ) 

422 properties = subset 

423 else: 

424 properties = full_set 

425 

426 if required is None: 

427 required = set() 

428 else: 

429 if not required.issubset(full_set): 

430 raise ValueError(f"Requested required properties include unknowns: {required - full_set}") 

431 

432 # Loop over each property and request the translated form 

433 for property in properties: 

434 # prototype code 

435 method = f"to_{property}" 

436 

437 try: 

438 value = getattr(translator, method)() 

439 except NotImplementedError as e: 

440 raise NotImplementedError( 

441 f"No translation exists for property '{property}' using translator {translator.__class__}" 

442 ) from e 

443 except Exception as e: 

444 err_msg = ( 

445 f"Error calculating property '{property}' using " 

446 f"translator {translator.__class__}{file_info}" 

447 ) 

448 if pedantic or property in required: 

449 raise KeyError(err_msg) from e 

450 else: 

451 log.debug("Calculation of property '%s' failed with header: %s", property, header) 

452 log.log(failure_level, f"Ignoring {err_msg}: {e}") 

453 continue 

454 

455 definition = self.all_properties[property] 

456 # Some translators can return a compatible form that needs to 

457 # be coerced to the correct type (e.g., returning SkyCoord when you 

458 # need AltAz). In theory we could patch the translators to return 

459 # AltAz but code has historically not been as picky about this 

460 # until pydantic turned up. 

461 value = self._coerce_from_simple(definition, value, {}) 

462 if not definition.is_value_conformant(value): 

463 err_msg = ( 

464 f"Value calculated for property '{property}' is wrong type " 

465 f"({type(value)} != {definition.str_type}) using translator {translator.__class__}" 

466 f"{file_info}" 

467 ) 

468 if pedantic or property in required: 

469 raise TypeError(err_msg) 

470 else: 

471 log.debug( 

472 "Calculation of property '%s' had unexpected type with header: %s", property, header 

473 ) 

474 log.log(failure_level, f"Ignoring {err_msg}") 

475 

476 if value is None and property in required: 

477 raise KeyError(f"Calculation of required property {property} resulted in a value of None") 

478 

479 object.__setattr__(self, property, value) # allows setting even write-protected extensions 

480 

481 self._sealed = True 

482 

483 def _init_from_kwargs( 

484 self, 

485 *, 

486 filename: str | None, 

487 translator_class: type[MetadataTranslator] | None, 

488 **kwargs: Any, 

489 ) -> None: 

490 supplied_keys = set(kwargs) 

491 # Accept both the wire-format alias ``_translator`` and the field name 

492 # ``translator_name`` for robustness against callers that pass the 

493 # field name directly. 

494 translator_name = kwargs.pop("_translator", None) or kwargs.pop("translator_name", None) 

495 supplied_extensions = kwargs.pop("_extensions", None) 

496 if translator_name is not None: 

497 if translator_name not in MetadataTranslator.translators: 

498 raise KeyError(f"Unrecognized translator: {translator_name}") 

499 translator_class = MetadataTranslator.translators[translator_name] 

500 

501 if translator_class is not None and not issubclass(translator_class, MetadataTranslator): 

502 raise TypeError(f"Translator class must be a MetadataTranslator, not {translator_class}") 

503 

504 if supplied_extensions is not None: 

505 if translator_class is not None: 

506 raise ValueError("Provide either translator_class or _extensions, not both.") 

507 if not isinstance(supplied_extensions, dict): 

508 raise TypeError("_extensions must be a dictionary of PropertyDefinition entries.") 

509 extensions = supplied_extensions 

510 else: 

511 extensions = translator_class.extensions if translator_class is not None else {} 

512 

513 all_properties = self._get_all_properties(extensions) 

514 for key in kwargs: 

515 if key not in all_properties: 

516 raise KeyError(f"Unrecognized property '{key}' provided") 

517 

518 processed = {k: v for k, v in kwargs.items() if k in PROPERTIES and v is not None} 

519 processed = self._apply_constructor_defaults(processed, supplied_keys) 

520 

521 super().__init__(**processed) 

522 self._filename = filename 

523 self._sealed = False 

524 

525 # This configures both self.extensions and self.all_properties. 

526 self._declare_extensions(extensions) 

527 

528 # Handle extensions. 

529 ext_input = {k: v for k, v in kwargs.items() if k.startswith("ext_")} 

530 processed_ext = self._validate_property_mapping(ext_input, extensions) 

531 for key, value in processed_ext.items(): 

532 object.__setattr__(self, key, value) 

533 

534 if translator_class is not None: 

535 self._translator = translator_class({}) 

536 self._translator_class_name = translator_class.__name__ 

537 self.translator_name = translator_class.name 

538 

539 self._sealed = True 

540 

541 @staticmethod 

542 def _apply_constructor_defaults(processed: dict[str, Any], supplied_keys: set[str]) -> dict[str, Any]: 

543 """Apply derived/default values for kwargs-style construction. 

544 

545 Parameters 

546 ---------- 

547 processed : `dict` [`str`, `typing.Any`] 

548 Properties validated from kwargs input. 

549 supplied_keys : `set` [`str`] 

550 Property names explicitly supplied by the caller. 

551 

552 Returns 

553 ------- 

554 updated : `dict` [`str`, `typing.Any`] 

555 Updated property mapping with defaults/backfilled values applied. 

556 """ 

557 updated = dict(processed) 

558 for key in ("group_counter_start", "group_counter_end"): 

559 if ( 

560 key not in supplied_keys 

561 and "observation_counter" in supplied_keys 

562 and "observation_counter" in updated 

563 ): 

564 updated[key] = updated["observation_counter"] 

565 if "has_simulated_content" not in supplied_keys: 

566 updated["has_simulated_content"] = False 

567 return updated 

568 

569 @classmethod 

570 def from_header( 

571 cls, 

572 header: MutableMapping[str, Any], 

573 *, 

574 filename: str | None = None, 

575 translator_class: type[MetadataTranslator] | None = None, 

576 pedantic: bool = False, 

577 search_path: Sequence[str] | None = None, 

578 required: set[str] | None = None, 

579 subset: set[str] | None = None, 

580 quiet: bool = False, 

581 ) -> ObservationInfo: 

582 """Create an `ObservationInfo` by translating a metadata header. 

583 

584 Parameters 

585 ---------- 

586 header : `dict`-like 

587 Header mapping to translate. 

588 filename : `str`, optional 

589 Name of file associated with this header. 

590 translator_class : `MetadataTranslator`-class, optional 

591 Translator class to use. If `None`, translator will be 

592 auto-determined. 

593 pedantic : `bool`, optional 

594 If `True`, translation failures are fatal. 

595 search_path : `~collections.abc.Sequence` [`str`], optional 

596 Search paths for header corrections. 

597 required : `set` [`str`], optional 

598 Properties that must be translated and non-`None`. 

599 subset : `set` [`str`], optional 

600 Restrict translation to this subset of properties. 

601 quiet : `bool`, optional 

602 If `True`, warning level log messages that would be issued in non 

603 pedantic mode are converted to debug messages. 

604 

605 Returns 

606 ------- 

607 obsinfo : `ObservationInfo` 

608 Translated observation metadata. 

609 """ 

610 return cls( 

611 header=header, 

612 filename=filename, 

613 translator_class=translator_class, 

614 pedantic=pedantic, 

615 search_path=search_path, 

616 required=required, 

617 subset=subset, 

618 quiet=quiet, 

619 ) 

620 

621 @staticmethod 

622 def _get_all_properties( 

623 extensions: dict[str, PropertyDefinition] | None = None, 

624 ) -> dict[str, PropertyDefinition]: 

625 """Return the definitions of all properties. 

626 

627 Parameters 

628 ---------- 

629 extensions : `dict` [`str`: `PropertyDefinition`] 

630 List of extension property definitions, indexed by name (with no 

631 "ext_" prefix). 

632 

633 Returns 

634 ------- 

635 properties : `dict` [`str`: `PropertyDefinition`] 

636 Merged list of all property definitions, indexed by name. Extension 

637 properties will be listed with an ``ext_`` prefix. 

638 """ 

639 properties = dict(PROPERTIES) 

640 if extensions: 

641 properties.update({"ext_" + pp: dd for pp, dd in extensions.items()}) 

642 return properties 

643 

644 def _declare_extensions(self, extensions: dict[str, PropertyDefinition] | None) -> None: 

645 """Declare and set up extension properties. 

646 

647 This should always be called internally as part of the creation of a 

648 new `ObservationInfo`. 

649 

650 The core set of properties are declared as model fields at import 

651 time. Extension properties have to be configured at runtime (because 

652 we don't know what they will be until we look at the header and figure 

653 out what instrument we're dealing with), so we add them to the model 

654 and then use ``__setattr__`` to protect them as read-only. All 

655 extension properties are set to `None`. 

656 

657 Parameters 

658 ---------- 

659 extensions : `dict` [`str`: `PropertyDefinition`] 

660 List of extension property definitions, indexed by name (with no 

661 "ext_" prefix). 

662 """ 

663 if extensions: 

664 for name in extensions: 

665 field_name = "ext_" + name 

666 if not hasattr(self, field_name): 

667 object.__setattr__(self, field_name, None) 

668 self._extensions = extensions 

669 self._all_properties = self._get_all_properties(self._extensions) 

670 

671 def __setattr__(self, name: str, value: Any) -> Any: 

672 """Set attribute. 

673 

674 This provides read-only protection for all properties once the 

675 instance has been sealed. 

676 

677 Parameters 

678 ---------- 

679 name : `str` 

680 Name of attribute to set. 

681 value : `typing.Any` 

682 Value to set it to. 

683 """ 

684 if ( 

685 getattr(self, "_sealed", False) 

686 and hasattr(self, "all_properties") 

687 and name in self.all_properties 

688 ): 

689 raise AttributeError(f"Attribute {name} is read-only") 

690 return super().__setattr__(name, value) 

691 

692 @classmethod 

693 def __get_pydantic_json_schema__( 

694 cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler 

695 ) -> JsonSchemaValue: 

696 # The model_serializer below is mode="wrap" returning dict[str, Any]; 

697 # in serialization mode pydantic would otherwise emit a trivial 

698 # "object" schema. Hand the inner model-fields schema to the handler 

699 # so the field-level WithJsonSchema/PlainSerializer annotations are 

700 # honored. 

701 inner = core_schema.get("schema", core_schema) 

702 schema = handler(inner) 

703 schema = handler.resolve_ref_schema(schema) 

704 # Allow translator-specific extension properties under the ext_ prefix. 

705 schema["patternProperties"] = { 

706 "^ext_": {"description": "Extension property value (translator-specific)."}, 

707 } 

708 # Anything other than declared properties and ext_* keys is rejected. 

709 schema["additionalProperties"] = False 

710 return schema 

711 

712 @model_serializer(mode="wrap") 

713 def _serialize(self, handler: Any) -> dict[str, Any]: 

714 # Field serialization uses the per-field PlainSerializer annotations. 

715 # ``serialize_by_alias=True`` in model_config emits the ``_translator`` 

716 # alias instead of ``translator_name`` (and applies when nested). 

717 result: dict[str, Any] = handler(self) 

718 # Strip None entries; the wire format omits unset properties. 

719 result = {k: v for k, v in result.items() if v is not None} 

720 # Append values for any extension properties (dynamic ext_* attrs). 

721 # Kept flat for backwards compatibility with the existing wire format. 

722 for ext_name, ext_def in self._extensions.items(): 

723 field_name = f"ext_{ext_name}" 

724 value = getattr(self, field_name, None) 

725 if value is None: 

726 continue 

727 simplifier = ext_def.to_simple 

728 result[field_name] = simplifier(value) if simplifier else value 

729 return result 

730 

731 @classmethod 

732 def _validate_property_mapping( 

733 cls, 

734 data: MutableMapping[str, Any], 

735 extensions: dict[str, PropertyDefinition] | None, 

736 ) -> dict[str, Any]: 

737 # Validate extension properties. 

738 properties = {f"ext_{name}": definition for name, definition in (extensions or {}).items()} 

739 processed: dict[str, Any] = {} 

740 

741 for key, value in data.items(): 

742 if key not in properties: 

743 raise KeyError(f"Unrecognized property '{key}' provided") 

744 if value is None: 

745 continue 

746 processed[key] = cls._coerce_property_value(key, value, properties, processed) 

747 return processed 

748 

749 @classmethod 

750 def _coerce_property_value( 

751 cls, 

752 key: str, 

753 value: Any, 

754 properties: dict[str, PropertyDefinition], 

755 processed: dict[str, Any], 

756 ) -> Any: 

757 definition = properties[key] 

758 converted = cls._coerce_from_simple(definition, value, processed) 

759 if not definition.is_value_conformant(converted): 

760 raise TypeError( 

761 f"Supplied value {value} for property {key} " 

762 f"should be of class {definition.str_type} not {converted.__class__}" 

763 ) 

764 return converted 

765 

766 @classmethod 

767 def _coerce_from_simple( 

768 cls, 

769 definition: PropertyDefinition, 

770 value: Any, 

771 processed: dict[str, Any], 

772 ) -> Any: 

773 if definition.is_value_conformant(value): 

774 return value 

775 complexifier = definition.from_simple 

776 if complexifier is None: 

777 # Not the correct type, assumes caller will check. 

778 return value 

779 return complexifier(value, **processed) 

780 

781 @property 

782 def cards_used(self) -> frozenset[str]: 

783 """Header cards used for the translation. 

784 

785 Returns 

786 ------- 

787 used : `frozenset` of `str` 

788 Set of card used. 

789 """ 

790 if not self._translator: 

791 return frozenset() 

792 return self._translator.cards_used() 

793 

794 def stripped_header(self) -> MutableMapping[str, Any]: 

795 """Return a copy of the supplied header with used keywords removed. 

796 

797 Returns 

798 ------- 

799 stripped : `dict`-like 

800 Same class as header supplied to constructor, but with the 

801 headers used to calculate the generic information removed. 

802 """ 

803 hdr = copy.copy(self._header) 

804 used = self.cards_used 

805 for c in used: 

806 if c in hdr: 

807 del hdr[c] 

808 return hdr 

809 

810 def __str__(self) -> str: 

811 # Put more interesting answers at front of list 

812 # and then do remainder 

813 priority = ("instrument", "telescope", "datetime_begin") 

814 properties = sorted(set(self.all_properties) - set(priority)) 

815 

816 result = "" 

817 for p in itertools.chain(priority, properties): 

818 value = getattr(self, p) 

819 if isinstance(value, astropy.time.Time): 

820 value.format = "isot" 

821 value = str(value.value) 

822 result += f"{p}: {value}\n" 

823 

824 return result 

825 

826 def __eq__(self, other: Any) -> bool: 

827 """Check equality with another object. 

828 

829 Compares equal if standard properties are equal. 

830 

831 Parameters 

832 ---------- 

833 other : `typing.Any` 

834 Thing to compare with. 

835 """ 

836 if not isinstance(other, ObservationInfo): 

837 return NotImplemented 

838 

839 # Compare simplified forms. 

840 # Cannot compare directly because nan will not equate as equal 

841 # whereas they should be equal for our purposes 

842 self_simple = self.to_simple()