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

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): 462 ↛ 463line 462 didn't jump to line 463 because the condition on line 462 was never true

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 and translator_name in MetadataTranslator.translators: 

497 translator_class = MetadataTranslator.translators[translator_name] 

498 

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

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

501 

502 if supplied_extensions is not None: 

503 if translator_class is not None: 

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

505 if not isinstance(supplied_extensions, dict): 

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

507 extensions = supplied_extensions 

508 else: 

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

510 

511 all_properties = self._get_all_properties(extensions) 

512 unrecognized = [key for key in kwargs if key not in all_properties] 

513 if unrecognized: 

514 # Extension properties (the ``ext_`` prefixed keys) are defined by 

515 # the translator. When the translator could not be resolved its 

516 # extension definitions are unavailable, so report that as the 

517 # underlying cause rather than an opaque unknown property. 

518 extension_keys = sorted(key for key in unrecognized if key.startswith("ext_")) 

519 if extension_keys and translator_class is None: 

520 reason = ( 

521 f"translator '{translator_name}' is not registered" 

522 if translator_name is not None 

523 else "no translator was specified" 

524 ) 

525 raise KeyError( 

526 f"Unrecognized extension properties {extension_keys}: {reason}, so the " 

527 "extension property definitions are unavailable." 

528 ) 

529 text = ( 

530 f"property ({unrecognized[0]})" 

531 if len(unrecognized) == 1 

532 else f"properties ({', '.join(unrecognized)})" 

533 ) 

534 raise KeyError(f"Unrecognized {text} provided") 

535 

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

537 processed = self._apply_constructor_defaults(processed, supplied_keys) 

538 

539 super().__init__(**processed) 

540 self._filename = filename 

541 self._sealed = False 

542 

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

544 self._declare_extensions(extensions) 

545 

546 # Handle extensions. 

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

548 processed_ext = self._validate_property_mapping(ext_input, extensions) 

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

550 object.__setattr__(self, key, value) 

551 

552 if translator_class is not None: 

553 self._translator = translator_class({}) 

554 self._translator_class_name = translator_class.__name__ 

555 self.translator_name = translator_class.name 

556 elif translator_name is not None: 

557 # The named translator is not registered in this process. The 

558 # translator instance is only needed while constructing from a 

559 # header, so it is left as None and the name from the serialization 

560 # is recorded directly. 

561 self._translator_class_name = translator_name 

562 self.translator_name = translator_name 

563 

564 self._sealed = True 

565 

566 @staticmethod 

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

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

569 

570 Parameters 

571 ---------- 

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

573 Properties validated from kwargs input. 

574 supplied_keys : `set` [`str`] 

575 Property names explicitly supplied by the caller. 

576 

577 Returns 

578 ------- 

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

580 Updated property mapping with defaults/backfilled values applied. 

581 """ 

582 updated = dict(processed) 

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

584 if ( 

585 key not in supplied_keys 

586 and "observation_counter" in supplied_keys 

587 and "observation_counter" in updated 

588 ): 

589 updated[key] = updated["observation_counter"] 

590 if "has_simulated_content" not in supplied_keys: 

591 updated["has_simulated_content"] = False 

592 return updated 

593 

594 @classmethod 

595 def from_header( 

596 cls, 

597 header: MutableMapping[str, Any], 

598 *, 

599 filename: str | None = None, 

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

601 pedantic: bool = False, 

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

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

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

605 quiet: bool = False, 

606 ) -> ObservationInfo: 

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

608 

609 Parameters 

610 ---------- 

611 header : `dict`-like 

612 Header mapping to translate. 

613 filename : `str`, optional 

614 Name of file associated with this header. 

615 translator_class : `MetadataTranslator`-class, optional 

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

617 auto-determined. 

618 pedantic : `bool`, optional 

619 If `True`, translation failures are fatal. 

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

621 Search paths for header corrections. 

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

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

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

625 Restrict translation to this subset of properties. 

626 quiet : `bool`, optional 

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

628 pedantic mode are converted to debug messages. 

629 

630 Returns 

631 ------- 

632 obsinfo : `ObservationInfo` 

633 Translated observation metadata. 

634 """ 

635 return cls( 

636 header=header, 

637 filename=filename, 

638 translator_class=translator_class, 

639 pedantic=pedantic, 

640 search_path=search_path, 

641 required=required, 

642 subset=subset, 

643 quiet=quiet, 

644 ) 

