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

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 

21import math 

22from collections.abc import MutableMapping, Sequence 

23from typing import TYPE_CHECKING, Any, cast, overload 

24 

25import astropy.time 

26from lsst.resources import ResourcePath 

27from pydantic import ( 

28 BaseModel, 

29 ConfigDict, 

30 Field, 

31 PrivateAttr, 

32 ValidationInfo, 

33 field_validator, 

34 model_serializer, 

35) 

36 

37from .headers import fix_header 

38from .properties import ( 

39 PROPERTIES, 

40 PropertyDefinition, 

41) 

42from .translator import MetadataTranslator 

43 

44if TYPE_CHECKING: 

45 import astropy.coordinates 

46 import astropy.units 

47 

48log = logging.getLogger(__name__) 

49_CORE_FROM_SIMPLE_FIELDS = tuple(name for name, definition in PROPERTIES.items() if definition.from_simple) 

50 

51 

52class ObservationInfo(BaseModel): 

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

54 exposure observation. 

55 

56 Parameters 

57 ---------- 

58 header : `dict`-like 

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

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

61 filename : `str`, optional 

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

63 datasets with missing header information this can sometimes 

64 allow for some fixups in translations. 

65 translator_class : `MetadataTranslator`-class, optional 

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

67 into standard form. Otherwise each registered translator class will 

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

69 pedantic : `bool`, optional 

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

71 individual property translations must all be implemented but can fail 

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

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

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

75 ``header`` is specified. 

76 required : `set`, optional 

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

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

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

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

81 subset : `set`, optional 

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

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

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

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

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

87 **kwargs : `typing.Any` 

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

89 mode creates an `ObservationInfo` directly from supplied properties 

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

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

92 

93 Raises 

94 ------ 

95 ValueError 

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

97 registered translators. Also raised if the request property subset 

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

99 with kwargs. 

100 TypeError 

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

102 KeyError 

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

104 mode is enabled and any translations fails. 

105 NotImplementedError 

106 Raised if the selected translator does not support a required 

107 property. 

108 

109 Notes 

110 ----- 

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

112 Additional properties may be defined, either through the 

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

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

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

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

117 

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

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

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

121 remaining parameters control the behavior of the translator. 

122 

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

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

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

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

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

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

129 from which to extract extension definitions. 

130 

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

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

133 header after construction will modify the internal cached header. 

134 

135 Values of the properties are read-only. 

