Coverage for tests/test_flat_dict.py: 96%

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"""Tests for the flat dict serialization of ObservationInfo.""" 

13 

14import math 

15import os 

16import tempfile 

17import unittest 

18from collections.abc import Mapping 

19from typing import Any 

20 

21import astropy.time 

22import astropy.units as u 

23from astropy.coordinates import EarthLocation 

24from astropy.io import fits 

25 

26from astro_metadata_translator import ( 

27 FITS_FLAT_DICT_PREFIX, 

28 FITS_FLAT_DICT_SEP, 

29 ObservationInfo, 

30 PropertyDefinition, 

31 StubTranslator, 

32 makeObservationInfo, 

33) 

34from astro_metadata_translator.serialize.flat_dict import ( 

35 detect_prefix_and_sep, 

36 find_flat_keys, 

37 find_translator_name, 

38 flatten_simple, 

39 make_flat_key, 

40 unflatten, 

41 validate_sep, 

42) 

43from astro_metadata_translator.tests import read_test_file 

44 

45try: 

46 from lsst.daf.base import PropertyList 

47except ImportError: 

48 PropertyList = None 

49 

50try: 

51 import lsst.afw.fits as afw_fits 

52 import lsst.afw.image as afw_image 

53except ImportError: 

54 afw_fits = afw_image = None 

55 

56TESTDIR = os.path.abspath(os.path.dirname(__file__)) 

57DATADIR = os.path.join(TESTDIR, "data") 

58 

59HEADER_FILES = ( 

60 "fitsheader-decam.yaml", 

61 "fitsheader-hsc.yaml", 

62 "fitsheader-megaprime.yaml", 

63 "fitsheader-sdss-fpC-006377-g4-0399.yaml", 

64 "fitsheader-suprimecam-CORR40535770.yaml", 

65) 

66 

67PREFIX_FORMS = (("", "."), ("obs_info", "."), (FITS_FLAT_DICT_PREFIX, FITS_FLAT_DICT_SEP)) 

68 

69 

70# Registers itself through StubTranslator.__init_subclass__ so that 

71# ``_translator: "flatdict"`` resolves and its extensions are available. 

72class _FlatDictTranslator(StubTranslator): 

73 name = "flatdict" 

74 supported_instrument = "flatdict" 

75 extensions = dict( 

76 number=PropertyDefinition("A number", int), 

77 foo=PropertyDefinition("A string", str), 

78 pair=PropertyDefinition("A pair of floats", tuple), 

79 ) 

80 

81 @classmethod 

82 def can_translate(cls, header: Mapping[str, Any], filename: str | None = None) -> bool: 

83 return False 

84 

85 

86def _read_obsinfo(filename: str) -> ObservationInfo: 

87 return ObservationInfo(read_test_file(filename, dir=DATADIR)) 

88 

89 

90class FlattenTestCase(unittest.TestCase): 

91 """Test flattening of the simple form.""" 

92 

93 def test_constants(self) -> None: 

94 self.assertEqual(FITS_FLAT_DICT_PREFIX, "OBS_INFO") 

95 self.assertEqual(FITS_FLAT_DICT_SEP, " ") 

96 

97 def test_validate_sep(self) -> None: 

98 for good in (".", " ", "-", ":"): 

99 validate_sep(good) 

100 for bad in ("", "..", "a", "Z", "1", "_"): 

101 with self.assertRaises(ValueError): 

102 validate_sep(bad) 

103 

104 def test_make_flat_key(self) -> None: 

105 self.assertEqual(make_flat_key("instrument", prefix="", sep=".", use_upper=False), "instrument") 

106 self.assertEqual(make_flat_key("instrument", prefix="", sep=".", use_upper=True), "INSTRUMENT") 

107 self.assertEqual( 

108 make_flat_key("datetime_begin_0", prefix="obs_info", sep=".", use_upper=False), 

109 "obs_info.datetime_begin_0", 

110 ) 

111 self.assertEqual( 

112 make_flat_key("datetime_begin_0", prefix="OBS_INFO", sep=" ", use_upper=True), 

113 "OBS_INFO DATETIME_BEGIN_0", 

114 ) 

115 # The prefix is emitted exactly as given. 

116 self.assertEqual( 

117 make_flat_key("instrument", prefix="obs_info", sep=" ", use_upper=True), 

118 "obs_info INSTRUMENT", 

119 ) 

120 

