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 Generator, MutableMapping, Sequence 

22from contextlib import contextmanager 

23from typing import Any, cast, overload 

24 

25import astropy.coordinates 

26import astropy.time 

27import astropy.units 

28import numpy as np 

29from lsst.resources import ResourcePath 

30from pydantic import ( 

31 BaseModel, 

32 ConfigDict, 

33 Field, 

34 GetJsonSchemaHandler, 

35 PrivateAttr, 

36 model_serializer, 

37) 

38from pydantic.json_schema import JsonSchemaValue 

39from pydantic_core import CoreSchema 

40 

41from .headers import fix_header 

42from .properties import ( 

43 PROPERTIES, 

44 AltAzAnnotated, 

45 AngleAnnotated, 

46 EarthLocationAnnotated, 

47 ExposureTimeAnnotated, 

48 FocusZAnnotated, 

49 PressureAnnotated, 

50 PropertyDefinition, 

51 SkyCoordAnnotated, 

52 TemperatureAnnotated, 

53 TimeAnnotated, 

54 TimeDeltaAnnotated, 

55) 

56from .translator import MetadataTranslator 

57 

58log = logging.getLogger(__name__) 

59 

60 

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

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

63 

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

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

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

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

68 misleading. 

69 

70 Parameters 

71 ---------- 

72 key : `str` 

73 Property name in `PROPERTIES`. 

74 wire_form : `str` 

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

76 ``"a float in meters"``. 

77 

78 Returns 

79 ------- 

80 description : `str` 

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

82 """ 

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

84 

85 

86class ObservationInfo(BaseModel): 

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

88 exposure observation. 

89 

90 Parameters 

91 ---------- 

92 header : `dict`-like 

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

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

95 filename : `str`, optional 

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

97 datasets with missing header information this can sometimes 

98 allow for some fixups in translations. 

99 translator_class : `MetadataTranslator`-class, optional 

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

101 into standard form. Otherwise each registered translator class will 

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

103 pedantic : `bool`, optional 

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

105 individual property translations must all be implemented but can fail 

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

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

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

109 ``header`` is specified. 

110 required : `set`, optional 

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

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

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

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

115 subset : `set`, optional 

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

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

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

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

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

121 quiet : `bool`, optional 

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

123 pedantic mode are converted to debug messages. 

124 **kwargs : `typing.Any` 

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

126 mode creates an `ObservationInfo` directly from supplied properties 

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

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

129 

130 Raises 

131 ------ 

132 ValueError 

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

134 registered translators. Also raised if the request property subset 

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

136 with kwargs. 

137 TypeError 

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

139 KeyError 

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

141 mode is enabled and any translations fails. 

142 NotImplementedError 

143 Raised if the selected translator does not support a required 

144 property. 

145 

146 Notes 

147 ----- 

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

149 Additional properties may be defined, either through the 

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

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

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

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

154 

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

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

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

158 remaining parameters control the behavior of the translator. 

159 

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

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

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

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

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

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

166 from which to extract extension definitions. 

167 

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

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

170 header after construction will modify the internal cached header. 

171 

172 Values of the properties are read-only. 

173 """ 

174 

175 model_config = ConfigDict( 

176 extra="forbid", 

177 validate_assignment=False, 

178 populate_by_name=True, 

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

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

181 ) 

182 

183 translator_name: str | None = Field( 

184 default=None, 

185 alias="_translator", 

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

187 ) 

188 

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

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

191 location: EarthLocationAnnotated | None = Field( 

192 default=None, 

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

194 ) 

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

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

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

198 datetime_begin: TimeAnnotated | None = Field( 

199 default=None, 

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

201 ) 

202 datetime_end: TimeAnnotated | None = Field( 

203 default=None, 

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

205 ) 

206 exposure_time: ExposureTimeAnnotated | None = Field( 

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

208 ) 

209 exposure_time_requested: ExposureTimeAnnotated | None = Field( 

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

211 ) 

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

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

214 boresight_rotation_angle: AngleAnnotated | None = Field( 

215 default=None, 

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

217 ) 

218 boresight_rotation_coord: str | None = Field( 

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

220 ) 

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

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

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

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

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

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

227 focus_z: FocusZAnnotated | None = Field( 

228 default=None, 

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

230 ) 

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

