Coverage for tests / test_visit_geometry.py: 18%

1# This file is part of obs_base. 

2# 

3# Developed for the LSST Data Management System. 

4# This product includes software developed by the LSST Project 

5# (https://www.lsst.org). 

6# See the COPYRIGHT file at the top-level directory of this distribution 

7# for details of code ownership. 

8# 

9# This program is free software: you can redistribute it and/or modify 

10# it under the terms of the GNU General Public License as published by 

11# the Free Software Foundation, either version 3 of the License, or 

12# (at your option) any later version. 

13# 

14# This program is distributed in the hope that it will be useful, 

15# but WITHOUT ANY WARRANTY; without even the implied warranty of 

16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 

17# GNU General Public License for more details. 

18# 

19# You should have received a copy of the GNU General Public License 

20# along with this program. If not, see <http://www.gnu.org/licenses/>. 

21 

22import tempfile 

23import unittest 

24 

25import numpy as np 

26 

27from lsst.daf.butler import Butler, DatasetType, DimensionRecord 

28from lsst.obs.base.visit_geometry import VisitGeometry 

29from lsst.sphgeom import Angle, ConvexPolygon, LonLat, UnitVector3d 

30 

31 

32class VisitGeometryTestCase(unittest.TestCase): 

33 """Tests for VisitGeometry.""" 

34 

35 def setUp(self): 

36 root = self.enterContext(tempfile.TemporaryDirectory(ignore_cleanup_errors=True)) 

37 Butler.makeRepo(root) 

38 self.butler: Butler = self.enterContext(Butler.from_config(root, run="geometry")) 

39 self.rng = np.random.default_rng(500) 

40 self.rot_about = UnitVector3d(*self.rng.uniform(0.0, 1.0, size=3).tolist()) 

41 self.rot_amount = Angle.fromDegrees(self.rng.uniform(0.0, 5.0)) 

42 

43 def test_round_trip(self): 

44 """Test round-tripping through a JSON string.""" 

45 self.butler.import_(filename="resource://lsst.daf.butler/tests/registry_data/base.yaml") 

46 self.butler.import_(filename="resource://lsst.daf.butler/tests/registry_data/spatial.yaml") 

47 (visit_record,) = self.butler.query_dimension_records("visit", visit=1, instrument="Cam1") 

48 visit_detector_region_records = self.butler.query_dimension_records( 

49 "visit_detector_region", visit=1, instrument="Cam1" 

50 ) 

51 vg1 = VisitGeometry( 

52 boresight_ra=45.0, 

53 boresight_dec=60.0, 

54 orientation=30.0, 

55 visit_region=visit_record.region, 

56 detector_regions={r.detector: r.region for r in visit_detector_region_records}, 

57 ) 

58 json_data = vg1.model_dump_json() 

59 vg2 = VisitGeometry.model_validate_json(json_data) 

60 self.assertEqual(vg1.boresight_ra, vg2.boresight_ra) 

61 self.assertEqual(vg1.boresight_dec, vg2.boresight_dec) 

62 self.assertEqual(vg1.orientation, vg2.orientation) 

63 self.assertEqual(vg1.visit_region, vg2.visit_region) 

64 self.assertEqual(vg1.detector_regions, vg2.detector_regions) 

65 

66 def test_update_dimension_records(self): 

67 """Test the update_dimension_records method.""" 

68 self.butler.import_(filename="resource://lsst.daf.butler/tests/registry_data/base.yaml") 

69 self.butler.import_(filename="resource://lsst.daf.butler/tests/registry_data/spatial.yaml") 

70 old_visit_records = {r.id: r for r in self.butler.query_dimension_records("visit")} 

71 old_exposure_records = { 

72 r.id: self._invent_and_insert_exposure_record(r) for r in old_visit_records.values() 

73 } 

74 self.butler.registry.registerDatasetType( 

75 DatasetType("visit_geometry", self.butler.dimensions.conform(["visit"]), "VisitGeometry") 

76 ) 

77 visit_geometries: dict[int, VisitGeometry] = {} 

78 self.assertEqual(len(old_visit_records), 2) 

79 for visit_id, visit_record in old_visit_records.items(): 

80 exposure_record = old_exposure_records[visit_id] 