121 def test_flatten_scalars_no_prefix(self) -> None: 

122 simple = { 

123 "instrument": "HSC", 

124 "exposure_id": 42, 

125 "boresight_airmass": 1.2, 

126 "can_see_sky": True, 

127 "_translator": "HSC", 

128 } 

129 self.assertEqual( 

130 flatten_simple(simple), 

131 { 

132 "instrument": "HSC", 

133 "exposure_id": 42, 

134 "boresight_airmass": 1.2, 

135 "can_see_sky": True, 

136 "translator": "HSC", 

137 }, 

138 ) 

139 

140 def test_flatten_sequences(self) -> None: 

141 simple = {"datetime_begin": (2460000.5, 0.25), "location": [1.0, 2.0, 3.0], "ext_pair": (1, 2)} 

142 self.assertEqual( 

143 flatten_simple(simple), 

144 { 

145 "datetime_begin_0": 2460000.5, 

146 "datetime_begin_1": 0.25, 

147 "location_0": 1.0, 

148 "location_1": 2.0, 

149 "location_2": 3.0, 

150 "ext_pair_0": 1, 

151 "ext_pair_1": 2, 

152 }, 

153 ) 

154 

155 def test_flatten_non_finite(self) -> None: 

156 simple = { 

157 "temperature": math.inf, 

158 "pressure": -math.inf, 

159 "relative_humidity": math.nan, 

160 "location": (math.nan, 1.0, math.inf), 

161 "detector_num": 1, 

162 "object": "NAN", 

163 } 

164 # By default non-finite floats pass through untouched. 

165 flat = flatten_simple(simple) 

166 self.assertTrue(math.isnan(flat["relative_humidity"])) 

167 self.assertEqual(flat["temperature"], math.inf) 

168 self.assertTrue(math.isnan(flat["location_0"])) 

169 

170 flat = flatten_simple(simple, encode_non_finite=True) 

171 self.assertEqual(flat["temperature"], "+INFINITY") 

172 self.assertEqual(flat["pressure"], "-INFINITY") 

173 self.assertEqual(flat["relative_humidity"], "NAN") 

174 self.assertEqual(flat["location_0"], "NAN") 

175 self.assertEqual(flat["location_1"], 1.0) 

176 self.assertEqual(flat["location_2"], "+INFINITY") 

177 self.assertEqual(flat["detector_num"], 1) 

178 self.assertEqual(flat["object"], "NAN") 

179 

180 def test_flatten_prefix_and_case(self) -> None: 

181 simple = {"datetime_begin": (1.0, 2.0), "instrument": "X"} 

182 self.assertEqual( 

183 flatten_simple(simple, prefix="obs_info"), 

184 { 

185 "obs_info.datetime_begin_0": 1.0, 

186 "obs_info.datetime_begin_1": 2.0, 

187 "obs_info.instrument": "X", 

188 }, 

189 ) 

190 self.assertEqual( 

191 flatten_simple(simple, prefix="OBS_INFO", sep=" "), 

192 { 

193 "OBS_INFO DATETIME_BEGIN_0": 1.0, 

194 "OBS_INFO DATETIME_BEGIN_1": 2.0, 

195 "OBS_INFO INSTRUMENT": "X", 

196 }, 

197 ) 

198 self.assertEqual( 

199 flatten_simple(simple, prefix="OBS_INFO", sep=" ", use_upper=False), 

200 { 

201 "OBS_INFO datetime_begin_0": 1.0, 

202 "OBS_INFO datetime_begin_1": 2.0, 

203 "OBS_INFO instrument": "X", 

204 }, 

205 ) 

206 self.assertEqual( 

207 flatten_simple(simple, use_upper=True), 

208 {"DATETIME_BEGIN_0": 1.0, "DATETIME_BEGIN_1": 2.0, "INSTRUMENT": "X"}, 

209 ) 

210 

211 def test_flatten_rejects_bad_values(self) -> None: 

212 with self.assertRaises(TypeError): 

213 flatten_simple({"ext_bad": {"a": 1}}) 

214 with self.assertRaises(TypeError): 

215 flatten_simple({"ext_nested": [[1.0, 2.0]]}) 

216 with self.assertRaises(TypeError): 

217 flatten_simple({"ext_none": None}) 

218 

219 def test_flatten_rejects_bad_sep(self) -> None: 

220 with self.assertRaises(ValueError): 