136 """ 

137 

138 model_config = ConfigDict( 

139 extra="forbid", 

140 arbitrary_types_allowed=True, 

141 validate_assignment=False, 

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

143 ) 

144 

145 filename: str | None = Field(default=None, exclude=True) 

146 translator_class_name: str = Field(default="<None>", exclude=True) 

147 extensions: dict[str, PropertyDefinition] = Field(default_factory=dict, exclude=True) 

148 all_properties: dict[str, PropertyDefinition] = Field(default_factory=dict, exclude=True) 

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

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

151 location: astropy.coordinates.EarthLocation | None = Field( 

152 default=None, description=PROPERTIES["location"].doc 

153 ) 

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

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

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

157 datetime_begin: astropy.time.Time | None = Field( 

158 default=None, description=PROPERTIES["datetime_begin"].doc 

159 ) 

160 datetime_end: astropy.time.Time | None = Field(default=None, description=PROPERTIES["datetime_end"].doc) 

161 exposure_time: astropy.units.Quantity | None = Field( 

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

163 ) 

164 exposure_time_requested: astropy.units.Quantity | None = Field( 

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

166 ) 

167 dark_time: astropy.units.Quantity | None = Field(default=None, description=PROPERTIES["dark_time"].doc) 

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

169 boresight_rotation_angle: astropy.coordinates.Angle | None = Field( 

170 default=None, description=PROPERTIES["boresight_rotation_angle"].doc 

171 ) 

172 boresight_rotation_coord: str | None = Field( 

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

174 ) 

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

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

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

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

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

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

181 focus_z: astropy.units.Quantity | None = Field(default=None, description=PROPERTIES["focus_z"].doc) 

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

183 temperature: astropy.units.Quantity | None = Field( 

184 default=None, description=PROPERTIES["temperature"].doc 

185 ) 

186 pressure: astropy.units.Quantity | None = Field(default=None, description=PROPERTIES["pressure"].doc) 

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

188 tracking_radec: astropy.coordinates.SkyCoord | None = Field( 

189 default=None, description=PROPERTIES["tracking_radec"].doc 

190 ) 

191 altaz_begin: astropy.coordinates.AltAz | None = Field( 

192 default=None, description=PROPERTIES["altaz_begin"].doc 

193 ) 

194 altaz_end: astropy.coordinates.AltAz | None = Field(default=None, description=PROPERTIES["altaz_end"].doc) 

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

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

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

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

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

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

201 observing_day_offset: astropy.time.TimeDelta | None = Field( 

202 default=None, description=PROPERTIES["observing_day_offset"].doc 

203 ) 

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

205 has_simulated_content: bool | None = Field( 

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

207 ) 

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

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

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

211 

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

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

214 _sealed: bool = PrivateAttr(default=False) 

215 

216 @field_validator(*_CORE_FROM_SIMPLE_FIELDS, mode="before") 

217 @classmethod 

218 def _before_core_from_simple(cls, value: Any, info: ValidationInfo) -> Any: 

219 assert info.field_name is not None 

220 definition = PROPERTIES[info.field_name] 

221 context = info.data if isinstance(info.data, dict) else {} 

222 return cls._coerce_from_simple(definition, value, context) 

223 

224 @overload 

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

226 self, 

227 header: MutableMapping[str, Any], 

228 filename: str | ResourcePath | None = None, 

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

230 pedantic: bool = False, 

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

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

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

234 ) -> None: ... 

235 

236 @overload 

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

238 self, 

239 header: None = None, 

240 filename: str | ResourcePath | None = None, 

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

242 **kwargs: Any, 

243 ) -> None: ... 

244 

245 def __init__( 

246 self, 

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

248 filename: str | ResourcePath | None = None, 

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

250 pedantic: bool = False, 

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

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

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

254 **kwargs: Any, 

255 ) -> None: 

256 if filename is not None: 

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

258 if header is not None: 

259 if kwargs: 

260 raise ValueError( 

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

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

263 ) 

264 self._init_from_header( 

265 header, 

266 filename=filename, 

267 translator_class=translator_class, 

268 pedantic=pedantic, 

269 search_path=search_path, 

270 required=required, 

271 subset=subset, 

272 ) 

273 return 

274 

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

276 

277 def _init_from_header( 

278 self, 

279 header: MutableMapping[str, Any], 

280 *, 

281 filename: str | None, 

282 translator_class: type[MetadataTranslator] | None, 

283 pedantic: bool, 

284 search_path: Sequence[str] | None, 

285 required: set[str] | None, 

286 subset: set[str] | None, 

287 ) -> None: 

288 super().__init__(filename=filename) 

289 self._sealed = False 

290 # Initialize the empty object 

291 self._header = {} 

292 self._translator = None 

293 

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

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

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

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

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

299 if translator_class is None: 

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

301 elif not issubclass(translator_class, MetadataTranslator): 

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

303 

304 # Fix up the header (if required) 

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

306 

307 # Store the supplied header for later stripping 

308 self._header = header 

309 

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

311 self._declare_extensions(translator_class.extensions) 

312 

313 # Create an instance for this header 

314 translator = translator_class(header, filename=filename) 

315 

316 # Store the translator 

317 self._translator = translator 

318 self.translator_class_name = translator_class.__name__ 

319 

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

321 if filename: 

322 file_info = f" and file {filename}" 

323 else: 

324 file_info = "" 

325 

326 # Determine the properties of interest 

327 full_set = set(self.all_properties) 

328 if subset is not None: 

329 if not subset: 

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

331 if not subset.issubset(full_set): 

332 raise ValueError( 

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

334 ) 

335 properties = subset 

336 else: 

337 properties = full_set 

338 

339 if required is None: 

340 required = set() 

341 else: 

342 if not required.issubset(full_set): 

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

344 

345 # Loop over each property and request the translated form 

346 for property in properties: 

347 # prototype code 

348 method = f"to_{property}" 

349 

350 try: 

351 value = getattr(translator, method)() 

352 except NotImplementedError as e: 

353 raise NotImplementedError( 

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

355 ) from e 

356 except Exception as e: 

357 err_msg = ( 

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

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

360 ) 

361 if pedantic or property in required: 

362 raise KeyError(err_msg) from e 

363 else: 

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

365 log.warning(f"Ignoring {err_msg}: {e}") 

366 continue 

367 

368 definition = self.all_properties[property] 

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

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

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

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

373 # until pydantic turned up. 

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

375 if not definition.is_value_conformant(value): 

376 err_msg = ( 

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

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

379 f"{file_info}" 

380 ) 

381 if pedantic or property in required: 

382 raise TypeError(err_msg) 

383 else: 

384 log.debug( 

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

386 ) 

387 log.warning(f"Ignoring {err_msg}") 

388 

389 if value is None and property in required: 

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

391 

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

393 

394 self._sealed = True 

395 

396 def _init_from_kwargs( 

397 self, 

398 *, 

399 filename: str | None, 

400 translator_class: type[MetadataTranslator] | None, 

401 **kwargs: Any, 

402 ) -> None: 

403 supplied_keys = set(kwargs) 

404 translator_name = kwargs.pop("_translator", None) 

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

406 if translator_name is not None: 

407 if translator_name not in MetadataTranslator.translators: 

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

409 translator_class = MetadataTranslator.translators[translator_name] 

410 

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

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

413 

414 if supplied_extensions is not None: 

415 if translator_class is not None: 

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

417 if not isinstance(supplied_extensions, dict): 

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

419 extensions = supplied_extensions 

420 else: 

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

422 

423 all_properties = self._get_all_properties(extensions) 

424 for key in kwargs: 

425 if key not in all_properties: 

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

427 

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

429 processed = self._apply_constructor_defaults(processed, supplied_keys) 

430 

431 super().__init__(filename=filename, **processed) 

432 self._sealed = False 

433 

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

435 self._declare_extensions(extensions) 

436 

437 # Handle extensions. 

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

439 processed_ext = self._validate_property_mapping(ext_input, extensions) 

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

441 object.__setattr__(self, key, value) 

442 

443 if translator_class is not None: 

444 self._translator = translator_class({}) 

445 self.translator_class_name = translator_class.__name__ 

446 

447 self._sealed = True 

448 

449 @staticmethod 

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

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

452 

453 Parameters 

454 ---------- 

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

456 Properties validated from kwargs input. 

457 supplied_keys : `set` [`str`] 

458 Property names explicitly supplied by the caller. 

459 

460 Returns 

461 ------- 

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

463 Updated property mapping with defaults/backfilled values applied. 

464 """ 