232 temperature: TemperatureAnnotated | None = Field( 

233 default=None, 

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

235 ) 

236 pressure: PressureAnnotated | None = Field( 

237 default=None, 

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

239 ) 

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

241 tracking_radec: SkyCoordAnnotated | None = Field( 

242 default=None, 

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

244 ) 

245 altaz_begin: AltAzAnnotated | None = Field( 

246 default=None, 

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

248 ) 

249 altaz_end: AltAzAnnotated | None = Field( 

250 default=None, 

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

252 ) 

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

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

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

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

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

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

259 observing_day_offset: TimeDeltaAnnotated | None = Field( 

260 default=None, 

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

262 ) 

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

264 has_simulated_content: bool | None = Field( 

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

266 ) 

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

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

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

270 

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

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

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

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

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

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

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

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

279 _sealed: bool = PrivateAttr(default=False) 

280 

281 @property 

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

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

284 return self._filename 

285 

286 @filename.setter 

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

288 self._filename = value 

289 

290 @property 

291 def translator_class_name(self) -> str: 

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

293 return self._translator_class_name 

294 

295 @property 

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

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

298 return self._extensions 

299 

300 @property 

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

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

303 return self._all_properties 

304 

305 @overload 

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

307 self, 

308 header: MutableMapping[str, Any], 

309 filename: str | ResourcePath | None = None, 

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

311 pedantic: bool = False, 

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

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

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

315 quiet: bool = False, 

316 ) -> None: ... 

317 

318 @overload 

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

320 self, 

321 header: None = None, 

322 filename: str | ResourcePath | None = None, 

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

324 **kwargs: Any, 

325 ) -> None: ... 

326 

327 def __init__( 

328 self, 

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

330 filename: str | ResourcePath | None = None, 

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

332 pedantic: bool = False, 

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

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

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

336 quiet: bool = False, 

337 **kwargs: Any, 

338 ) -> None: 

339 if filename is not None: 

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

341 if header is not None: 

342 if kwargs: 

343 raise ValueError( 

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

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

346 ) 

347 self._init_from_header( 

348 header, 

349 filename=filename, 

350 translator_class=translator_class, 

351 pedantic=pedantic, 

352 search_path=search_path, 

353 required=required, 

354 subset=subset, 

355 quiet=quiet, 

356 ) 

357 return 

358 

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

360 

361 def _init_from_header( 

362 self, 

363 header: MutableMapping[str, Any], 

364 *, 

365 filename: str | None, 

366 translator_class: type[MetadataTranslator] | None, 

367 pedantic: bool, 

368 search_path: Sequence[str] | None, 

369 required: set[str] | None, 

370 subset: set[str] | None, 

371 quiet: bool, 

372 ) -> None: 

373 super().__init__() 

374 self._filename = filename 

375 self._sealed = False 

376 # Initialize the empty object 

377 self._header = {} 

378 self._translator = None 

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

380 

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

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

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

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

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

386 if translator_class is None: 

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

388 elif not issubclass(translator_class, MetadataTranslator): 

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

390 

391 # Fix up the header (if required) 

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

393 

394 # Store the supplied header for later stripping 

395 self._header = header 

396 

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

398 self._declare_extensions(translator_class.extensions) 

399 

400 # Create an instance for this header 

401 translator = translator_class(header, filename=filename) 

402 

403 # Store the translator 

404 self._translator = translator 

405 self._translator_class_name = translator_class.__name__ 

406 self.translator_name = translator_class.name 

407 

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

409 if filename: 

410 file_info = f" and file {filename}" 

411 else: 

412 file_info = "" 

413 

414 # Determine the properties of interest 

415 full_set = set(self.all_properties) 

416 if subset is not None: 

417 if not subset: 

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

419 if not subset.issubset(full_set): 

420 raise ValueError( 

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

422 ) 

423 properties = subset 

424 else: 

425 properties = full_set 

426 

427 if required is None: 

428 required = set() 

429 else: 

430 if not required.issubset(full_set): 

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

432 

433 # Loop over each property and request the translated form 

434 for property in properties: 

435 # prototype code 

436 method = f"to_{property}" 

437 

438 try: 

439 value = getattr(translator, method)() 

440 except NotImplementedError as e: 

441 raise NotImplementedError( 

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

443 ) from e 

444 except Exception as e: 

445 err_msg = ( 

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

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

448 ) 

