Coverage for python / astro_metadata_translator / bin / translate.py: 14%
112 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-24 01:11 -0800
1# This file is part of astro_metadata_translator.
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 LICENSE 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.
12"""Implementation of the ``astrometadata translate`` command-line.
14Read file metadata from the specified files and report the translated content.
15"""
17from __future__ import annotations
19__all__ = ("translate_or_dump_headers",)
21import logging
22import math
23import traceback
24from collections import defaultdict
25from collections.abc import Sequence
26from typing import IO, TYPE_CHECKING, Any
28import astropy.time
29import astropy.units as u
30import yaml
31from astropy.table import Column, MaskedColumn, QTable
32from lsst.resources import ResourcePath
34from astro_metadata_translator import MetadataTranslator, ObservationInfo, fix_header
36from ..file_helpers import find_files, read_basic_metadata_from_file
37from ..properties import PROPERTIES
39if TYPE_CHECKING:
40 from lsst.resources import ResourcePathExpression
42log = logging.getLogger(__name__)
44# Output mode choices
45OUTPUT_MODES = ("auto", "verbose", "table", "yaml", "fixed", "yamlnative", "fixednative", "none")
47# Number of rows per table page.
48# This is a minimum given that DECam data files can include multiple headers.
49MAX_TABLE_PAGE_SIZE = 500
51# Definitions for table columns
52TABLE_COLUMNS = (
53 {"format": "<", "attr": "observation_id", "label": "ObsId"},
54 {
55 "format": "<",
56 "attr": "observation_type",
57 "label": "ImgType",
58 },
59 {
60 "format": "<",
61 "attr": "object",
62 "label": "Object",
63 },
64 {
65 "format": "<",
66 "attr": "physical_filter",
67 "label": "Filter",
68 },
69 {"format": ">8.8s", "attr": "detector_unique_name", "label": "Detector"},
70 {
71 "format": None,
72 "attr": "exposure_time",
73 "label": "ExpTime",
74 "bad": math.nan * u.s,
75 },
76 {"attr": "datetime_begin", "label": "Observation Date", "bad": astropy.time.Time(0.0, format="jd")},
77)
80def read_file(
81 file: ResourcePathExpression,
82 hdrnum: int,
83 print_trace: bool,
84 output_columns: defaultdict[str, list],
85 outstream: IO | None = None,
86 output_mode: str = "verbose",
87 write_heading: bool = False,
88) -> bool:
89 """Read the specified file and process it.
91 Parameters
92 ----------
93 file : `str` or `lsst.resources.ResourcePathExpression`
94 The file from which the header is to be read.
95 hdrnum : `int`
96 The HDU number to read. The primary header is always read and
97 merged with the header from this HDU.
98 print_trace : `bool`
99 If there is an error reading the file and this parameter is `True`,
100 a full traceback of the exception will be reported. If `False` prints
101 a one line summary of the error condition.
102 output_columns : `collections.defaultdict` [ `str`, `list` ]
103 For table mode, a place to store the columns.
104 outstream : `io.StringIO`, optional
105 Output stream to use for standard messages. Defaults to `None` which
106 uses the default output stream.
107 output_mode : `str`, optional
108 Output mode to use. Must be one of "verbose", "none", "table",
109 "yaml", or "fixed". "yaml" and "fixed" can be modified with a
110 "native" suffix to indicate that the output should be a representation
111 of the native object type representing the header (which can be
112 PropertyList or an Astropy header). Without this modify headers
113 will be dumped as simple `dict` form.
114 "auto" is used to indicate that a single file has been specified
115 but the output will depend on whether the file is a multi-extension
116 FITS file or not.
117 write_heading : `bool`, optional
118 If `True` and in table mode, write a table heading out before writing
119 the content.
121 Returns
122 -------
123 success : `bool`
124 `True` if the file was handled successfully, `False` if the file
125 could not be processed.
126 """
127 if output_mode not in OUTPUT_MODES:
128 raise ValueError(f"Output mode of '{output_mode}' is not understood.")
130 # This gets in the way in tabular mode
131 if output_mode != "table":
132 log.info("Analyzing %s...", file)
134 uri = ResourcePath(file, forceDirectory=False)
135 try:
136 md = read_basic_metadata_from_file(uri, hdrnum, can_raise=True)
137 if md is None:
138 raise RuntimeError(f"Failed to read file {uri} HDU={hdrnum}")
140 if output_mode.endswith("native"):
141 # Strip native and don't change type of md
142 output_mode = output_mode[: -len("native")]
143 else:
144 # Rewrite md as simple dict for output
145 md = dict(md.items())
147 if output_mode in ("yaml", "fixed"):
148 if output_mode == "fixed":
149 fix_header(md, filename=str(uri))
151 # The header should be written out in the insertion order
152 print(yaml.dump(md, sort_keys=False), file=outstream)
153 return True
155 # Try to work out a translator class.
156 translator_class = MetadataTranslator.determine_translator(md, filename=str(uri))
158 # Work out which headers to translate, assuming the default if
159 # we have a YAML test file.
160 if uri.getExtension() == ".yaml":
161 headers = [md]
162 else:
163 headers = list(translator_class.determine_translatable_headers(uri, md))
164 if output_mode == "auto":
165 output_mode = "table" if len(headers) > 1 else "verbose"
167 if output_mode == "table" and output_columns is None:
168 raise ValueError("Table mode requires output columns")
170 for md in headers:
171 obs_info = ObservationInfo(md, pedantic=False, filename=str(uri))
172 if output_mode == "table":
173 for c in TABLE_COLUMNS:
174 output_columns[c["label"]].append(getattr(obs_info, c["attr"]))
175 elif output_mode == "verbose":
176 print(f"{obs_info}", file=outstream)
177 elif output_mode == "none":
178 pass
179 else:
180 raise RuntimeError(f"Output mode of '{output_mode}' not recognized but should be known.")
182 except Exception as e:
183 if print_trace:
184 traceback.print_exc(file=outstream)
185 else:
186 print(f"Failure processing {uri}: {e}", file=outstream)
187 return False
188 return True
191def _fill_bad_values(value: Any, fillvalue: Any) -> Any:
192 """Convert None values to the fill value.
194 Parameters
195 ----------
196 value : `~typing.Any`
197 Value to check.
198 fillvalue : `~typing.Any`
199 Value to use if `None`.
201 Returns
202 -------
203 filled : `~typing.Any`
204 Original value or the fill value.
205 """
206 return value if value is not None else fillvalue
209def _dump_columns(output_columns: dict[str, list], outstream: IO | None = None) -> None:
210 """Write columns to output stream as a table.
212 Parameters
213 ----------
214 output_columns : `dict` [`str`, `list`]
215 The columns to be written, indexed by column name.
216 outstream : `io.StringIO` or `None`, optional
217 Output stream to use for standard messages. Defaults to `None` which
218 uses the default output stream.
219 """
220 if not output_columns:
221 return
223 qt = QTable()
224 for c in TABLE_COLUMNS:
225 data = output_columns[c["label"]]
226 # If the column has no content, assume no columns have content and
227 # abandon table output.
228 if not len(data):
229 return
231 need_mask = False
232 mask = []
233 for v in data:
234 is_none = v is None
235 if is_none:
236 need_mask = True
237 mask.append(is_none)
238 col_format = c.get("format")
240 if need_mask:
241 data = [_fill_bad_values(v, c.get("bad", "-")) for v in output_columns[c["label"]]]
243 # Quantity have to be handled in special way since they need
244 # to be merged into a single entity before they can be stored
245 # in a column.
246 if issubclass(PROPERTIES[c["attr"]].py_type, u.Quantity):
247 data = u.Quantity(data)
248 elif issubclass(PROPERTIES[c["attr"]].py_type, astropy.time.Time):
249 # Force to ISO string.
250 data = astropy.time.Time(data).isot
252 if need_mask:
253 data = MaskedColumn(name=c["label"], data=data, mask=mask, format=col_format)
254 else:
255 data = Column(data=data, name=c["label"], format=col_format)
256 qt[c["label"]] = data
258 print("\n".join(qt.pformat(max_lines=-1, max_width=-1)), file=outstream)
261def translate_or_dump_headers(
262 files: Sequence[str],
263 regex: str,
264 hdrnum: int,
265 print_trace: bool,
266 outstream: IO | None = None,
267 output_mode: str = "auto",
268) -> tuple[list[str], list[str]]:
269 """Read and translate metadata from the specified files.
271 Parameters
272 ----------
273 files : iterable of `str`
274 The files or directories from which the headers are to be read.
275 regex : `str`
276 Regular expression string used to filter files when a directory is
277 scanned.
278 hdrnum : `int`
279 The HDU number to read. The primary header is always read and
280 merged with the header from this HDU.
281 print_trace : `bool`
282 If there is an error reading the file and this parameter is `True`,
283 a full traceback of the exception will be reported. If `False` prints
284 a one line summary of the error condition.
285 outstream : `io.StringIO` or `None`, optional
286 Output stream to use for standard messages. Defaults to `None` which
287 uses the default output stream.
288 output_mode : `str`, optional
289 Output mode to use for the translated information.
290 "auto" switches based on how many files are found.
292 Returns
293 -------
294 okay : `list` of `str`
295 All the files that were processed successfully.
296 failed : `list` of `str`
297 All the files that could not be processed.
298 """
299 found_files = find_files(files, regex)
301 # Convert "auto" to correct mode but for a single file keep it
302 # auto in case that file has multiple headers
303 if output_mode == "auto":
304 if len(found_files) > 1:
305 output_mode = "table"
307 # Process each file
308 failed = []
309 okay = []
310 heading = True
311 output_columns: defaultdict[str, list] = defaultdict(list)
312 for path in sorted(found_files):
313 isok = read_file(path, hdrnum, print_trace, output_columns, outstream, output_mode, heading)
314 heading = False
315 if isok:
316 okay.append(str(path))
317 else:
318 failed.append(str(path))
320 # Check if we have exceeded the page size and should dump the table.
321 if output_mode == "table" and len(output_columns[TABLE_COLUMNS[0]["label"]]) >= MAX_TABLE_PAGE_SIZE:
322 _dump_columns(output_columns, outstream)
323 output_columns = defaultdict(list)
325 if output_columns:
326 _dump_columns(output_columns, outstream)
328 return okay, failed