465 updated = dict(processed) 

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

467 if ( 

468 key not in supplied_keys 

469 and "observation_counter" in supplied_keys 

470 and "observation_counter" in updated 

471 ): 

472 updated[key] = updated["observation_counter"] 

473 if "has_simulated_content" not in supplied_keys: 

474 updated["has_simulated_content"] = False 

475 return updated 

476 

477 @classmethod 

478 def from_header( 

479 cls, 

480 header: MutableMapping[str, Any], 

481 *, 

482 filename: str | None = None, 

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

484 pedantic: bool = False, 

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

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

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

488 ) -> ObservationInfo: 

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

490 

491 Parameters 

492 ---------- 

493 header : `dict`-like 

494 Header mapping to translate. 

495 filename : `str`, optional 

496 Name of file associated with this header. 

497 translator_class : `MetadataTranslator`-class, optional 

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

499 auto-determined. 

500 pedantic : `bool`, optional 

501 If `True`, translation failures are fatal. 

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

503 Search paths for header corrections. 

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

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

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

507 Restrict translation to this subset of properties. 

508 

509 Returns 

510 ------- 

511 obsinfo : `ObservationInfo` 

512 Translated observation metadata. 

513 """ 

514 return cls( 

515 header=header, 

516 filename=filename, 

517 translator_class=translator_class, 

518 pedantic=pedantic, 

519 search_path=search_path, 

520 required=required, 

521 subset=subset, 

522 ) 

523 

524 @staticmethod 

525 def _get_all_properties( 

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

527 ) -> dict[str, PropertyDefinition]: 

528 """Return the definitions of all properties. 