449 if pedantic or property in required: 

450 raise KeyError(err_msg) from e 

451 else: 

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

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

454 continue 

455 

456 definition = self.all_properties[property] 

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

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

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

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

461 # until pydantic turned up. 

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

463 if not definition.is_value_conformant(value): 463 ↛ 464line 463 didn't jump to line 464 because the condition on line 463 was never true

464 err_msg = ( 

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

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

467 f"{file_info}" 

468 ) 

469 if pedantic or property in required: 

470 raise TypeError(err_msg) 

471 else: 

472 log.debug( 

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

474 ) 

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

476 

477 if value is None and property in required: 

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

479 

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

481 

482 self._sealed = True 

483 

484 def _init_from_kwargs( 

485 self, 

486 *, 

487 filename: str | None, 

488 translator_class: type[MetadataTranslator] | None, 

489 **kwargs: Any, 

490 ) -> None: 

491 supplied_keys = set(kwargs) 

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

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

494 # field name directly. 

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

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

497 if translator_name is not None and translator_name in MetadataTranslator.translators: 

498 translator_class = MetadataTranslator.translators[translator_name] 

499 

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

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

502 

503 if supplied_extensions is not None: 

504 if translator_class is not None: 

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

506 if not isinstance(supplied_extensions, dict): 

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

508 extensions = supplied_extensions 

509 else: 

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

511 

512 all_properties = self._get_all_properties(extensions) 

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

514 if unrecognized: 

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

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

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

518 # underlying cause rather than an opaque unknown property. 

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

520 if extension_keys and translator_class is None: 

521 reason = ( 

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

523 if translator_name is not None 

524 else "no translator was specified" 

525 ) 

526 raise KeyError( 

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

528 "extension property definitions are unavailable." 

529 ) 

530 text = ( 

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

532 if len(unrecognized) == 1 

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

534 ) 

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

536 

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

538 processed = self._apply_constructor_defaults(processed, supplied_keys) 

539 

540 super().__init__(**processed) 

541 self._filename = filename 

542 self._sealed = False 

543 

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

545 self._declare_extensions(extensions) 

546 

547 # Handle extensions. 

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

549 processed_ext = self._validate_property_mapping(ext_input, extensions) 

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

551 object.__setattr__(self, key, value) 

552 

553 if translator_class is not None: 

554 self._translator = translator_class({}) 

555 self._translator_class_name = translator_class.__name__ 

556 self.translator_name = translator_class.name 

557 elif translator_name is not None: 

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

559 # translator instance is only needed while constructing from a 

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

561 # is recorded directly. 

562 self._translator_class_name = translator_name 

563 self.translator_name = translator_name 

564 

565 self._sealed = True 

566 

567 @staticmethod 

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

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

570 

571 Parameters 

572 ---------- 

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

574 Properties validated from kwargs input. 

575 supplied_keys : `set` [`str`] 

576 Property names explicitly supplied by the caller. 

577 

578 Returns 

579 ------- 

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

581 Updated property mapping with defaults/backfilled values applied. 

582 """ 

583 updated = dict(processed) 

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

585 if ( 

586 key not in supplied_keys 

587 and "observation_counter" in supplied_keys 

588 and "observation_counter" in updated 

589 ): 

590 updated[key] = updated["observation_counter"] 

591 if "has_simulated_content" not in supplied_keys: 

592 updated["has_simulated_content"] = False 

593 return updated 

594 

595 @classmethod 

596 def from_header( 

597 cls, 

598 header: MutableMapping[str, Any], 

599 *, 

600 filename: str | None = None, 

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

602 pedantic: bool = False, 

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

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

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

606 quiet: bool = False, 

607 ) -> ObservationInfo: 

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

609 

610 Parameters 

611 ---------- 

612 header : `dict`-like 

613 Header mapping to translate. 

614 filename : `str`, optional 

615 Name of file associated with this header. 

616 translator_class : `MetadataTranslator`-class, optional 

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

618 auto-determined. 

619 pedantic : `bool`, optional 

620 If `True`, translation failures are fatal. 

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

622 Search paths for header corrections. 

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

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

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

626 Restrict translation to this subset of properties. 

627 quiet : `bool`, optional 

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

629 pedantic mode are converted to debug messages. 

630 

631 Returns 

632 ------- 

633 obsinfo : `ObservationInfo` 

634 Translated observation metadata. 

635 """ 

