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 

21from collections.abc import MutableMapping, Sequence 

22from typing import TYPE_CHECKING, Any, cast, overload 

23 

24import astropy.time 

25import numpy as np 

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 quiet : `bool`, optional 

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

89 pedantic mode are converted to debug messages. 

90 **kwargs : `typing.Any` 

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

92 mode creates an `ObservationInfo` directly from supplied properties 

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

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

95 

96 Raises 

97 ------ 

98 ValueError 

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

100 registered translators. Also raised if the request property subset 

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

102 with kwargs. 

103 TypeError 

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

105 KeyError 

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

107 mode is enabled and any translations fails. 

108 NotImplementedError 

109 Raised if the selected translator does not support a required 

110 property. 

111 

112 Notes 

113 ----- 

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

115 Additional properties may be defined, either through the 

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

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

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

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

120 

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

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

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

124 remaining parameters control the behavior of the translator. 

125 

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

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

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

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

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

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

132 from which to extract extension definitions. 

133 

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

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

136 header after construction will modify the internal cached header. 

137 

138 Values of the properties are read-only. 

139 """ 

140 

141 model_config = ConfigDict( 

142 extra="forbid", 

143 arbitrary_types_allowed=True, 

144 validate_assignment=False, 

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

146 ) 

147 

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

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

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

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

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

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

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

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

156 ) 

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

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

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

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

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

162 ) 

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

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

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

166 ) 

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

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

169 ) 

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

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

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

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

174 ) 

175 boresight_rotation_coord: str | None = Field( 

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

177 ) 

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

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

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

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

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

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

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

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

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

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

188 ) 

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

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

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

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

193 ) 

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

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

196 ) 

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

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

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

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

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

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

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

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

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

206 ) 

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

208 has_simulated_content: bool | None = Field( 

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

210 ) 

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

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

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

214 

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

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

217 _sealed: bool = PrivateAttr(default=False) 

218 

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

220 @classmethod 

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

222 assert info.field_name is not None 

223 definition = PROPERTIES[info.field_name] 

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

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

226 

227 @overload 

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

229 self, 

230 header: MutableMapping[str, Any], 

231 filename: str | ResourcePath | None = None, 

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

233 pedantic: bool = False, 

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

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

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

237 quiet: bool = False, 

238 ) -> None: ... 

239 

240 @overload 

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

242 self, 

243 header: None = None, 

244 filename: str | ResourcePath | None = None, 

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

246 **kwargs: Any, 

247 ) -> None: ... 

248 

249 def __init__( 

250 self, 

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

252 filename: str | ResourcePath | None = None, 

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

254 pedantic: bool = False, 

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

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

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

258 quiet: bool = False, 

259 **kwargs: Any, 

260 ) -> None: 

261 if filename is not None: 

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

263 if header is not None: 

264 if kwargs: 

265 raise ValueError( 

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

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

268 ) 

269 self._init_from_header( 

270 header, 

271 filename=filename, 

272 translator_class=translator_class, 

273 pedantic=pedantic, 

274 search_path=search_path, 

275 required=required, 

276 subset=subset, 

277 quiet=quiet, 

278 ) 

279 return 

280 

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

282 

283 def _init_from_header( 

284 self, 

285 header: MutableMapping[str, Any], 

286 *, 

287 filename: str | None, 

288 translator_class: type[MetadataTranslator] | None, 

289 pedantic: bool, 

290 search_path: Sequence[str] | None, 

291 required: set[str] | None, 

292 subset: set[str] | None, 

293 quiet: bool, 

294 ) -> None: 

295 super().__init__(filename=filename) 

296 self._sealed = False 

297 # Initialize the empty object 

298 self._header = {} 

299 self._translator = None 

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

301 

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

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

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

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

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

307 if translator_class is None: 

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

309 elif not issubclass(translator_class, MetadataTranslator): 

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

311 

312 # Fix up the header (if required) 

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

314 

315 # Store the supplied header for later stripping 

316 self._header = header 

317 

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

319 self._declare_extensions(translator_class.extensions) 

320 

321 # Create an instance for this header 

322 translator = translator_class(header, filename=filename) 

323 

324 # Store the translator 

325 self._translator = translator 

326 self.translator_class_name = translator_class.__name__ 

327 

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

329 if filename: 

330 file_info = f" and file {filename}" 

331 else: 

332 file_info = "" 

333 

334 # Determine the properties of interest 

335 full_set = set(self.all_properties) 

336 if subset is not None: 

337 if not subset: 

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

339 if not subset.issubset(full_set): 

340 raise ValueError( 

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

342 ) 

343 properties = subset 

344 else: 

345 properties = full_set 

346 

347 if required is None: 

348 required = set() 

349 else: 

350 if not required.issubset(full_set): 

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

352 

353 # Loop over each property and request the translated form 

354 for property in properties: 

355 # prototype code 

356 method = f"to_{property}" 

357 

358 try: 

359 value = getattr(translator, method)() 

360 except NotImplementedError as e: 

361 raise NotImplementedError( 

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

363 ) from e 

364 except Exception as e: 

365 err_msg = ( 

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

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

368 ) 

369 if pedantic or property in required: 

370 raise KeyError(err_msg) from e 

371 else: 

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

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

374 continue 

375 

376 definition = self.all_properties[property] 

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

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

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

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

381 # until pydantic turned up. 

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

383 if not definition.is_value_conformant(value): 

384 err_msg = ( 

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

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

387 f"{file_info}" 

388 ) 

389 if pedantic or property in required: 

390 raise TypeError(err_msg) 

391 else: 

392 log.debug( 

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

394 ) 

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

396 

397 if value is None and property in required: 

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

399 

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

401 

402 self._sealed = True 

403 

404 def _init_from_kwargs( 

405 self, 

406 *, 

407 filename: str | None, 

408 translator_class: type[MetadataTranslator] | None, 

409 **kwargs: Any, 

410 ) -> None: 

411 supplied_keys = set(kwargs) 

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

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

414 if translator_name is not None: 

415 if translator_name not in MetadataTranslator.translators: 

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

417 translator_class = MetadataTranslator.translators[translator_name] 

418 

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

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

421 

422 if supplied_extensions is not None: 

423 if translator_class is not None: 

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

425 if not isinstance(supplied_extensions, dict): 

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

427 extensions = supplied_extensions 

428 else: 

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

430 

431 all_properties = self._get_all_properties(extensions) 

432 for key in kwargs: 

433 if key not in all_properties: 

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

435 

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

437 processed = self._apply_constructor_defaults(processed, supplied_keys) 

438 

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

440 self._sealed = False 

441 

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

443 self._declare_extensions(extensions) 

444 

445 # Handle extensions. 

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

447 processed_ext = self._validate_property_mapping(ext_input, extensions) 

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

449 object.__setattr__(self, key, value) 

450 

451 if translator_class is not None: 

452 self._translator = translator_class({}) 

453 self.translator_class_name = translator_class.__name__ 

454 

455 self._sealed = True 

456 

457 @staticmethod 

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

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

460 

461 Parameters 

462 ---------- 

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

464 Properties validated from kwargs input. 

465 supplied_keys : `set` [`str`] 

466 Property names explicitly supplied by the caller. 

467 

468 Returns 

469 ------- 

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

471 Updated property mapping with defaults/backfilled values applied. 

472 """ 

