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