Coverage for tests/test_serialization.py: 21%

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"""Test that archival ObservationInfo JSON files can be read back. 

13 

14The JSON files in ``tests/data`` are stored long term, so we need to 

15guarantee that future versions of the code can still deserialize them. 

16""" 

17 

18import os 

19import unittest 

20from collections.abc import Mapping 

21from typing import Any 

22 

23import astropy.coordinates 

24import astropy.time 

25import astropy.units as u 

26import pydantic 

27 

28from astro_metadata_translator import ( 

29 ObservationInfo, 

30 PropertyDefinition, 

31 StubTranslator, 

32) 

33 

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

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

36 

37 

38# Auto-registers via StubTranslator.__init_subclass__ when this module is 

39# imported, so the "_translator": "fixture" reference in the archival JSON 

40# resolves and its extension definitions are available. 

41class _FixtureTranslator(StubTranslator): 

42 name = "fixture" 

43 supported_instrument = "fixture" 

44 extensions = dict( 

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

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

47 ) 

48 

49 @classmethod 

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

51 # Never used for header-based auto-detection; this translator only 

52 # exists so the "_translator": "fixture" reference in the archival 

53 # JSON can be resolved. 

54 return False 

55 

56 

57class ObservationInfoSerializationTestCase(unittest.TestCase): 

58 """Read archival ObservationInfo JSON fixtures and check the values.""" 

59 

60 def _check_core_properties(self, obsinfo: ObservationInfo) -> None: 

61 self.assertEqual(obsinfo.telescope, "Test Telescope") 

62 self.assertEqual(obsinfo.instrument, "TestCam") 

63 self.assertEqual(obsinfo.exposure_id, 987654321) 

64 self.assertEqual(obsinfo.visit_id, 987654000) 

65 self.assertEqual(obsinfo.physical_filter, "r") 

66 self.assertEqual(obsinfo.detector_num, 42) 

67 self.assertEqual(obsinfo.detector_name, "S00") 

68 self.assertEqual(obsinfo.detector_unique_name, "R22_S00") 

69 self.assertEqual(obsinfo.detector_serial, "ITL-3800C-145") 

70 self.assertEqual(obsinfo.detector_group, "R22") 

71 self.assertEqual(obsinfo.detector_exposure_id, 98765432142) 

72 self.assertEqual(obsinfo.object, "HD12345") 

73 self.assertEqual(obsinfo.science_program, "2025A-001") 

74 self.assertEqual(obsinfo.observation_type, "science") 

75 self.assertEqual(obsinfo.observation_id, "TC_O_20250521_000123") 

76 self.assertEqual(obsinfo.observation_reason, "science") 

77 self.assertEqual(obsinfo.exposure_group, "20250521_000123") 

78 self.assertEqual(obsinfo.observing_day, 20250520) 

79 self.assertEqual(obsinfo.observation_counter, 123) 

80 self.assertEqual(obsinfo.group_counter_start, 123) 

81 self.assertEqual(obsinfo.group_counter_end, 123) 

82 self.assertEqual(obsinfo.boresight_rotation_coord, "sky") 

83 self.assertEqual(obsinfo.has_simulated_content, False) 

84 self.assertEqual(obsinfo.can_see_sky, True) 

85 

86 self.assertAlmostEqual(obsinfo.boresight_airmass, 1.1037) 

87 self.assertAlmostEqual(obsinfo.relative_humidity, 42.5) 

88 

89 self.assertIsInstance(obsinfo.location, astropy.coordinates.EarthLocation) 

90 self.assertAlmostEqual(obsinfo.location.height.to_value(u.m), 2663.0, places=3) 

91 

92 self.assertIsInstance(obsinfo.datetime_begin, astropy.time.Time) 

93 self.assertEqual(obsinfo.datetime_begin.scale, "tai") 

94 self.assertEqual( 

95 obsinfo.datetime_begin.tai.isot, 

96 "2025-05-21T01:23:45.000", 

97 ) 

98 self.assertEqual( 

99 obsinfo.datetime_end.tai.isot, 

100 "2025-05-21T01:24:15.000", 

101 ) 

102 

103 self.assertAlmostEqual(obsinfo.exposure_time.to_value(u.s), 30.0) 

104 self.assertAlmostEqual(obsinfo.exposure_time_requested.to_value(u.s), 30.0) 

105 self.assertAlmostEqual(obsinfo.dark_time.to_value(u.s), 31.0) 

106 self.assertAlmostEqual(obsinfo.focus_z.to_value(u.m), 0.001) 