473 updated = dict(processed) 

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

475 if ( 

476 key not in supplied_keys 

477 and "observation_counter" in supplied_keys 

478 and "observation_counter" in updated 

479 ): 

480 updated[key] = updated["observation_counter"] 

481 if "has_simulated_content" not in supplied_keys: 

482 updated["has_simulated_content"] = False 

483 return updated 

484 

485 @classmethod 

486 def from_header( 

487 cls, 

488 header: MutableMapping[str, Any], 

489 *, 

490 filename: str | None = None, 

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

492 pedantic: bool = False, 

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

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

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

496 quiet: bool = False, 

497 ) -> ObservationInfo: 

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

499 

500 Parameters 

501 ---------- 

502 header : `dict`-like 

503 Header mapping to translate. 

504 filename : `str`, optional 

505 Name of file associated with this header. 

506 translator_class : `MetadataTranslator`-class, optional 

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

508 auto-determined. 

509 pedantic : `bool`, optional 

510 If `True`, translation failures are fatal. 

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

512 Search paths for header corrections. 

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

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

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

516 Restrict translation to this subset of properties. 

517 quiet : `bool`, optional 

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

519 pedantic mode are converted to debug messages. 

520 

521 Returns 

522 ------- 

523 obsinfo : `ObservationInfo` 

524 Translated observation metadata. 

525 """ 

526 return cls( 

527 header=header, 

528 filename=filename, 

529 translator_class=translator_class, 

530 pedantic=pedantic, 

531 search_path=search_path, 

532 required=required, 

533 subset=subset, 

534 quiet=quiet, 

535 ) 

536 

537 @staticmethod 

538 def _get_all_properties( 

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

540 ) -> dict[str, PropertyDefinition]: 

541 """Return the definitions of all properties. 

542 

543 Parameters 

544 ---------- 

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

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

547 "ext_" prefix). 

548 

549 Returns 

550 ------- 

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

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

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

554 """ 

555 properties = dict(PROPERTIES) 

556 if extensions: 

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

558 return properties 

559 

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

561 """Declare and set up extension properties. 

562 

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

564 new `ObservationInfo`. 

565 

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

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

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

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

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

571 extension properties are set to `None`. 

572 

573 Parameters 

574 ---------- 

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

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

577 "ext_" prefix). 

578 """ 

579 if extensions: 

580 for name in extensions: 

581 field_name = "ext_" + name 

582 if not hasattr(self, field_name): 

583 object.__setattr__(self, field_name, None) 

584 self.extensions = extensions 

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

586 

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

588 """Set attribute. 

