Coverage for python / lsst / obs / base / instrument_tests.py: 53%

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 

22"""Helpers for writing tests against subclassses of Instrument. 

23 

24These are not tests themselves, but can be subclassed (plus unittest.TestCase) 

25to get a functional test of an Instrument. 

26""" 

27 

28from __future__ import annotations 

29 

30__all__ = [ 

31 "CuratedCalibration", 

32 "DummyCam", 

33 "DummyCamYamlWcsFormatter", 

34 "InstrumentTestData", 

35 "InstrumentTests", 

36] 

37 

38import abc 

39import dataclasses 

40from collections.abc import Sequence 

41from functools import lru_cache 

42from typing import TYPE_CHECKING, Any, ClassVar 

43 

44from pydantic import BaseModel 

45 

46from lsst.daf.butler import CollectionType, DatasetType 

47from lsst.daf.butler.formatters.yaml import YamlFormatter 

48from lsst.daf.butler.tests.utils import create_populated_sqlite_registry 

49from lsst.obs.base import FilterDefinition, FilterDefinitionCollection, Instrument 

50from lsst.obs.base.yamlCamera import makeCamera 

51from lsst.resources import ResourcePath 

52from lsst.utils.introspection import get_full_type_name 

53 

54from .utils import createInitialSkyWcsFromBoresight 

55 

56if TYPE_CHECKING: 

57 from lsst.daf.butler import Butler 

58 

59DUMMY_FILTER_DEFINITIONS = FilterDefinitionCollection( 

60 FilterDefinition(physical_filter="dummy_u", band="u"), 

61 FilterDefinition(physical_filter="dummy_g", band="g"), 

62) 

63 

64 

65class DummyCamYamlWcsFormatter(YamlFormatter): 

66 """Specialist formatter for tests that can make a WCS.""" 

67 

68 @classmethod 

69 def makeRawSkyWcsFromBoresight(cls, boresight, orientation, detector): 

70 """Class method to make a raw sky WCS from boresight and detector. 

71 

72 This uses the API expected by define-visits. A working example 

73 can be found in `FitsRawFormatterBase`. 

74 

75 Notes 

76 ----- 

77 This makes no attempt to create a proper WCS from geometry. 

78 """ 

79 return createInitialSkyWcsFromBoresight(boresight, orientation, detector, flipX=False) 

80 

81 

82class CuratedCalibration(BaseModel): 

83 """Class that implements minimal read/write interface needed to support 

84 curated calibration ingest. 

85 """ 

86 

87 metadata: dict[str, Any] 

88 values: list[int] 

89 

90 @classmethod 

91 def readText(cls, path: str) -> CuratedCalibration: 

92 with open(path) as f: 

93 data = f.read() 

94 return cls.model_validate_json(data) 

95 

96 def writeText(self, path: str) -> None: 

97 with open(path, "w") as f: 

98 print(self.model_dump_json(), file=f) 

99 

100 def getMetadata(self) -> dict[str, Any]: 

101 return self.metadata 

102 

103 

104class DummyCam(Instrument): 

105 """Instrument class used for testing.""" 

106 

107 filterDefinitions = DUMMY_FILTER_DEFINITIONS 

108 additionalCuratedDatasetTypes = frozenset(["testCalib"]) 

109 policyName = "dummycam" 

110 dataPackageDir: str | None = "" 

111 raw_definition = ( 

112 "raw_dict", 

113 ("instrument", "detector", "exposure"), 

114 "StructuredDataDict", 

115 ) 

116 

117 @classmethod 

118 def getName(cls): 

119 return "DummyCam" 

120 

121 @classmethod 

122 @lru_cache # For mypy 

123 def getObsDataPackageDir(cls) -> str | None: 

124 return cls.dataPackageDir 

125 

126 def getCamera(self): 

127 # Return something that can be indexed by detector number 

128 # but also has to support getIdIter. 

129 with ResourcePath("resource://lsst.obs.base/test/dummycam.yaml").as_local() as local_file: 

130 return makeCamera(local_file.ospath) 

131 

132 def register(self, registry, update=False): 

133 """Insert Instrument, physical_filter, and detector entries into a 

134 `Registry`. 

135 """ 

136 detector_max = 2 

137 dataId = { 

138 "instrument": self.getName(), 

139 "class_name": get_full_type_name(DummyCam), 

140 "detector_max": detector_max, 

141 "visit_max": 1_000_000, 

142 "exposure_max": 1_000_000, 

143 } 

144 

145 with registry.transaction(): 

146 registry.syncDimensionData("instrument", dataId, update=update) 

147 self._registerFilters(registry, update=update) 

148 for d in range(detector_max): 

