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

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 Callable, 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, DataId, 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 import lsst.afw.cameraGeom 

58 import lsst.afw.geom 

59 import lsst.geom 

60 import lsst.pex.config 

61 from lsst.daf.butler import Butler, Formatter, FormatterV2, Registry 

62 

63DUMMY_FILTER_DEFINITIONS = FilterDefinitionCollection( 

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

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

66) 

67 

68 

69class DummyCamYamlWcsFormatter(YamlFormatter): 

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

71 

72 @classmethod 

73 def makeRawSkyWcsFromBoresight( 

74 cls, 

75 boresight: lsst.geom.SpherePoint, 

76 orientation: lsst.geom.Angle, 

77 detector: lsst.afw.cameraGeom.Detector, 

78 ) -> lsst.afw.geom.SkyWcs: 

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

80 

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

82 can be found in `~lsst.obs.base.FitsRawFormatterBase`. 

83 

84 Notes 

85 ----- 

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

87 """ 

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

89 

90 

91class CuratedCalibration(BaseModel): 

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

93 curated calibration ingest. 

94 """ 

95 

96 metadata: dict[str, Any] 

97 values: list[int] 

98 

99 @classmethod 

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

101 with open(path) as f: 

102 data = f.read() 

103 return cls.model_validate_json(data) 

104 

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

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

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

108 

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

110 return self.metadata 

111 

112 

113class DummyCam(Instrument): 

114 """Instrument class used for testing.""" 

115 

116 filterDefinitions: FilterDefinitionCollection = DUMMY_FILTER_DEFINITIONS 

117 additionalCuratedDatasetTypes = frozenset(["testCalib"]) 

118 policyName = "dummycam" 

119 dataPackageDir: str | None = "" 

120 raw_definition = ( 

121 "raw_dict", 

122 ("instrument", "detector", "exposure"), 

123 "StructuredDataDict", 

124 ) 

125 

126 @classmethod 

127 def getName(cls) -> str: 

128 return "DummyCam" 

129 

130 @classmethod 

131 @lru_cache # For mypy 

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

133 return cls.dataPackageDir 

134 

135 def getCamera(self) -> lsst.afw.cameraGeom.Camera: 

136 # Return something that can be indexed by detector number 

137 # but also has to support getIdIter. 

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

139 return makeCamera(local_file.ospath) 

140 

141 def register(self, registry: Registry, *, update: bool = False) -> None: 

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

143 `~lsst.daf.butler.Registry`. 

144 """ 

145 detector_max = 2 

146 dataId = { 

147 "instrument": self.getName(), 

148 "class_name": get_full_type_name(DummyCam), 

149 "detector_max": detector_max, 

150 "visit_max": 1_000_000, 

151 "exposure_max": 1_000_000, 

152 } 

153 

154 with registry.transaction(): 

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

156 self._registerFilters(registry, update=update) 

157 for d in range(detector_max): 

158 registry.syncDimensionData( 

159 "detector", 

160 dict( 

161 instrument=self.getName(), 

162 id=d, 

163 full_name=f"RXX_S0{d}", 

164 ), 

165 update=update, 

166 ) 

167 

168 def getRawFormatter(self, dataId: DataId) -> type[Formatter | FormatterV2]: 

169 # Docstring inherited from Instrument.getRawFormatter. 

170 return DummyCamYamlWcsFormatter 

171 

172 def applyConfigOverrides(self, name: str, config: lsst.pex.config.Config) -> None: 

173 pass 

174 

175 def writeAdditionalCuratedCalibrations( 

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

177 ) -> None: 

178 # We want to test the standard curated calibration ingest 

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

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

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

182 # method but call the standard curated calibration code. 

183 if collection is None: 

184 collection = self.makeCalibrationCollectionName(*labels) 

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

186 

187 datasetType = DatasetType( 

188 "testCalib", 

189 universe=butler.dimensions, 

190 isCalibration=True, 

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

192 storageClass="CuratedCalibration", 

193 ) 

194 runs: set[str] = set() 

195 self._writeSpecificCuratedCalibrationDatasets( 

196 butler, datasetType, collection, runs=runs, labels=labels 

197 ) 

198 

199 

200@dataclasses.dataclass 

201class InstrumentTestData: 

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

203 

204 name: str 

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

206 

207 nDetectors: int 

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

209 

210 firstDetectorName: str 

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

212 

213 physical_filters: set[str] 

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

215 

216 

217class InstrumentTests(metaclass=abc.ABCMeta): 

218 """Tests of subclasses of Instrument. 

219 

220 TestCase subclasses must derive from this, then `unittest.TestCase`, and 

221 override ``data`` and ``instrument``. 

222 """ 

223 

224 if TYPE_CHECKING: 

225 # When run will also inherit from unittest.TestCase. 

226 assertEqual: Callable 

227 assertFalse: Callable 

228 assertIn: Callable 

229 assertGreaterEqual: Callable 

230 

231 data: ClassVar[InstrumentTestData | None] = None 

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

233 

234 instrument: ClassVar[Instrument | None] = None 

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

236 

237 def test_name(self) -> None: 

238 assert self.instrument is not None, "Subclass must initialize instrument property" 

239 assert self.data is not None, "Subclass must initialize data property" 

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

241 

242 def test_getCamera(self) -> None: 

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

244 assert self.instrument is not None, "Subclass must initialize instrument property" 

245 assert self.data is not None, "Subclass must initialize data property" 

246 camera = self.instrument.getCamera() 

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

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

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

250 

251 def test_register(self) -> None: 

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

253 assert self.instrument is not None, "Subclass must initialize instrument property" 

254 assert self.data is not None, "Subclass must initialize data property" 

255 with create_populated_sqlite_registry() as butler: 

256 registry = butler.registry 

257 # Check that the registry starts out empty. 

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

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

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

261 

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

263 self.instrument.register(registry) 

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

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

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

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

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

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

270 detectorNames = { 

271 dataId.records["detector"].full_name # type: ignore[union-attr] 

272 for dataId in detectorDataIds 

273 } 

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

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

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

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

278 

279 # Check that the instrument class can be retrieved. 

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

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

282 

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

284 self.instrument.register(registry)