589 

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

591 instance has been sealed. 

592 

593 Parameters 

594 ---------- 

595 name : `str` 

596 Name of attribute to set. 

597 value : `typing.Any` 

598 Value to set it to. 

599 """ 

600 if ( 

601 getattr(self, "_sealed", False) 

602 and hasattr(self, "all_properties") 

603 and name in self.all_properties 

604 ): 

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

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

607 

608 @model_serializer(mode="plain") 

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

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

611 if self._translator and self._translator.name: 

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

613 

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

615 value = getattr(self, p) 

616 if value is None: 

617 continue 

618 simplifier = definition.to_simple 

619 if simplifier is None: 

620 simple[p] = value 

621 else: 

622 simple[p] = simplifier(value) 

623 

624 return simple 

625 

626 @classmethod 

627 def _validate_property_mapping( 

628 cls, 

629 data: MutableMapping[str, Any], 

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

631 ) -> dict[str, Any]: 

632 # Validate extension properties. 

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

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

635 

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

637 if key not in properties: 

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

639 if value is None: 

640 continue 

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

642 return processed 

643 

644 @classmethod 

645 def _coerce_property_value( 

646 cls, 

647 key: str, 

648 value: Any, 

649 properties: dict[str, PropertyDefinition], 

650 processed: dict[str, Any], 

651 ) -> Any: 

652 definition = properties[key] 

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

654 if not definition.is_value_conformant(converted): 

655 raise TypeError( 

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

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

658 ) 

659 return converted 

660 

661 @classmethod 

662 def _coerce_from_simple( 

663 cls, 

664 definition: PropertyDefinition, 

665 value: Any, 

666 processed: dict[str, Any], 

667 ) -> Any: 

668 if definition.is_value_conformant(value): 

669 return value 

670 complexifier = definition.from_simple 

671 if complexifier is None: 

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

673 return value 

674 return complexifier(value, **processed) 

675 

676 @property 

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

678 """Header cards used for the translation. 

679 

680 Returns 

681 ------- 

682 used : `frozenset` of `str` 

683 Set of card used. 

684 """ 

685 if not self._translator: 

686 return frozenset() 

687 return self._translator.cards_used() 

688 

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

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

691 

692 Returns 

693 ------- 

694 stripped : `dict`-like 

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

696 headers used to calculate the generic information removed. 

697 """ 

698 hdr = copy.copy(self._header) 

699 used = self.cards_used 

700 for c in used: 

701 if c in hdr: 

702 del hdr[c] 

703 return hdr 

704 

705 def __str__(self) -> str: 

706 # Put more interesting answers at front of list 

707 # and then do remainder 

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

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

710 

711 result = "" 

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

713 value = getattr(self, p) 

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

715 value.format = "isot" 

716 value = str(value.value) 

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

718 

719 return result 

720 

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

722 """Check equality with another object. 

723 

724 Compares equal if standard properties are equal. 

725 

726 Parameters 

727 ---------- 

728 other : `typing.Any` 

729 Thing to compare with. 

730 """ 

731 if not isinstance(other, ObservationInfo): 

732 return NotImplemented 

733 

734 # Compare simplified forms. 

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

736 # whereas they should be equal for our purposes 

737 self_simple = self.to_simple() 

738 other_simple = other.to_simple() 

739 

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

741 self_simple.pop("_translator", None) 

742 other_simple.pop("_translator", None) 

743 

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

745 self_value = self_simple[k] 

746 other_value = other_simple[k] 

747 if self_value != other_value: 

748 if isinstance(self_value, float) or ( 

749 isinstance(self_value, tuple) and isinstance(self_value[0], float) 

750 ): 

751 close = np.allclose(self_value, other_value, equal_nan=True) 

752 if close: 

753 continue 

754 return False 

755 return True 

756 

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

758 if not isinstance(other, ObservationInfo): 

759 return NotImplemented 

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

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

762 return self.datetime_begin < other.datetime_begin 

763 

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

765 if not isinstance(other, ObservationInfo): 

766 return NotImplemented 

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

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

769 return self.datetime_begin > other.datetime_begin 

770 

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

772 """Get pickleable state. 

773 

774 Returns the properties. Deliberately does not preserve the full 

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

776 translator. 

777 

778 Returns 

779 ------- 

780 state : `dict` 

781 Pickled state. 

782 """ 

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

784 for p in self.all_properties: 

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

786 

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

788 

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

790 """Set object state from pickle. 

791 

792 Parameters 

793 ---------- 

794 state : `dict` 

795 Pickled state. 

796 """ 

797 state_any = cast(Any, state) 

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

799 state = state_any["state"] 

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

801 else: 

802 try: 

803 state, extensions = state_any 

804 except ValueError: 

805 # Backwards compatibility for pickles generated before DM-34175