Coverage for python / lsst / obs / base / _read_curated_calibs.py: 9%
87 statements
« prev ^ index » next coverage.py v7.13.1, created at 2026-01-20 01:41 -0800
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# (http://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/>.
22from __future__ import annotations
24__all__ = ["CuratedCalibration", "read_all"]
26import datetime
27import glob
28import os
29from collections.abc import Mapping
30from typing import TYPE_CHECKING, Any, Protocol
32if TYPE_CHECKING:
33 import lsst.afw.cameraGeom
36class CuratedCalibration(Protocol):
37 """Protocol that describes the methods needed by this class when dealing
38 with curated calibration datasets.
39 """
41 @classmethod
42 def readText(cls, path: str) -> CuratedCalibration: ...
44 def getMetadata(self) -> Mapping: ...
47def read_one_calib(
48 path: tuple[str, ...],
49 chip_id: int | None,
50 filter_name: str | None,
51 calib_class: type[CuratedCalibration],
52) -> tuple[dict[datetime.datetime, CuratedCalibration], str]:
53 """Read data for a particular path from the standard format at a
54 particular root.
56 Parameters
57 ----------
58 path : `tuple` [`str`]
59 This tuple contains the top level of the data tree at index=0,
60 and then further optional subdirectories in subsequent
61 indices. See Notes below for more details.
62 chip_id : `int` or None
63 The identifier for the sensor in question. To be used in
64 validation.
65 filter_name : `str` or None
66 The identifier for the filter in question. To be used in
67 validation.
68 calib_class : `Any`
69 The class to use to read the curated calibration text file. Must
70 support the ``readText()`` method.
72 Returns
73 -------
74 data_dict : `dict`
75 A dictionary of calibration objects constructed from the
76 appropriate factory class. The key is the validity start time
77 as a `datetime.datetime` object.
78 calib_type : `str`
79 The type of calibrations that have been read and are included
80 in the ``data_dict``.
82 Notes
83 -----
84 Curated calibrations are read from the appropriate ``obs_ _data``
85 package, and are required to have a common directory structure to
86 be identified and ingested properly. The top-level directories
87 are organized by the instrument's ``policyName``. These names are
88 generally all lower-case, but this is not universally true.
90 Below the top-level instrument directory, subdirectories named
91 after the curated calibration type contained within, with the
92 dataset_type_name forced to lowercase. For calibrations that
93 depend on the detector (i.e., the defects), the next level of
94 subdirectories should contain directories named with the detector
95 name, again forced to lowercase.
97 For filter dependent calibrations that do not depend on the
98 detector (i.e., transmission_filter), the calibrations should be
99 grouped into directories named with the physical filter name
100 (again, all lowercase) below the dataset_type_name directory.
101 Filter dependent calibrations that do depend on the detector
102 (i.e., transmission_system), have physical filter named
103 directories below the detector level directories.
105 The files themselves are required to be named with the validity start
106 date and the appropriate file extension. The date must be in an ISO format
107 that can be parsed by `datetime.datetime.fromisoformat`. This can include
108 variants commonly used data packages such as ``2025-04-30T12:23:00`` or the
109 more compact ``20250430T123000``.
110 """
111 files = []
112 extensions = (".ecsv", ".yaml", ".json")
113 for ext in extensions:
114 files.extend(glob.glob(os.path.join(*path, f"*{ext}")))
116 parts = os.path.split(path[0])
117 instrument = os.path.split(parts[0])[1] # convention is that these reside at <instrument>/<calib_type>
118 calib_type = parts[1]
119 data_dict: dict[datetime.datetime, Any] = {}
120 for f in files:
121 date_str = os.path.splitext(os.path.basename(f))[0]
122 # Assume files are using some form of ISO date string.
123 valid_start = datetime.datetime.fromisoformat(date_str)
124 data_dict[valid_start] = calib_class.readText(f)
125 check_metadata(data_dict[valid_start], valid_start, instrument, chip_id, filter_name, f, calib_type)
126 return data_dict, calib_type
129def check_metadata(
130 obj: Any,
131 valid_start: datetime.datetime,
132 instrument: str,
133 chip_id: int | None,
134 filter_name: str | None,
135 filepath: str,
136 calib_type: str,
137) -> None:
138 """Check that the metadata is complete and self consistent.
140 Parameters
141 ----------
142 obj : object of same type as the factory
143 Object to retrieve metadata from in order to compare with
144 metadata inferred from the path.
145 valid_start : `datetime`
146 Start of the validity range for data.
147 instrument : `str`
148 Name of the instrument in question.
149 chip_id : `int`
150 Identifier of the sensor in question.
151 filter_name : `str`
152 Identifier of the filter in question.
153 filepath : `str`
154 Path of the file read to construct the data.
155 calib_type : `str`
156 Name of the type of data being read.
158 Returns
159 -------
160 None
162 Raises
163 ------
164 ValueError
165 If the metadata from the path and the metadata encoded
166 in the path do not match for any reason.
167 """
168 md = obj.getMetadata()
169 # It is an error if these two do not exist.
170 finst = md["INSTRUME"]
171 fcalib_type = md["OBSTYPE"]
172 # These may optionally not exist.
173 fchip_id = md.get("DETECTOR", None)
174 ffilter_name = md.get("FILTER", None)
176 if chip_id is not None:
177 fchip_id = int(fchip_id)
178 if filter_name is not None:
179 ffilter_name = ffilter_name.lower()
180 filter_name = filter_name.lower()
182 if not (
183 (finst.lower(), fchip_id, ffilter_name, fcalib_type.lower())
184 == (instrument.lower(), chip_id, filter_name, calib_type.lower())
185 ):
186 raise ValueError(
187 "Path and file metadata do not agree:\n"
188 f"Path metadata: {instrument} {chip_id} {filter_name} {calib_type}\n"
189 f"File metadata: {finst} {fchip_id} {ffilter_name} {fcalib_type}\n"
190 f"File read from : {filepath}\n"
191 )
194def read_all(
195 root: str,
196 camera: lsst.afw.cameraGeom.Camera,
197 calib_class: type[CuratedCalibration],
198 required_dimensions: list[str],
199 filters: set[str],
200) -> tuple[dict[tuple[str, ...], dict[datetime.datetime, CuratedCalibration]], str]:
201 """Read all data from the standard format at a particular root.
203 Parameters
204 ----------
205 root : `str`
206 Path to the top level of the data tree. This is expected to hold
207 directories named after the sensor names. They are expected to be
208 lower case.
209 camera : `lsst.afw.cameraGeom.Camera`
210 The camera that goes with the data being read.
211 calib_class : `Any`
212 The class to use to read the curated calibration text file. Must
213 support the ``readText()`` and ``getMetadata()`` methods.
214 required_dimensions : `list` [`str`]
215 Dimensions required for the calibration.
216 filters : `list` [`str`]
217 List of the known filters for this camera. Used to identify
218 filter-dependent calibrations.
220 Returns
221 -------
222 calibration_data : `dict`
223 A dictionary of dictionaries of calibration objects
224 constructed with the appropriate factory class. The first key
225 is the sensor name in lower case, and the second is the
226 validity start time as a `datetime.datetime` object.
227 calib_type : `str`
228 The type of calibrations that have been read and are included
229 in the ``data_dict``.
231 Notes
232 -----
233 Each leaf object in the constructed dictionary has metadata associated with
234 it. The detector ID may be retrieved from the DETECTOR entry of that
235 metadata.
236 """
237 calibration_data = {}
239 root = os.path.normpath(root)
240 dirs = os.listdir(root) # assumes all directories contain data
241 dirs = [d for d in dirs if os.path.isdir(os.path.join(root, d))]
242 if not dirs:
243 dirs = [root]
245 calib_types = set()
246 # We assume the directories have been lowered.
247 detector_map = {det.getName().lower(): det.getName() for det in camera}
248 filter_map = {filterName.lower().replace(" ", "_"): filterName for filterName in filters}
250 paths_to_search: list[tuple[str, ...]] = []
251 for d in dirs:
252 dir_name = os.path.basename(d)
253 # Catch possible mistakes:
254 if "detector" in required_dimensions:
255 if dir_name not in detector_map:
256 # Top level directories must be detectors if they're
257 # required.
258 detectors = list(detector_map)
259 max_detectors = 10
260 note_str = "knows"
261 if len(detectors) > max_detectors:
262 # report example subset
263 note_str = "examples"
264 detectors = detectors[:max_detectors]
265 raise RuntimeError(
266 f"Detector {dir_name} not known to supplied camera "
267 f"{camera.getName()} ({note_str}: {','.join(detectors)})"
268 )
269 elif "physical_filter" in required_dimensions:
270 # If the calibration depends on both detector and
271 # physical_filter, the subdirs here should contain the
272 # filter name.
273 subdirs = os.listdir(os.path.join(root, dir_name))
274 subdirs = [d for d in subdirs if os.path.isdir(os.path.join(root, dir_name, d))]
275 for sd in subdirs:
276 subdir_name = os.path.basename(sd)
277 if subdir_name not in filter_map:
278 raise RuntimeError(f"Filter {subdir_name} not known to supplied camera.")
279 else:
280 paths_to_search.append((root, dir_name, subdir_name))
281 else:
282 paths_to_search.append((root, dir_name))
283 elif "physical_filter" in required_dimensions:
284 # If detector is not required, but physical_filter is,
285 # then the top level should contain the filter
286 # directories.
287 if dir_name not in filter_map:
288 raise RuntimeError(f"Filter {dir_name} not known to supplied camera.")
289 paths_to_search.append((root, dir_name))
290 else:
291 # Neither detector nor physical_filter are required, so
292 # the calibration is global, and will not be found in
293 # subdirectories.
294 paths_to_search.append((root,))
296 for path in paths_to_search:
297 chip_id = None
298 filter_name = None
299 if "detector" in required_dimensions:
300 chip_id = camera[detector_map[path[1]]].getId()
301 if "physical_filter" in required_dimensions:
302 filter_name = filter_map[path[-1]]
304 calibration_data[path], calib_type = read_one_calib(path, chip_id, filter_name, calib_class)
306 calib_types.add(calib_type)
307 if len(calib_types) != 1: # set.add(None) has length 1 so None is OK here.
308 raise ValueError(f"Error mixing calib types: {calib_types}")
310 no_data = all(v == {} for v in calibration_data.values())
311 if no_data:
312 raise RuntimeError("No data to ingest")
314 return calibration_data, calib_type