529 

530 Parameters 

531 ---------- 

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

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

534 "ext_" prefix). 

535 

536 Returns 

537 ------- 

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

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

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

541 """ 

542 properties = dict(PROPERTIES) 

543 if extensions: 

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

545 return properties 

546 

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

548 """Declare and set up extension properties. 

549 

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

551 new `ObservationInfo`. 

552 

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

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

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

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

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

558 extension properties are set to `None`. 

559 

560 Parameters 

561 ---------- 

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

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

564 "ext_" prefix). 

565 """ 

566 if extensions: 

567 for name in extensions: 

568 field_name = "ext_" + name 

569 if not hasattr(self, field_name): 

570 object.__setattr__(self, field_name, None) 

571 self.extensions = extensions 

572 self.all_properties = self._get_all_properties(self.extensions) 

573 

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

575 """Set attribute. 

576 

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

578 instance has been sealed. 

579 

580 Parameters 

581 ---------- 

582 name : `str` 

583 Name of attribute to set. 

584 value : `typing.Any` 

585 Value to set it to. 

586 """ 

587 if ( 

588 getattr(self, "_sealed", False) 

589 and hasattr(self, "all_properties") 

590 and name in self.all_properties 

591 ): 

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

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

594 

595 @model_serializer(mode="plain") 

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

597 simple: dict[str, Any] = {} 

598 if self._translator and self._translator.name: 

599 simple["_translator"] = self._translator.name 

600 

601 for p, definition in self.all_properties.items(): 

602 value = getattr(self, p) 

603 if value is None: 

604 continue 

605 simplifier = definition.to_simple 

606 if simplifier is None: 

607 simple[p] = value 

608 else: 

609 simple[p] = simplifier(value) 

610 

611 return simple 

612 

613 @classmethod 

614 def _validate_property_mapping( 

615 cls, 

616 data: MutableMapping[str, Any], 

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

618 ) -> dict[str, Any]: 

619 # Validate extension properties. 

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

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

622 

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

624 if key not in properties: 

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

626 if value is None: 

627 continue 

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

629 return processed 

630 

631 @classmethod 

632 def _coerce_property_value( 

633 cls, 

634 key: str, 

635 value: Any, 

636 properties: dict[str, PropertyDefinition], 

637 processed: dict[str, Any], 

638 ) -> Any: 

639 definition = properties[key] 

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

641 if not definition.is_value_conformant(converted): 

642 raise TypeError( 

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

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

645 ) 

646 return converted 

647 

648 @classmethod 

649 def _coerce_from_simple( 

650 cls, 

651 definition: PropertyDefinition, 

652 value: Any, 

653 processed: dict[str, Any], 

654 ) -> Any: 

655 if definition.is_value_conformant(value): 

656 return value 

657 complexifier = definition.from_simple 

658 if complexifier is None: 

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

660 return value 

661 return complexifier(value, **processed) 

662 

663 @property 

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

665 """Header cards used for the translation. 

666 

667 Returns 

668 ------- 

669 used : `frozenset` of `str` 

670 Set of card used. 