645 

646 @staticmethod 

647 def _get_all_properties( 

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

649 ) -> dict[str, PropertyDefinition]: 

650 """Return the definitions of all properties. 

651 

652 Parameters 

653 ---------- 

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

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

656 "ext_" prefix). 

657 

658 Returns 

659 ------- 

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

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

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

663 """ 

664 properties = dict(PROPERTIES) 

665 if extensions: 

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

667 return properties 

668 

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

670 """Declare and set up extension properties. 

671 

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

673 new `ObservationInfo`. 

674 

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

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

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

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

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

680 extension properties are set to `None`. 

681 

682 Parameters 

683 ---------- 

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

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

686 "ext_" prefix). 

687 """ 

688 if extensions: 

689 for name in extensions: 

690 field_name = "ext_" + name 

691 if not hasattr(self, field_name): 691 ↛ 689line 691 didn't jump to line 689 because the condition on line 691 was always true

692 object.__setattr__(self, field_name, None) 

693 self._extensions = extensions 

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

695 

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

697 """Set attribute. 

698 

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

700 instance has been sealed. 

701 

702 Parameters 

703 ---------- 

704 name : `str` 

705 Name of attribute to set. 

706 value : `typing.Any` 

707 Value to set it to. 

708 """ 

709 if ( 

710 getattr(self, "_sealed", False) 

711 and hasattr(self, "all_properties") 

712 and name in self.all_properties 

713 ): 

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

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

716 

717 @classmethod 

718 def __get_pydantic_json_schema__( 

719 cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler 

720 ) -> JsonSchemaValue: 

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

722 # in serialization mode pydantic would otherwise emit a trivial 

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

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

725 # honored. 

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

727 schema = handler(inner) 

728 schema = handler.resolve_ref_schema(schema) 

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

730 schema["patternProperties"] = { 

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

732 } 

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

734 schema["additionalProperties"] = False 

735 return schema 

736 

737 @model_serializer(mode="wrap") 

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

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

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

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

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

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

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

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

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

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

748 field_name = f"ext_{ext_name}" 

749 value = getattr(self, field_name, None) 

750 if value is None: 750 ↛ 751line 750 didn't jump to line 751 because the condition on line 750 was never true

751 continue 

752 simplifier = ext_def.to_simple 

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

754 return result 

755 

756 @classmethod 

757 def _validate_property_mapping( 

758 cls, 

759 data: MutableMapping[str, Any], 

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

761 ) -> dict[str, Any]: 

762 # Validate extension properties. 

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

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

765 

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

767 if key not in properties: 767 ↛ 768line 767 didn't jump to line 768 because the condition on line 767 was never true

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

769 if value is None: 

770 continue 

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

772 return processed 

773 

774 @classmethod 

775 def _coerce_property_value( 

776 cls, 

777 key: str, 

778 value: Any, 

779 properties: dict[str, PropertyDefinition], 

780 processed: dict[str, Any], 

781 ) -> Any: 

782 definition = properties[key] 

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

784 if not definition.is_value_conformant(converted): 

785 raise TypeError( 

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

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

788 ) 

789 return converted 

790 

791 @classmethod 

792 def _coerce_from_simple( 

793 cls, 

794 definition: PropertyDefinition, 

795 value: Any, 

796 processed: dict[str, Any], 

797 ) -> Any: 

798 if definition.is_value_conformant(value): 

799 return value 

800 complexifier = definition.from_simple 

801 if complexifier is None: 801 ↛ 804line 801 didn't jump to line 804 because the condition on line 801 was always true

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

803 return value 

804 return complexifier(value, **processed) 

805 

806 @property 

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

808 """Header cards used for the translation. 

809 

810 Returns 

811 ------- 

812 used : `frozenset` of `str` 

813 Set of card used. 

814 """ 

815 if not self._translator: 

816 return frozenset() 

817 return self._translator.cards_used() 

818 

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

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

821 

822 Returns 

823 ------- 

824 stripped : `dict`-like 

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

826 headers used to calculate the generic information removed. 

827 """ 

828 hdr = copy.copy(self._header) 

829 used = self.cards_used 

830 for c in used: 

831 if c in hdr: 

832 del hdr[c] 

833 return hdr 

834 

835 def __str__(self) -> str: 

836 # Put more interesting answers at front of list 

837 # and then do remainder 

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

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