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

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 @classmethod 

136 @lru_cache # For mypy 

137 def getObsDataPackageRoot(cls) -> ResourcePath | None: 

138 if cls.dataPackageDir is None: 

139 return None 

140 return ResourcePath(cls.dataPackageDir) 

141 

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

143 # Return something that can be indexed by detector number 

144 # but also has to support getIdIter. 

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

146 return makeCamera(local_file.ospath) 

147 

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

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

150 `~lsst.daf.butler.Registry`. 

151 """ 

152 detector_max = 2 

153 dataId = { 

154 "instrument": self.getName(), 

155 "class_name": get_full_type_name(DummyCam), 

156 "detector_max": detector_max, 

157 "visit_max": 1_000_000, 

158 "exposure_max": 1_000_000, 

159 } 

160 

161 with registry.transaction(): 

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

163 self._registerFilters(registry, update=update) 

164 for d in range(detector_max): 

165 registry.syncDimensionData( 

166 "detector", 

167 dict( 

168 instrument=self.getName(), 

169 id=d, 

170 full_name=f"RXX_S0{d}", 

171 ), 

172 update=update, 

173 ) 

174 

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

176 # Docstring inherited from Instrument.getRawFormatter. 

177 return DummyCamYamlWcsFormatter 

178 

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

180 pass 

181 

182 def writeAdditionalCuratedCalibrations( 

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

184 ) -> None: 

185 # We want to test the standard curated calibration ingest 

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

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

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

189 # method but call the standard curated calibration code. 

190 if collection is None: 

191 collection = self.makeCalibrationCollectionName(*labels) 

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

193 

194 datasetType = DatasetType( 

195 "testCalib", 

196 universe=butler.dimensions, 

197 isCalibration=True, 

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

199 storageClass="CuratedCalibration", 

200 ) 

201 runs: set[str] = set() 

202 self._writeSpecificCuratedCalibrationDatasets( 

203 butler, datasetType, collection, runs=runs, labels=labels 

204 ) 

205 

206 

207@dataclasses.dataclass 

208class InstrumentTestData: 

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

210 

211 name: str 

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

213 

214 nDetectors: int 

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

216 

217 firstDetectorName: str 

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

219 

220 physical_filters: set[str] 

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

222 

223 

224class InstrumentTests(metaclass=abc.ABCMeta): 

225 """Tests of subclasses of Instrument. 

226 

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

228 override ``data`` and ``instrument``. 

229 """ 

230 

231 if TYPE_CHECKING: 

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

233 assertEqual: Callable 

234 assertFalse: Callable 

235 assertIn: Callable 

236 assertGreaterEqual: Callable 

237 

238 data: ClassVar[InstrumentTestData | None] = None 

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

240 

241 instrument: ClassVar[Instrument | None] = None 

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

243 

244 def test_name(self) -> None: 

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

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

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

248 

249 def test_getCamera(self) -> None: 

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

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

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

253 camera = self.instrument.getCamera() 

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

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

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

257 

258 def test_register(self) -> None: 

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

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

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

262 with create_populated_sqlite_registry() as butler: 

263 registry = butler.registry 

264 # Check that the registry starts out empty. 

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

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

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

268 

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

270 self.instrument.register(registry) 

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

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

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

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

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

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

277 detectorNames = { 

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

279 for dataId in detectorDataIds 

280 } 

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

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

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

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

285 

286 # Check that the instrument class can be retrieved. 

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

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

289 

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

291 self.instrument.register(registry)