671 """ 

672 if not self._translator: 

673 return frozenset() 

674 return self._translator.cards_used() 

675 

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

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

678 

679 Returns 

680 ------- 

681 stripped : `dict`-like 

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

683 headers used to calculate the generic information removed. 

684 """ 

685 hdr = copy.copy(self._header) 

686 used = self.cards_used 

687 for c in used: 

688 if c in hdr: 

689 del hdr[c] 

690 return hdr 

691 

692 def __str__(self) -> str: 

693 # Put more interesting answers at front of list 

694 # and then do remainder 

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

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

697 

698 result = "" 

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

700 value = getattr(self, p) 

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

702 value.format = "isot" 

703 value = str(value.value) 

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

705 

706 return result 

707 

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

709 """Check equality with another object. 

710 

711 Compares equal if standard properties are equal. 

712 

713 Parameters 

714 ---------- 

715 other : `typing.Any` 

716 Thing to compare with. 

717 """ 

718 if not isinstance(other, ObservationInfo): 

719 return NotImplemented 

720 

721 # Compare simplified forms. 

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

723 # whereas they should be equal for our purposes 

724 self_simple = self.to_simple() 

725 other_simple = other.to_simple() 

726 

727 # We don't care about the translator internal detail 

728 self_simple.pop("_translator", None) 

729 other_simple.pop("_translator", None) 

730 

731 for k in self_simple.keys() & other_simple.keys(): 

732 self_value = self_simple[k] 

733 other_value = other_simple[k] 

734 if self_value != other_value: 

735 try: 

736 both_nan = math.isnan(self_value) and math.isnan(other_value) 

737 except TypeError: 

738 both_nan = False 

739 if both_nan: 

740 # If both are nan this is fine 

741 continue 

742 return False 

743 return True 

744 

745 def __lt__(self, other: Any) -> bool: 

746 if not isinstance(other, ObservationInfo): 

747 return NotImplemented 

748 if self.datetime_begin is None or other.datetime_begin is None: 

749 raise TypeError("Cannot compare ObservationInfo without datetime_begin values") 

750 return self.datetime_begin < other.datetime_begin 

751 

752 def __gt__(self, other: Any) -> bool: 

753 if not isinstance(other, ObservationInfo): 

754 return NotImplemented 

755 if self.datetime_begin is None or other.datetime_begin is None: 

756 raise TypeError("Cannot compare ObservationInfo without datetime_begin values") 

757 return self.datetime_begin > other.datetime_begin 

758 

759 def __getstate__(self) -> dict[str, Any]: 

760 """Get pickleable state. 

761 

762 Returns the properties. Deliberately does not preserve the full 

763 current state; in particular, does not return the full header or 

764 translator. 

765 

766 Returns 

767 ------- 

768 state : `dict` 

769 Pickled state. 

770 """ 

771 state: dict[str, Any] = {} 

772 for p in self.all_properties: 

773 state[p] = getattr(self, p) 

774 

775 return {"state": state, "extensions": self.extensions} 

776 

777 def __setstate__(self, state: dict[Any, Any]) -> None: 

778 """Set object state from pickle. 

779 

780 Parameters 

781 ---------- 

782 state : `dict` 

783 Pickled state. 

784 """ 

785 state_any = cast(Any, state) 

786 if isinstance(state_any, dict) and "state" in state_any: 

787 state = state_any["state"] 

788 extensions = state_any.get("extensions", {}) 

789 else: 

790 try: 

791 state, extensions = state_any 

792 except ValueError: 

793 # Backwards compatibility for pickles generated before DM-34175 

794 extensions = {} 

795 super().__init__() 

796 self._sealed = False 

797 self._declare_extensions(extensions) 

798 for p in self.all_properties: 

799 # allows setting even write-protected extensions 

800 object.__setattr__(self, p, state[p]) 

801 self._sealed = True 

802 

803 def to_simple(self) -> MutableMapping[str, Any]: 

804 """Convert the contents of this object to simple dict form. 

805 

806 The keys of the dict are the standard properties but the values