Coverage for python/lsst/images/tests/extract_obs_info_test_data.py: 37%

1# This file is part of lsst-images. 

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# 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"""Extract the ObservationInfo conversion test fixtures from various CI 

13test data packages. 

14 

15Run from a development checkout with:: 

16 

17 lsst-images-admin extract-obs-info-test-data 

18 

19For this script to run, the following packages must be set up: 

20 

21 - ``obs_lsst`` 

22 - ``obs_decam`` 

23 - ``obs_subaru`` 

24 - ``testdata_ci_hsc`` 

25 - ``testdata_ci_lsstcam_m49`` 

26 - ``ap_verify_ci_hits2015`` 

27 

28Raw files are located with ``lsst.utils.getPackageDir``, so this requires an 

29EUPS-managed environment, or the ``AP_VERIFY_CI_HITS2015_DIR``, 

30``TESTDATA_CI_HSC_DIR`` and ``TESTDATA_CI_LSSTCAM_M49_DIR`` environment 

31variables set explicitly. 

32""" 

33 

34from __future__ import annotations 

35 

36__all__ = () 

37 

38import dataclasses 

39from pathlib import Path 

40from typing import TYPE_CHECKING 

41 

42import click 

43from astro_metadata_translator import ObservationInfo 

44 

45from ..cameras import Detector 

46from ..serialization import write_archive 

47 

48if TYPE_CHECKING: 

49 import astropy.io.fits 

50 

51_REPO_ROOT = Path(__file__).parents[4] 

52_HEADERS_DIR = _REPO_ROOT / "tests" / "data" / "headers" 

53_OBS_INFO_DIR = _REPO_ROOT / "tests" / "data" / "obs_info" 

54_DETECTORS_DIR = _REPO_ROOT / "tests" / "data" / "detectors" 

55 

56 

57@dataclasses.dataclass(frozen=True) 

58class FixtureCase: 

59 """The provenance and generation inputs for one fixture case.""" 

60 

61 package: str 

62 """EUPS test data package containing the raw file.""" 

63 

64 raw_path: str 

65 """Raw FITS file path within the test data package.""" 

66 

67 original_filename: str 

68 """Original observation filename passed to 

69 ``ObservationInfo.from_header``. 

70 """ 

71 instrument_class: str 

72 """Full type name of the Instrument class used to load camera geometry.""" 

73 

74 extensions: tuple[int, ...] 

75 """Non-primary HDU indexes whose headers are merged (INHERIT-aware) into 

76 the primary header.. 

77 """ 

78 

79 @property 

80 def full_raw_path(self) -> Path: 

81 from lsst.utils import getPackageDir 

82 

83 return Path(getPackageDir(self.package)) / self.raw_path 

84 

85 

86_CASES = { 

87 "decam-c4d_150218_052850-N25": FixtureCase( 

88 package="ap_verify_ci_hits2015", 

89 raw_path="raw/c4d_150218_052850_ori.fits.fz", 

90 original_filename="c4d_150218_052850_ori.fits.fz", 

91 instrument_class="lsst.obs.decam.DarkEnergyCamera", 

92 extensions=(1,), # the N25 CCD. 

93 ), 

94 "hsc-HSCA90333426": FixtureCase( 

95 package="testdata_ci_hsc", 

96 raw_path="raw/HSCA90333426.fits", 

97 original_filename="HSCA90333426.fits", 

98 instrument_class="lsst.obs.subaru.HyperSuprimeCam", 

99 extensions=(1,), # the file's only (detector) extension. 

100 ), 

101 "lsstcam-MC_O_20250501_000292_R21_S11": FixtureCase( 

102 package="testdata_ci_lsstcam_m49", 

103 raw_path="LSSTCam/raw/all/raw/20250501/MC_O_20250501_000292/MC_O_20250501_000292_R21_S11.fits", 

104 original_filename="MC_O_20250501_000292_R21_S11.fits", 

105 instrument_class="lsst.obs.lsst.LsstCam", 

106 extensions=(17,), # the REB_COND table; segment images are shared 

107 # across the group's detectors and the detector identity lives in the 

108 # primary header (via the filename). 

109 ), 

110} 

111 

112 

113def extract_header(raw_path: Path, extensions: tuple[int, ...], output_path: Path) -> astropy.io.fits.Header: 

114 """Write the pipeline-processed FITS header of one raw file. 