149 registry.syncDimensionData( 

150 "detector", 

151 dict( 

152 instrument=self.getName(), 

153 id=d, 

154 full_name=f"RXX_S0{d}", 

155 ), 

156 update=update, 

157 ) 

158 

159 def getRawFormatter(self, dataId): 

160 # Docstring inherited fromt Instrument.getRawFormatter. 

161 return DummyCamYamlWcsFormatter 

162 

163 def applyConfigOverrides(self, name, config): 

164 pass 

165 

166 def writeAdditionalCuratedCalibrations( 

167 self, butler: Butler, collection: str | None = None, labels: Sequence[str] = () 

168 ) -> None: 

169 # We want to test the standard curated calibration ingest 

170 # but we do not have a standard class to use. There is no way 

171 # at the moment to inject a new class into the standard list 

172 # that is a package constant, so instead use this "Additional" 

173 # method but call the standard curated calibration code. 

174 if collection is None: 

175 collection = self.makeCalibrationCollectionName(*labels) 

176 butler.registry.registerCollection(collection, type=CollectionType.CALIBRATION) 

177 

178 datasetType = DatasetType( 

179 "testCalib", 

180 universe=butler.dimensions, 

181 isCalibration=True, 

182 dimensions=("instrument", "detector"), 

183 storageClass="CuratedCalibration", 

184 ) 

185 runs: set[str] = set() 

186 self._writeSpecificCuratedCalibrationDatasets( 

187 butler, datasetType, collection, runs=runs, labels=labels 

188 ) 

189 

190 

191@dataclasses.dataclass 

192class InstrumentTestData: 

193 """Values to test against in subclasses of `InstrumentTests`.""" 

194 

195 name: str 

196 """The name of the Camera this instrument describes.""" 

197 

198 nDetectors: int 

199 """The number of detectors in the Camera.""" 

200 

201 firstDetectorName: str 

202 """The name of the first detector in the Camera.""" 

203 

204 physical_filters: set[str] 

205 """A subset of the physical filters should be registered.""" 

206 

207 

208class InstrumentTests(metaclass=abc.ABCMeta): 

209 """Tests of sublcasses of Instrument. 

210 

211 TestCase subclasses must derive from this, then `TestCase`, and override 

212 ``data`` and ``instrument``. 

213 """ 

214 

215 data: ClassVar[InstrumentTestData | None] = None 

216 """`InstrumentTestData` containing the values to test against.""" 

217 

218 instrument: ClassVar[Instrument | None] = None 

219 """The `~lsst.obs.base.Instrument` to be tested.""" 

220 

221 def test_name(self): 

222 self.assertEqual(self.instrument.getName(), self.data.name) 

223 

224 def test_getCamera(self): 

225 """Test that getCamera() returns a reasonable Camera definition.""" 

226 camera = self.instrument.getCamera() 

227 self.assertEqual(camera.getName(), self.instrument.getName()) 

228 self.assertEqual(len(camera), self.data.nDetectors) 

229 self.assertEqual(next(iter(camera)).getName(), self.data.firstDetectorName) 

230 

231 def test_register(self): 

232 """Test that register() sets appropriate Dimensions.""" 

233 with create_populated_sqlite_registry() as butler: 

234 registry = butler.registry 

235 # Check that the registry starts out empty. 

236 self.assertFalse(registry.queryDataIds(["instrument"]).toSequence()) 

237 self.assertFalse(registry.queryDataIds(["detector"]).toSequence()) 

238 self.assertFalse(registry.queryDataIds(["physical_filter"]).toSequence()) 

239 

240 # Register the instrument and check that certain dimensions appear. 

241 self.instrument.register(registry) 

242 instrumentDataIds = registry.queryDataIds(["instrument"]).toSequence() 

243 self.assertEqual(len(instrumentDataIds), 1) 

244 instrumentNames = {dataId["instrument"] for dataId in instrumentDataIds} 

245 self.assertEqual(instrumentNames, {self.data.name}) 

246 detectorDataIds = registry.queryDataIds(["detector"]).expanded().toSequence() 

247 self.assertEqual(len(detectorDataIds), self.data.nDetectors) 

248 detectorNames = {dataId.records["detector"].full_name for dataId in detectorDataIds} 

249 self.assertIn(self.data.firstDetectorName, detectorNames) 

250 physicalFilterDataIds = registry.queryDataIds(["physical_filter"]).toSequence() 

251 filterNames = {dataId["physical_filter"] for dataId in physicalFilterDataIds} 

252 self.assertGreaterEqual(filterNames, self.data.physical_filters) 

253 

254 # Check that the instrument class can be retrieved. 

255 registeredInstrument = Instrument.fromName(self.instrument.getName(), registry) 

256 self.assertEqual(type(registeredInstrument), type(self.instrument)) 

257 

258 # Check that re-registration is not an error. 

259 self.instrument.register(registry)