221 flatten_simple({"instrument": "X"}, prefix="p", sep="_") 

222 with self.assertRaises(ValueError): 

223 flatten_simple({"instrument": "X"}, sep="ab") 

224 

225 

226class DetectTestCase(unittest.TestCase): 

227 """Test prefix detection and key collection.""" 

228 

229 PREFIXED = { 

230 "OBS_INFO DATETIME_BEGIN_0": 2460000.5, 

231 "OBS_INFO DATETIME_BEGIN_1": 0.25, 

232 "OBS_INFO INSTRUMENT": "HSC", 

233 "OBS_INFO OBJECT": "real", 

234 "OBS_INFO TRANSLATOR": "HSC", 

235 "OBS_INFO EXT_FOO": "bar", 

236 "OBS_INFO EXT_PAIR_0": 1, 

237 "OBJECT": "decoy", 

238 "TEMPERATURE": 3.0, 

239 "EXPTIME": 30.0, 

240 } 

241 

242 def test_detect_nothing(self) -> None: 

243 self.assertIsNone(detect_prefix_and_sep({})) 

244 self.assertIsNone(detect_prefix_and_sep({"EXPTIME": 30.0, "FILTER": "g"})) 

245 

246 def test_detect_decoys_without_anchor(self) -> None: 

247 # Core names appear but nothing indexed, so this is not content. 

248 self.assertIsNone(detect_prefix_and_sep({"OBJECT": "M31", "TEMPERATURE": 3.0})) 

249 self.assertIsNone(detect_prefix_and_sep({"instrument": "X", "object": "Y"})) 

250 

251 def test_detect_anchor_names(self) -> None: 

252 # By default any indexed core name anchors detection. 

253 flat = {"OBJECT": "M31", "OBJECT_1": "x", "TEMPERATURE": 3.0} 

254 self.assertEqual(detect_prefix_and_sep(flat), ("", "")) 

255 # With explicit anchor names a stray indexed scalar does not. 

256 self.assertIsNone(detect_prefix_and_sep(flat, anchor_names=("datetime_begin", "location"))) 

257 self.assertEqual(find_flat_keys(flat, anchor_names=("datetime_begin",)), {}) 

258 flat["datetime_begin_0"] = 1.0 

259 flat["datetime_begin_1"] = 2.0 

260 self.assertEqual(detect_prefix_and_sep(flat, anchor_names=("datetime_begin",)), ("", "")) 

261 

262 def test_detect_plain(self) -> None: 

263 flat = {"datetime_begin_0": 1.0, "datetime_begin_1": 2.0, "instrument": "X"} 

264 self.assertEqual(detect_prefix_and_sep(flat), ("", "")) 

265 

266 def test_detect_prefixed_with_decoys(self) -> None: 

267 self.assertEqual(detect_prefix_and_sep(self.PREFIXED), ("obs_info", " ")) 

268 

269 def test_detect_prefix_containing_sep(self) -> None: 

270 flat = {"LSST OBS_INFO DATETIME_BEGIN_0": 1.0, "LSST OBS_INFO EXPOSURE_ID": 5} 

271 self.assertEqual(detect_prefix_and_sep(flat), ("lsst obs_info", " ")) 

272 

273 def test_detect_dotted(self) -> None: 

274 flat = {"lsst.obs_info.datetime_begin_0": 1.0, "lsst.obs_info.exposure_id": 5} 

275 self.assertEqual(detect_prefix_and_sep(flat), ("lsst.obs_info", ".")) 

276 

277 def test_detect_majority(self) -> None: 

278 flat = {"a.datetime_begin_0": 1.0, "a.instrument": "x", "b.datetime_begin_0": 1.0} 

279 self.assertEqual(detect_prefix_and_sep(flat), ("a", ".")) 

280 

281 def test_detect_tie(self) -> None: 

282 flat = {"a.datetime_begin_0": 1.0, "b.datetime_begin_0": 1.0} 

283 with self.assertRaises(ValueError): 

284 detect_prefix_and_sep(flat) 

285 

286 def test_find_keys_auto(self) -> None: 

287 found = find_flat_keys(self.PREFIXED) 