115 

116 

117 Parameters 

118 ---------- 

119 raw_path : `pathlib.Path` 

120 The raw FITS file to read. 

121 extensions : `tuple` [ `int` ] 

122 Non-primary HDU indexes whose headers are merged into the primary 

123 header. 

124 output_path : `pathlib.Path` 

125 The ``.hdr`` text file to write. 

126 

127 Returns 

128 ------- 

129 header : `astropy.io.fits.Header` 

130 The processed header. 

131 """ 

132 from astro_metadata_translator import MetadataTranslator, fix_header, merge_headers 

133 from astro_metadata_translator.file_helpers import read_basic_metadata_from_file 

134 

135 from lsst.daf.base import PropertyList 

136 

137 def read_metadata(ext: int) -> PropertyList: 

138 md = read_basic_metadata_from_file(str(raw_path), ext, can_raise=True) 

139 assert isinstance(md, PropertyList) 

140 return md 

141 

142 headers: list[PropertyList] = [read_metadata(0)] 

143 if extensions: 

144 headers.extend(read_metadata(ext) for ext in extensions) 

145 else: 

146 # Read the first extension too, as RawIngestTask.extractMetadata 

147 # does when the primary header alone can not identify a translator. 

148 try: 

149 MetadataTranslator.determine_translator(headers[0], filename=str(raw_path)) 

150 except ValueError: 

151 headers.append(read_metadata(1)) 

152 merged = merge_headers(headers, mode="overwrite") 

153 # merge_headers preserves the type of the headers it is given. 

154 assert isinstance(merged, PropertyList) 

155 header = merged.to_astropy_header() 

156 translator_class = MetadataTranslator.determine_translator(header, filename=str(raw_path)) 

157 fix_header(header, translator_class=translator_class, filename=str(raw_path)) 

158 # fix_header records when the corrections were applied; that is fixture 

159 # build bookkeeping, not observation metadata, and would make 

160 # regeneration non-deterministic. 

161 fix_date = "HIERARCH ASTRO METADATA FIX DATE" 

162 if fix_date in header: 

163 del header[fix_date] 

164 header.totextfile(str(output_path), endcard=True, overwrite=True) 

165 # Strip card padding and end with a newline, exactly as the repo's 

166 # trailing-whitespace and EOF-fixer pre-commit hooks would, so 

167 # regeneration does not show phantom churn against the committed file. 

168 text = output_path.read_text() 

169 output_path.write_text("\n".join(line.rstrip() for line in text.splitlines()) + "\n") 

170 return header 

171 

172 

173def extract_obs_info( 

174 header: astropy.io.fits.Header, 

175 original_filename: str, 

176 output_path: Path, 

177) -> ObservationInfo: 

178 """Write the ``ObservationInfo`` JSON wire form of one header. 

179 

180 Parameters 

181 ---------- 

182 header : `astropy.io.fits.Header` 

183 The processed header to translate. 

184 original_filename : `str` 

185 The original observation filename, passed to 

186 ``ObservationInfo.from_header`` so correction YAMLs apply. 

187 output_path : `pathlib.Path` 

188 The JSON file to write the serialized `ObservationInfo` to. 

189 

190 Returns 

191 ------- 

192 obs_info : `~astro_metadata_translator.ObservationInfo` 

193 The translated observation metadata. 

194 """ 

195 obs_info = ObservationInfo.from_header(header, filename=original_filename, quiet=True) 

196 output_path.write_text(obs_info.model_dump_json(indent=2) + "\n") 

197 return obs_info 

198 

199 

200def extract_detector( 

201 instrument_class_name: str, 

202 instrument: str, 

203 detector_id: int, 

204 output_path: Path, 

205) -> None: 

206 """Write the serialized detector archive of one fixture case. 

207 