81 old_boresight = LonLat.fromDegrees(exposure_record.tracking_ra, exposure_record.tracking_dec) 

82 new_boresight = LonLat(self._modify_point(UnitVector3d(old_boresight))) 

83 visit_geometry = VisitGeometry( 

84 boresight_ra=new_boresight.getLon().asDegrees(), 

85 boresight_dec=new_boresight.getLat().asDegrees(), 

86 # We *mostly* try to modify the geometry self-consistently, but 

87 # the orientation angle is just arbitrary. Nothing in the test 

88 # right now requires any kind of consistency between the 

89 # regions and the exposure angle fields. 

90 orientation=self.rng.uniform(0.0, 5.0), 

91 visit_region=self._modify_polygon(visit_record.region), 

92 detector_regions={ 

93 r.detector: self._modify_polygon(r.region) 

94 for r in self.butler.query_dimension_records( 

95 "visit_detector_region", data_id=visit_record.dataId 

96 ) 

97 }, 

98 ) 

99 visit_geometries[visit_id] = visit_geometry 

100 self.butler.put(visit_geometry, "visit_geometry", visit_record.dataId) 

101 n_updated = VisitGeometry.update_dimension_records(self.butler, instrument="Cam1") 

102 self.assertEqual(n_updated, 2) 

103 for new_visit_record in self.butler.query_dimension_records("visit"): 

104 visit_id = new_visit_record.id 

105 old_visit_record = old_visit_records[visit_id] 

106 visit_geometry = visit_geometries[visit_id] 

107 self.assertNotEqual(new_visit_record.region, old_visit_record.region) 

108 self.assertEqual(new_visit_record.region, visit_geometry.visit_region) 

109 for new_vdr_record in self.butler.query_dimension_records( 

110 "visit_detector_region", data_id=new_visit_record.dataId 

111 ): 

112 self.assertEqual( 

113 new_vdr_record.region, visit_geometry.detector_regions[new_vdr_record.detector] 

114 ) 

115 for new_exposure_record in self.butler.query_dimension_records("exposure"): 

116 visit_id = new_exposure_record.id 

117 visit_geometry = visit_geometries[visit_id] 

118 self.assertEqual(new_exposure_record.tracking_ra, visit_geometry.boresight_ra) 

119 self.assertEqual(new_exposure_record.tracking_dec, visit_geometry.boresight_dec) 

120 self.assertEqual(new_exposure_record.sky_angle, visit_geometry.orientation) 

121 

122 def _invent_and_insert_exposure_record(self, visit_record: DimensionRecord) -> DimensionRecord: 

123 group_record = self.butler.dimensions["group"].RecordClass( 

124 instrument=visit_record.instrument, name=str(visit_record.id) 

125 ) 

126 self.butler.registry.insertDimensionData("group", group_record, skip_existing=True) 

127 center = LonLat(visit_record.region.getCentroid()) 

128 exposure_record = self.butler.dimensions["exposure"].RecordClass( 

129 instrument=visit_record.instrument, 

130 id=visit_record.id, 

131 day_obs=visit_record.day_obs, 

132 group=group_record.name, 

133 physical_filter=visit_record.physical_filter, 

134 tracking_ra=center.getLon().asDegrees(), 

135 tracking_dec=center.getLat().asDegrees(), 

136 sky_angle=float(self.rng.uniform(0.0, 5.0)), 

137 ) 

138 self.butler.registry.insertDimensionData("exposure", exposure_record) 

139 visit_definition_record = self.butler.dimensions["visit_definition"].RecordClass( 

140 instrument=visit_record.instrument, 

141 visit=visit_record.id, 

142 exposure=exposure_record.id, 

143 ) 

144 self.butler.registry.insertDimensionData("visit_definition", visit_definition_record) 

145 return exposure_record 

146 

147 def _modify_polygon(self, polygon: ConvexPolygon) -> ConvexPolygon: 

148 return ConvexPolygon([self._modify_point(v) for v in polygon.getVertices()]) 

149 

150 def _modify_point(self, u: UnitVector3d) -> UnitVector3d: 

151 return u.rotatedAround(self.rot_about, self.rot_amount) 

152 

153 

154if __name__ == "__main__": 154 ↛ 155line 154 didn't jump to line 155 because the condition on line 154 was never true

155 unittest.main()