288 self.assertEqual( 

289 found, 

290 { 

291 "OBS_INFO DATETIME_BEGIN_0": "datetime_begin_0", 

292 "OBS_INFO DATETIME_BEGIN_1": "datetime_begin_1", 

293 "OBS_INFO INSTRUMENT": "instrument", 

294 "OBS_INFO OBJECT": "object", 

295 "OBS_INFO TRANSLATOR": "translator", 

296 "OBS_INFO EXT_FOO": "ext_foo", 

297 "OBS_INFO EXT_PAIR_0": "ext_pair_0", 

298 }, 

299 ) 

300 

301 def test_find_keys_auto_nothing(self) -> None: 

302 self.assertEqual(find_flat_keys({"OBJECT": "M31"}), {}) 

303 

304 def test_find_keys_explicit(self) -> None: 

305 found = find_flat_keys(self.PREFIXED, prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

306 self.assertEqual(len(found), 7) 

307 self.assertNotIn("OBJECT", found) 

308 # Explicit empty prefix needs no indexed anchor. 

309 self.assertEqual( 

310 find_flat_keys({"instrument": "X", "EXPTIME": 1}, prefix=""), {"instrument": "instrument"} 

311 ) 

312 # Explicit prefix with the wrong separator finds nothing. 

313 self.assertEqual(find_flat_keys(self.PREFIXED, prefix="OBS_INFO", sep="."), {}) 

314 # Explicit prefix is matched case-insensitively. 

315 found = find_flat_keys(self.PREFIXED, prefix="obs_info", sep=" ") 

316 self.assertEqual(len(found), 7) 

317 

318 def test_find_keys_ignores_non_string_keys(self) -> None: 

319 flat: dict[Any, Any] = {1: "x", "datetime_begin_0": 1.0, "datetime_begin_1": 2.0} 

320 expected = {"datetime_begin_0": "datetime_begin_0", "datetime_begin_1": "datetime_begin_1"} 

321 self.assertEqual(find_flat_keys(flat), expected) 

322 self.assertEqual(find_flat_keys(flat, prefix=""), expected) 

323 

324 def test_find_keys_ignores_unknown_names(self) -> None: 

325 flat = {"obs_info.datetime_begin_0": 1.0, "obs_info.not_a_property": 2, "obs_info.ext_": 3} 

326 self.assertEqual( 

327 find_flat_keys(flat, prefix="obs_info"), {"obs_info.datetime_begin_0": "datetime_begin_0"} 

328 ) 

329 

330 def test_find_keys_duplicate(self) -> None: 

331 flat = {"OBS_INFO INSTRUMENT": "a", "obs_info instrument": "b", "OBS_INFO DATETIME_BEGIN_0": 1.0} 

332 with self.assertRaises(ValueError): 

333 find_flat_keys(flat, prefix="OBS_INFO", sep=" ") 

334 

335 def test_find_keys_duplicates_allowed(self) -> None: 

336 flat = {"OBS_INFO INSTRUMENT": "a", "obs_info instrument": "b", "OBS_INFO DATETIME_BEGIN_0": 1.0} 

337 found = find_flat_keys(flat, prefix="OBS_INFO", sep=" ", unique=False) 

338 self.assertEqual(set(found), set(flat)) 

339 self.assertEqual(found["obs_info instrument"], "instrument") 

340 

341 def test_find_keys_bad_sep(self) -> None: 

342 with self.assertRaises(ValueError): 

343 find_flat_keys({"p_instrument": "x"}, prefix="p", sep="_") 

344 

345 

346class UnflattenTestCase(unittest.TestCase): 

347 """Test rebuilding the simple form from flat keys.""" 

348 

349 def test_unflatten_core(self) -> None: 

350 flat = { 

351 "OBS_INFO DATETIME_BEGIN_1": 0.25, 

352 "OBS_INFO DATETIME_BEGIN_0": 2460000.5, 

353 "OBS_INFO LOCATION_0": 1.0, 

354 "OBS_INFO LOCATION_1": 2.0, 

355 "OBS_INFO LOCATION_2": 3.0, 

356 "OBS_INFO INSTRUMENT": "HSC", 

357 "OBS_INFO TRANSLATOR": "HSC", 

358 "OBJECT": "decoy", 

359 } 

360 flat_keys = find_flat_keys(flat) 

361 self.assertEqual(find_translator_name(flat, flat_keys), "HSC") 

362 self.assertEqual( 

363 unflatten(flat, flat_keys), 

364 { 

365 "datetime_begin": (2460000.5, 0.25), 

366 "location": (1.0, 2.0, 3.0), 

367 "instrument": "HSC", 

368 "_translator": "HSC", 

369 }, 

370 ) 

371 

372 def test_unflatten_no_translator(self) -> None: 

373 flat = {"datetime_begin_0": 1.0, "datetime_begin_1": 2.0} 

374 flat_keys = find_flat_keys(flat) 

375 self.assertIsNone(find_translator_name(flat, flat_keys)) 

376 self.assertEqual(unflatten(flat, flat_keys), {"datetime_begin": (1.0, 2.0)}) 

377 

378 def test_unflatten_gap(self) -> None: 

379 flat = {"datetime_begin_0": 1.0, "datetime_begin_2": 2.0} 

380 with self.assertRaises(ValueError): 

381 unflatten(flat, find_flat_keys(flat)) 

382 flat = {"datetime_begin_1": 2.0} 

383 with self.assertRaises(ValueError): 

384 unflatten(flat, find_flat_keys(flat)) 

385 

386 def test_unflatten_expected_lengths(self) -> None: 

387 flat = {"datetime_begin_0": 1.0, "datetime_begin_1": 2.0, "location_0": 1.0, "location_1": 2.0} 

388 flat_keys = find_flat_keys(flat) 

389 lengths = {"datetime_begin": 2, "location": 3} 

390 # A truncated tuple is incomplete even though its indexes are 

391 # contiguous from 0. 

392 with self.assertRaises(ValueError): 

393 unflatten(flat, flat_keys, expected_lengths=lengths) 

394 # Without a declared length only contiguity is checked. 

395 self.assertEqual(unflatten(flat, flat_keys), {"datetime_begin": (1.0, 2.0), "location": (1.0, 2.0)}) 

396 del flat["location_0"], flat["location_1"] 

397 self.assertEqual( 

398 unflatten(flat, find_flat_keys(flat), expected_lengths=lengths), {"datetime_begin": (1.0, 2.0)} 

399 ) 

400 

401 def test_unflatten_scalar_and_indexed_conflict(self) -> None: 

402 flat = {"datetime_begin": 1.0, "datetime_begin_0": 1.0, "datetime_begin_1": 2.0} 

403 with self.assertRaises(ValueError): 

404 unflatten(flat, find_flat_keys(flat)) 

405 

406 def test_unflatten_extensions(self) -> None: 

407 flat = { 

408 "obs_info.datetime_begin_0": 1.0, 

409 "obs_info.datetime_begin_1": 2.0, 

410 "obs_info.ext_foo": "bar", 

411 "obs_info.ext_pair_0": 1, 

412 "obs_info.ext_pair_1": 2, 

413 "obs_info.ext_other_1": 3, 

414 } 

415 flat_keys = find_flat_keys(flat, prefix="obs_info") 

416 # Known extension names restore their spelling and regroup. 

417 simple = unflatten(flat, flat_keys, extension_names=("Foo", "pair")) 

418 self.assertEqual( 

419 simple, 

420 { 

421 "datetime_begin": (1.0, 2.0), 

422 "ext_Foo": "bar", 

423 "ext_pair": (1, 2), 

424 "ext_other_1": 3, 

425 }, 

426 ) 

427 # Unknown extension names pass through as plain lower-case names. 

428 simple = unflatten(flat, flat_keys) 

429 self.assertEqual( 

430 simple, 

431 { 

432 "datetime_begin": (1.0, 2.0), 

433 "ext_foo": "bar", 

434 "ext_pair_0": 1, 

435 "ext_pair_1": 2, 

436 "ext_other_1": 3, 

437 }, 

438 ) 

439 

440 def test_unflatten_float_names(self) -> None: 

441 flat = { 

442 "temperature": "+INFINITY", 

443 "pressure": "-Infinity", 

444 "relative_humidity": "NAN", 

445 "boresight_airmass": "nan", 

446 "location_0": "NAN", 

447 "location_1": 1.0, 

448 "location_2": "+INFINITY", 

449 "object": "NAN", 

450 "ext_foo": "NAN", 

451 } 

452 flat_keys = find_flat_keys(flat, prefix="") 

453 float_names = { 

454 "temperature", 

455 "pressure", 

456 "relative_humidity", 

457 "boresight_airmass", 

458 "location", 

459 "ext_Foo", 

460 } 

461 simple = unflatten(flat, flat_keys, extension_names=("Foo",), float_names=float_names) 

462 self.assertEqual(simple["temperature"], math.inf) 

463 self.assertEqual(simple["pressure"], -math.inf) 

464 self.assertTrue(math.isnan(simple["relative_humidity"])) 

465 self.assertTrue(math.isnan(simple["boresight_airmass"])) 

466 self.assertTrue(math.isnan(simple["location"][0])) 

467 self.assertEqual(simple["location"][1:], (1.0, math.inf)) 

468 self.assertEqual(simple["object"], "NAN") 

469 self.assertTrue(math.isnan(simple["ext_Foo"])) 

470 

471 # Without float names the strings are left alone. 

472 simple = unflatten(flat, flat_keys) 

473 self.assertEqual(simple["temperature"], "+INFINITY") 

474 

475 # A string that is not a number is an error for a float property. 

476 with self.assertRaises(ValueError): 

477 unflatten({"temperature": "warm"}, {"temperature": "temperature"}, float_names={"temperature"}) 

478 

479 

480class ObservationInfoFlatDictTestCase(unittest.TestCase): 

481 """Test the flat dict methods on ObservationInfo.""" 

482 

483 def assert_round_trip(self, original: ObservationInfo, restored: ObservationInfo) -> None: 

484 # Compare the full simple forms rather than relying on ``==``, 

485 # which only compares the properties present on both sides. 

486 self.assertEqual(original.to_simple(), restored.to_simple()) 

487 

488 def test_round_trip_headers(self) -> None: 

489 for filename in HEADER_FILES: 

490 obsinfo = _read_obsinfo(filename) 

491 for prefix, sep in PREFIX_FORMS: 

492 with self.subTest(filename=filename, prefix=prefix): 

493 flat = obsinfo.to_flat_dict(prefix=prefix, sep=sep) 

494 for value in flat.values(): 

495 self.assertIsInstance(value, str | int | float | bool) 

496 self.assertTrue(ObservationInfo.has_flat_dict(flat)) 

497 self.assert_round_trip(obsinfo, ObservationInfo.from_flat_dict(flat)) 

498 restored = ObservationInfo.from_flat_dict(flat, prefix=prefix, sep=sep) 

499 self.assert_round_trip(obsinfo, restored) 

500 

501 def test_fits_form_keys(self) -> None: 

502 obsinfo = _read_obsinfo("fitsheader-decam.yaml") 

503 flat = obsinfo.to_flat_dict(prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

504 self.assertIn("OBS_INFO DATETIME_BEGIN_0", flat) 

505 self.assertIn("OBS_INFO DATETIME_BEGIN_1", flat) 

506 self.assertIn("OBS_INFO TRACKING_RADEC_0", flat) 

507 self.assertIn("OBS_INFO LOCATION_2", flat) 

508 self.assertEqual(flat["OBS_INFO TRANSLATOR"], "DECam") 

509 self.assertNotIn("OBS_INFO _TRANSLATOR", flat) 

510 for key in flat: 

511 self.assertEqual(key, key.upper()) 

512 

513 def test_round_trip_hand_built(self) -> None: 

514 begin = astropy.time.Time("2025-01-01T00:00:00", scale="tai") 

515 obsinfo = makeObservationInfo(observation_id="A", exposure_time=30.0 * u.s, datetime_begin=begin) 

516 flat = obsinfo.to_flat_dict() 

517 self.assertNotIn("translator", flat) 

518 self.assert_round_trip(obsinfo, ObservationInfo.from_flat_dict(flat)) 

519 

520 # Without any tuple-valued property there is no indexed anchor so 

521 # auto-detection finds nothing and an explicit prefix is needed. 

522 obsinfo = makeObservationInfo(observation_id="A", exposure_time=30.0 * u.s) 

523 flat = obsinfo.to_flat_dict() 

524 self.assertFalse(ObservationInfo.has_flat_dict(flat)) 

525 with self.assertRaises(ValueError): 

526 ObservationInfo.from_flat_dict(flat) 

527 self.assertTrue(ObservationInfo.has_flat_dict(flat, prefix="")) 

528 self.assert_round_trip(obsinfo, ObservationInfo.from_flat_dict(flat, prefix="")) 

529 

530 def test_round_trip_extensions(self) -> None: 

531 begin = astropy.time.Time("2025-01-01T00:00:00", scale="tai") 

532 obsinfo = makeObservationInfo( 

533 translator_class=_FlatDictTranslator, 

534 instrument="flatdict", 

535 datetime_begin=begin, 

536 ext_number=42, 

537 ext_foo="bar", 

538 ext_pair=(1.5, 2.5), 

539 ) 

540 for prefix, sep in PREFIX_FORMS: 

541 with self.subTest(prefix=prefix): 

542 flat = obsinfo.to_flat_dict(prefix=prefix, sep=sep) 

543 restored = ObservationInfo.from_flat_dict(flat) 

544 self.assertEqual(restored.translator_name, "flatdict") 

545 self.assertEqual(restored.ext_number, 42) 

546 self.assertEqual(restored.ext_foo, "bar") 

547 self.assertEqual(restored.ext_pair, (1.5, 2.5)) 

548 self.assert_round_trip(obsinfo, restored) 

549 flat = obsinfo.to_flat_dict(prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

550 self.assertEqual(flat["OBS_INFO EXT_PAIR_0"], 1.5) 

551 self.assertEqual(flat["OBS_INFO EXT_PAIR_1"], 2.5) 

552 

553 def test_non_finite_round_trip(self) -> None: 

554 obsinfo = makeObservationInfo( 

555 observation_id="A", 

556 object="NAN", 

557 relative_humidity=math.nan, 

558 temperature=math.inf * u.K, 

559 pressure=-math.inf * u.hPa, 

560 location=EarthLocation(math.nan, 1.0, math.inf, unit=u.m), 

561 ) 

562 for prefix, sep in PREFIX_FORMS: 

563 with self.subTest(prefix=prefix): 

564 flat = obsinfo.to_flat_dict(prefix=prefix, sep=sep, encode_non_finite=True) 

565 self.assertTrue(all(math.isfinite(v) for v in flat.values() if isinstance(v, float))) 

566 strings = {v for v in flat.values() if isinstance(v, str)} 

567 self.assertTrue({"NAN", "+INFINITY", "-INFINITY"} <= strings) 

568 

569 restored = ObservationInfo.from_flat_dict(flat, prefix=prefix, sep=sep) 

570 self.assertEqual(restored.object, "NAN") 

571 self.assertTrue(math.isnan(restored.relative_humidity)) 

572 self.assertEqual(restored.temperature.value, math.inf) 

573 self.assertEqual(restored.pressure.value, -math.inf) 

574 self.assertTrue(math.isnan(restored.location.x.value)) 

575 self.assertEqual(restored.location.y.value, 1.0) 

576 self.assertEqual(restored.location.z.value, math.inf) 

577 self.assertEqual(restored, obsinfo) 

578 

579 def test_from_flat_dict_decodes_json_spellings(self) -> None: 

580 # The spellings pydantic uses when writing JSON are accepted too. 

581 flat = { 

582 "observation_id": "A", 

583 "datetime_begin_0": 2460000.5, 

584 "datetime_begin_1": 0.25, 

585 "temperature": "Infinity", 

586 "pressure": "-Infinity", 

587 "relative_humidity": "NaN", 

588 } 

589 restored = ObservationInfo.from_flat_dict(flat, prefix="") 

590 self.assertEqual(restored.temperature.value, math.inf) 

591 self.assertEqual(restored.pressure.value, -math.inf) 

592 self.assertTrue(math.isnan(restored.relative_humidity)) 

593 

594 def test_has_and_strip(self) -> None: 

595 obsinfo = _read_obsinfo("fitsheader-hsc.yaml") 

596 flat = obsinfo.to_flat_dict(prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

597 n_flat = len(flat) 

598 header = {"SIMPLE": True, "OBJECT": "decoy", "TEMPERATURE": 3.0, **flat} 

599 self.assertTrue(ObservationInfo.has_flat_dict(header)) 

600 self.assertTrue( 

601 ObservationInfo.has_flat_dict(header, prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

602 ) 

603 ObservationInfo.strip_from_flat_dict(header) 

604 self.assertEqual(header, {"SIMPLE": True, "OBJECT": "decoy", "TEMPERATURE": 3.0}) 

605 self.assertEqual(len(flat), n_flat) 

606 self.assertFalse(ObservationInfo.has_flat_dict(header)) 

607 # Stripping again is a no-op. 

608 ObservationInfo.strip_from_flat_dict(header) 

609 self.assertEqual(len(header), 3) 

610 

611 def test_scalar_index_is_not_an_anchor(self) -> None: 

612 header = {"OBJECT": "M31", "OBJECT_1": "x", "TEMPERATURE": 3.0} 

613 self.assertFalse(ObservationInfo.has_flat_dict(header)) 

614 ObservationInfo.strip_from_flat_dict(header) 

615 self.assertEqual(len(header), 3) 

616 with self.assertRaises(ValueError): 

617 ObservationInfo.from_flat_dict(header) 

618 

619 def test_case_duplicates(self) -> None: 

620 obsinfo = _read_obsinfo("fitsheader-hsc.yaml") 

621 flat = obsinfo.to_flat_dict(prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

622 header = {"OBJECT": "decoy", "obs_info object": "stale", **flat} 

623 for kwargs in ({}, {"prefix": FITS_FLAT_DICT_PREFIX, "sep": FITS_FLAT_DICT_SEP}): 

624 with self.subTest(kwargs=kwargs): 

625 self.assertTrue(ObservationInfo.has_flat_dict(header, **kwargs)) 

626 with self.assertRaises(ValueError): 

627 ObservationInfo.from_flat_dict(header, **kwargs) 

628 ObservationInfo.strip_from_flat_dict(header, prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

629 self.assertEqual(header, {"OBJECT": "decoy"}) 

630 

631 def test_strip_explicit_prefix_without_content(self) -> None: 

632 header = {"OBJECT": "M31", "EXPTIME": 30.0} 

633 self.assertFalse( 

634 ObservationInfo.has_flat_dict(header, prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

635 ) 

636 ObservationInfo.strip_from_flat_dict(header, prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

637 self.assertEqual(header, {"OBJECT": "M31", "EXPTIME": 30.0}) 

638 

639 def test_from_flat_dict_errors(self) -> None: 

640 with self.assertRaises(ValueError): 

641 ObservationInfo.from_flat_dict({}) 

642 with self.assertRaises(ValueError): 

643 ObservationInfo.from_flat_dict({"OBJECT": "M31"}) 

644 

645 obsinfo = _read_obsinfo("fitsheader-decam.yaml") 

646 flat = obsinfo.to_flat_dict(prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

647 

648 gap = dict(flat) 

649 del gap["OBS_INFO DATETIME_BEGIN_1"] 

650 with self.assertRaises(ValueError): 

651 ObservationInfo.from_flat_dict(gap) 

652 

653 corrupt = dict(flat) 

654 corrupt["OBS_INFO EXPOSURE_ID"] = "not an integer" 

655 # Validation errors propagate rather than being mistaken for absence. 

656 with self.assertRaises(ValueError): 

657 ObservationInfo.from_flat_dict(corrupt) 

658 self.assertTrue(ObservationInfo.has_flat_dict(corrupt)) 

659 

660 def test_explicit_prefix_wrong_sep(self) -> None: 

661 obsinfo = _read_obsinfo("fitsheader-decam.yaml") 

662 flat = obsinfo.to_flat_dict(prefix=FITS_FLAT_DICT_PREFIX, sep=FITS_FLAT_DICT_SEP) 

663 self.assertFalse(ObservationInfo.has_flat_dict(flat, prefix=FITS_FLAT_DICT_PREFIX, sep=".")) 

664 with self.assertRaises(ValueError): 

665 ObservationInfo.from_flat_dict(flat, prefix=FITS_FLAT_DICT_PREFIX, sep=".") 

666 

667 def test_to_flat_dict_bad_sep(self) -> None: 

668 obsinfo = _read_obsinfo("fitsheader-decam.yaml") 

669 with self.assertRaises(ValueError): 

670 obsinfo.to_flat_dict(prefix="p", sep="_") 

671 

672 

673class FitsHeaderTestCase(unittest.TestCase): 

674 """Test the flat dict through a real FITS header file.""" 

675 

676 def assert_fits_round_trip(self, original: ObservationInfo, restored: ObservationInfo) -> None: 

677 # FITS writers keep at least 15 significant digits of a float 

678 # rather than its shortest exact representation, so floats are 

679 # compared to that precision and everything else exactly. 

680 before = original.to_simple() 

681 after = restored.to_simple() 

682 self.assertEqual(set(before), set(after)) 

683 for key, value in before.items(): 

684 if isinstance(value, float): 

685 self.assertTrue(math.isclose(after[key], value, rel_tol=1e-14), key) 

686 elif isinstance(value, tuple) and isinstance(value[0], float): 

687 for element_after, element_before in zip(after[key], value, strict=True):