107 self.assertAlmostEqual(obsinfo.temperature.to_value(u.K), 283.15) 

108 self.assertAlmostEqual(obsinfo.pressure.to_value(u.hPa), 750.0) 

109 self.assertAlmostEqual(obsinfo.boresight_rotation_angle.to_value(u.deg), 45.0) 

110 self.assertEqual(round(obsinfo.observing_day_offset.to_value(u.s)), 43200) 

111 

112 self.assertIsInstance(obsinfo.tracking_radec, astropy.coordinates.SkyCoord) 

113 self.assertAlmostEqual(obsinfo.tracking_radec.icrs.ra.to_value(u.deg), 123.456) 

114 self.assertAlmostEqual(obsinfo.tracking_radec.icrs.dec.to_value(u.deg), -45.678) 

115 

116 self.assertIsInstance(obsinfo.altaz_begin, astropy.coordinates.AltAz) 

117 self.assertAlmostEqual(obsinfo.altaz_begin.az.to_value(u.deg), 110.0) 

118 self.assertAlmostEqual(obsinfo.altaz_begin.alt.to_value(u.deg), 65.0) 

119 self.assertAlmostEqual(obsinfo.altaz_end.az.to_value(u.deg), 110.5) 

120 self.assertAlmostEqual(obsinfo.altaz_end.alt.to_value(u.deg), 64.5) 

121 

122 def test_read_full(self) -> None: 

123 """Read the fixture with every core property populated.""" 

124 path = os.path.join(DATADIR, "obsinfo-full.json") 

125 with open(path) as fh: 

126 obsinfo = ObservationInfo.from_json(fh.read()) 

127 

128 self.assertIsInstance(obsinfo, ObservationInfo) 

129 self._check_core_properties(obsinfo) 

130 # No extensions present in this fixture. 

131 self.assertEqual(obsinfo.extensions, {}) 

132 

133 # Round-tripping through JSON should be stable. 

134 round_tripped = ObservationInfo.from_json(obsinfo.to_json()) 

135 self.assertEqual(round_tripped, obsinfo) 

136 

137 def test_read_full_with_extensions(self) -> None: 

138 """Read the fixture that also contains extension keys.""" 

139 path = os.path.join(DATADIR, "obsinfo-full-extensions.json") 

140 with open(path) as fh: 

141 obsinfo = ObservationInfo.from_json(fh.read()) 

142 

143 self.assertIsInstance(obsinfo, ObservationInfo) 

144 self._check_core_properties(obsinfo) 

145 

146 # Extension properties round-trip via the "fixture" translator name 

147 # stored in the JSON. 

148 self.assertEqual(set(obsinfo.extensions), {"number", "foo"}) 

149 self.assertEqual(obsinfo.ext_foo, "bar") 

150 self.assertEqual(obsinfo.ext_number, 12345) 

151 

152 round_tripped = ObservationInfo.from_json(obsinfo.to_json()) 

153 self.assertEqual(round_tripped, obsinfo) 

154 self.assertEqual(round_tripped.ext_foo, "bar") 

155 self.assertEqual(round_tripped.ext_number, 12345) 

156 

157 def test_nested_in_pydantic_model(self) -> None: 

158 """Nesting in another Pydantic model preserves the wire format. 

159 

160 Nested serialization goes through Pydantic directly and does not 

161 call ``to_simple`` / ``to_json``, so the alias behavior must come 

162 from ``serialize_by_alias`` in the model config rather than from 

163 the call-site kwargs. 

164 """ 

165 

166 class Wrapper(pydantic.BaseModel): 

167 obs: ObservationInfo 

168 

169 path = os.path.join(DATADIR, "obsinfo-full-extensions.json") 

170 with open(path) as fh: 

171 obsinfo = ObservationInfo.from_json(fh.read()) 

172 

173 wrapper = Wrapper(obs=obsinfo) 

174 

175 nested = wrapper.model_dump()["obs"] 

176 self.assertIn("_translator", nested) 

177 self.assertNotIn("translator_name", nested) 

178 # Extension keys remain flat under the ext_ prefix. 

179 self.assertEqual(nested["ext_foo"], "bar") 

180 self.assertEqual(nested["ext_number"], 12345) 

181 

182 nested_json = wrapper.model_dump_json() 

183 round_tripped = Wrapper.model_validate_json(nested_json) 

184 self.assertEqual(round_tripped.obs, obsinfo) 

185 

186 

187if __name__ == "__main__": 

188 unittest.main()