636 return cls( 

637 header=header, 

638 filename=filename, 

639 translator_class=translator_class, 

640 pedantic=pedantic, 

641 search_path=search_path, 

642 required=required, 

643 subset=subset, 

644 quiet=quiet, 

645 ) 

646 

647 @staticmethod 

648 def _get_all_properties( 

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

650 ) -> dict[str, PropertyDefinition]: 

651 """Return the definitions of all properties. 

652 

653 Parameters 

654 ---------- 

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

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

657 "ext_" prefix). 

658 

659 Returns 

660 ------- 

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

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

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

664 """ 

665 properties = dict(PROPERTIES) 

666 if extensions: 

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

668 return properties 

669 

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

671 """Declare and set up extension properties. 

672 

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

674 new `ObservationInfo`. 

675 

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

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

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

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

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

681 extension properties are set to `None`. 

682 

683 Parameters 

684 ---------- 

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

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

687 "ext_" prefix). 

688 """ 

689 if extensions: 

690 for name in extensions: 

691 field_name = "ext_" + name 

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

693 object.__setattr__(self, field_name, None) 

694 self._extensions = extensions 

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

696 

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

698 """Set attribute. 

699 

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

701 instance has been sealed. 

702 

703 Parameters 

704 ---------- 

705 name : `str` 

706 Name of attribute to set. 

707 value : `typing.Any` 

708 Value to set it to. 

709 """ 

710 if ( 

711 getattr(self, "_sealed", False) 

712 and hasattr(self, "all_properties") 

713 and name in self.all_properties 

714 ): 

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

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

717 

718 @classmethod 

719 def __get_pydantic_json_schema__( 

720 cls, core_schema: CoreSchema, handler: GetJsonSchemaHandler 

721 ) -> JsonSchemaValue: 

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

723 # in serialization mode pydantic would otherwise emit a trivial 

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

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

726 # honored. 

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

728 schema = handler(inner) 

729 schema = handler.resolve_ref_schema(schema) 

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

731 schema["patternProperties"] = { 

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

733 } 

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

735 schema["additionalProperties"] = False 

736 return schema 

737 

738 @model_serializer(mode="wrap") 

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

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

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

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

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

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

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

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

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

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

749 field_name = f"ext_{ext_name}" 

750 value = getattr(self, field_name, None) 

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

752 continue 

753 simplifier = ext_def.to_simple 

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

755 return result 

756 

757 @classmethod 

758 def _validate_property_mapping( 

759 cls, 

760 data: MutableMapping[str, Any], 

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

762 ) -> dict[str, Any]: 

763 # Validate extension properties. 

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

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

766 

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

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

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

770 if value is None: 

771 continue 

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

773 return processed 

774 

775 @classmethod 

776 def _coerce_property_value( 

777 cls, 

778 key: str, 

779 value: Any, 

780 properties: dict[str, PropertyDefinition], 

781 processed: dict[str, Any], 

782 ) -> Any: 

783 definition = properties[key] 

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

785 if not definition.is_value_conformant(converted): 

786 raise TypeError( 

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

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

789 ) 

790 return converted 

791 

792 @classmethod 

793 def _coerce_from_simple( 

794 cls, 

795 definition: PropertyDefinition, 

796 value: Any, 

797 processed: dict[str, Any], 

798 ) -> Any: 

799 if definition.is_value_conformant(value): 

800 return value 

801 complexifier = definition.from_simple 

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

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

804 return value 

805 return complexifier(value, **processed) 

806 

807 @property 

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

809 """Header cards used for the translation. 

810 

811 Returns 

812 ------- 

813 used : `frozenset` of `str` 

814 Set of card used. 

815 """ 

816 if not self._translator: 

817 return frozenset() 

818 return self._translator.cards_used() 

819 

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

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

822 

823 Returns 

824 ------- 

825 stripped : `dict`-like 

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

827 headers used to calculate the generic information removed. 

828 """ 

829 hdr = copy.copy(self._header) 

830 used = self.cards_used 

831 for c in used: 

832 if c in hdr: 

833 del hdr[c] 

834 return hdr 

835 

836 def __str__(self) -> str: 

837 # Put more interesting answers at front of list 

838 # and then do remainder 

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