208 The detector is converted from the real instrument camera's detector to 

209 an `lsst.images.cameras.Detector` archive: a small, diffable JSON file, 

210 unlike afw camera FITS files, which drag along the camera's entire 

211 shared transform graph. 

212 

213 Parameters 

214 ---------- 

215 instrument_class_name : `str` 

216 Fully-qualified instrument class name, e.g. ``lsst.obs.lsst.LsstCam``. 

217 instrument : `str` 

218 Instrument name to record on the detector. 

219 detector_id : `int` 

220 Identifier of the detector to take from the instrument camera. 

221 output_path : `pathlib.Path` 

222 The JSON archive file to write. 

223 """ 

224 from lsst.obs.base import Instrument 

225 

226 instrument_obj = Instrument.from_string(instrument_class_name) 

227 legacy_detector = instrument_obj.getCamera()[detector_id] 

228 detector = Detector.from_legacy(legacy_detector, instrument=instrument) 

229 write_archive(detector, output_path) 

230 # End with a newline, as the repo's EOF-fixer pre-commit hook requires, 

231 # so regeneration does not show phantom churn against the committed file. 

232 with open(output_path, "a") as fh: 

233 fh.write("\n") 

234 

235 

236@click.command("extract_obs_info_test_data") 

237@click.option( 

238 "--testdata-dir", 

239 type=Path, 

240 default=None, 

241 help="Directory containing unpacked test-data packages, used when they" 

242 " are not set up in the current EUPS environment.", 

243) 

244@click.option( 

245 "-H", 

246 "--headers-dir", 

247 type=Path, 

248 default=_HEADERS_DIR, 

249 show_default=True, 

250 help="Directory to write the .hdr header fixtures to.", 

251) 

252@click.option( 

253 "-o", 

254 "--output-dir", 

255 type=Path, 

256 default=_OBS_INFO_DIR, 

257 show_default=True, 

258 help="Directory to write the ObservationInfo JSON fixtures to.", 

259) 

260@click.option( 

261 "-d", 

262 "--detectors-dir", 

263 type=Path, 

264 default=_DETECTORS_DIR, 

265 show_default=True, 

266 help="Directory to write the Detector JSON archives to.", 

267) 

268def extract_obs_info_test_data( 

269 testdata_dir: Path | None, 

270 headers_dir: Path, 

271 output_dir: Path, 

272 detectors_dir: Path, 

273) -> None: # numpydoc ignore=PR01 

274 """Regenerate legacy-conversion fixtures from raw test data.""" 

275 try: 

276 # The LSSTCam fixture's metadata translator is an obs_lsst plugin, 

277 # which astro_metadata_translator does not load on its own; importing 

278 # the package registers it. 

279 import lsst.obs.lsst # noqa: F401 

280 except ImportError as err: 

281 err.add_note( 

282 "Regenerating these fixtures requires a full Rubin development" 

283 " environment with at least 'obs_decam', 'obs_lsst' and" 

284 " 'obs_subaru' importable, and the fixture cases' test-data" 

285 " packages available. This is not necessary for just running the" 

286 " tests." 

287 ) 

288 raise 

289 headers_dir.mkdir(parents=True, exist_ok=True) 

290 output_dir.mkdir(parents=True, exist_ok=True) 

291 detectors_dir.mkdir(parents=True, exist_ok=True) 

292 for stem, case in _CASES.items(): 

293 header_path = headers_dir / (stem + ".hdr") 

294 header = extract_header(case.full_raw_path, case.extensions, header_path) 

295 click.echo(f"Wrote {header_path}") 

296 obs_info_path = output_dir / (stem + ".json") 

297 obs_info = extract_obs_info(header, case.original_filename, obs_info_path) 

298 click.echo(f"Wrote {obs_info_path}") 

299 detector_path = detectors_dir / (stem + ".json") 

300 assert obs_info.detector_num is not None 

301 extract_detector( 

302 case.instrument_class, obs_info.instrument or "", obs_info.detector_num, detector_path 

303 ) 

304 click.echo(f"Wrote {detector_path}")