Coverage for python / astro_metadata_translator / cli / astrometadata.py: 31%
130 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-17 08:50 +0000
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.
12from __future__ import annotations
14__all__ = ("main",)
16import importlib
17import logging
18import os
19from collections.abc import Sequence
20from importlib.metadata import entry_points
22import click
24from ..bin.translate import translate_or_dump_headers
25from ..bin.writeindex import write_index_files
26from ..bin.writesidecar import write_sidecar_files
28# Default regex for finding data files
29re_default = r"\.fit[s]?\b"
31log = logging.getLogger("astro_metadata_translator")
33PACKAGES_VAR = "METADATA_TRANSLATORS"
35hdrnum_option = click.option(
36 "-n",
37 "--hdrnum",
38 default=-1,
39 help="HDU number to read. If the HDU can not be found, a warning is issued but"
40 " reading is attempted using the primary header. The primary header is"
41 " always read and merged with this header. Negative number (the default) "
42 " indicates that the second header will be merged if the FITS file supports"
43 " extended FITS.",
44)
45regex_option = click.option(
46 "-r",
47 "--regex",
48 default=re_default,
49 help="When looking in a directory, regular expression to use to determine whether"
50 f" a file should be examined. Default: '{re_default}'",
51)
52content_option = click.option(
53 "-c",
54 "--content",
55 default="translated",
56 type=click.Choice(["translated", "metadata"], case_sensitive=False),
57 help="Content to store in JSON file. Options are: "
58 "'translated' stores translated metadata in the file; "
59 "'metadata' stores raw FITS headers in the file.",
60)
63@click.group(
64 name="astrometadata",
65 context_settings={"help_option_names": ["-h", "--help"]},
66 invoke_without_command=True,
67)
68@click.option(
69 "--log-level",
70 type=click.Choice(["CRITICAL", "ERROR", "WARNING", "INFO", "DEBUG"], case_sensitive=False),
71 default="INFO",
72 help="Python logging level to use.",
73)
74@click.option(
75 "--traceback/--no-traceback", default=False, help="Give detailed trace back when any errors encountered."
76)
77@click.option(
78 "-p",
79 "--packages",
80 multiple=True,
81 help="Python packages or plugin names to import to register additional translators. This is in addition"
82 f" to any packages specified in the {PACKAGES_VAR} environment variable (colon-separated"
83 " python module names or plugin names).",
84)
85@click.option(
86 "--list-plugins/--no-list-plugins",
87 default=False,
88 help="List all available registered plugins. If true, the command will return immediately.",
89)
90@click.pass_context
91def main(
92 ctx: click.Context, log_level: int, traceback: bool, packages: Sequence[str], list_plugins: bool
93) -> None:
94 """Execute main click command-line."""
95 ctx.ensure_object(dict)
97 # Currently we set the log level globally for all Python loggers
98 # rather than having a metadata translator logger that all metadata
99 # translators can use as a parent. This can potentially cause spurious
100 # messages from numexpr. Try to hide those by setting the numexpr env var.
101 os.environ["NUMEXPR_MAX_THREADS"] = "8"
103 logging.basicConfig(level=log_level)
105 # Traceback needs to be known to subcommands
106 ctx.obj["TRACEBACK"] = traceback
108 plugins = {p.name: p for p in entry_points(group="astro_metadata_translators")}
109 if list_plugins:
110 from astro_metadata_translator import MetadataTranslator
112 print("Builtin translators:")
113 for t in sorted(MetadataTranslator.translators):
114 print(f"- {t}")
115 if plugins:
116 print("Available translator plugins grouped by label (use '-p <label>' to activate):")
117 for label in sorted(plugins):
118 print(f"* {label}:")
119 try:
120 func = plugins[label].load()
121 except Exception as e:
122 print(f" - Unable to load plugin [{e}]")
123 continue
124 translators = func()
125 for t in translators:
126 print(f" - {t}")
127 # Exit early with good status.
128 raise click.exceptions.Exit(0)
130 if ctx.invoked_subcommand is None:
131 # Print the help if we were invoked without a subcommand.
132 click.echo(ctx.get_help())
133 raise click.exceptions.Exit(0)
135 packages_set = set(packages)
136 if PACKAGES_VAR in os.environ:
137 new_packages = os.environ[PACKAGES_VAR].split(":")
138 packages_set.update(new_packages)
140 # Process import requests
141 for m in packages_set:
142 if m in plugins:
143 try:
144 # Loading is sufficient to register the translator.
145 plugins[m].load()
146 except Exception as e:
147 log.warning("Failed to import plugin %s: %s", m, e)
148 continue
149 try:
150 importlib.import_module(m)
151 except (ImportError, ModuleNotFoundError):
152 log.warning("Failed to import translator module: %s", m)
155@main.command(help="Translate metadata in supplied files and report.")
156@click.argument("files", nargs=-1)
157@click.option(
158 "-q",
159 "--quiet/--no-quiet",
160 default=False,
161 help="Do not report the translation content from each header. Only report failures.",
162)
163@hdrnum_option
164@click.option(
165 "-m",
166 "--mode",
167 default="auto",
168 type=click.Choice(["auto", "verbose", "table"], case_sensitive=False),
169 help="Output mode. 'verbose' prints all available information for each file found."
170 " 'table' uses tabular output for a cutdown set of metadata."
171 " 'auto' uses 'verbose' if one file found and 'table' if more than one is found.",
172)
173@regex_option
174@click.option(
175 "-t",
176 "--translator-name",
177 default=None,
178 help="Force a specific translator by name. It must have previously been imported.",
179)
180@click.pass_context
181def translate(
182 ctx: click.Context,
183 files: Sequence[str],
184 quiet: bool,
185 hdrnum: int,
186 mode: str,
187 regex: str,
188 translator_name: str | None,
189) -> None:
190 """Translate a header."""
191 # For quiet mode we want to translate everything but report nothing.
192 if quiet:
193 mode = "none"
195 okay, failed = translate_or_dump_headers(
196 files, regex, hdrnum, ctx.obj["TRACEBACK"], output_mode=mode, translator_name=translator_name
197 )
199 if failed:
200 click.echo("Files with failed translations:", err=True)
201 for f in failed:
202 click.echo(f"\t{f}", err=True)
204 if not okay:
205 # Good status if anything was returned in okay
206 raise click.exceptions.Exit(1)
209@main.command(help="Dump data header to standard out in YAML format.")
210@click.argument("files", nargs=-1)
211@hdrnum_option
212@click.option(
213 "-m",
214 "--mode",
215 default="yaml",
216 type=click.Choice(["yaml", "fixed", "yamlnative", "fixexnative"], case_sensitive=False),
217 help="Output mode. 'yaml' dumps the header in YAML format (this is the default)."
218 " 'fixed' dumps the header in YAML format after applying header corrections."
219 " 'yamlnative' is as for 'yaml' but dumps the native (astropy vs PropertyList) native form."
220 " 'yamlfixed' is as for 'fixed' but dumps the native (astropy vs PropertyList) native form.",
221)
222@regex_option
223@click.pass_context
224def dump(ctx: click.Context, files: Sequence[str], hdrnum: int, mode: str, regex: str) -> None:
225 """Dump a header."""
226 okay, failed = translate_or_dump_headers(files, regex, hdrnum, ctx.obj["TRACEBACK"], output_mode=mode)
228 if failed:
229 click.echo("Files with failed header extraction:", err=True)
230 for f in failed:
231 click.echo(f"\t{f}", err=True)
233 if not okay:
234 # Good status if anything was returned in okay
235 raise click.exceptions.Exit(1)
238@main.command(help="Write JSON sidecar files alongside each data file.")
239@click.argument("files", nargs=-1)
240@hdrnum_option
241@regex_option
242@content_option
243@click.pass_context
244def write_sidecar(ctx: click.Context, files: Sequence[str], hdrnum: int, regex: str, content: str) -> None:
245 """Write a sidecar file with header information."""
246 okay, failed = write_sidecar_files(files, regex, hdrnum, content, ctx.obj["TRACEBACK"])
248 if failed:
249 click.echo("Files with failed header extraction:", err=True)
250 for f in failed:
251 click.echo(f"\t{f}", err=True)
253 if not okay and not failed:
254 # No files found at all.
255 click.echo("Found no files matching regex.")
256 raise click.exceptions.Exit(1)
258 if not okay:
259 # Good status if anything was returned in okay
260 click.echo(f"No files processed successfully. Found {len(failed)}.", err=True)
261 raise click.exceptions.Exit(1)
264@main.command(help="Write JSON index file for entire directory.")
265@click.argument("files", nargs=-1)
266@hdrnum_option
267@regex_option
268@content_option
269@click.option(
270 "-o",
271 "--outpath",
272 type=str,
273 default=None,
274 help="If given, write a single index with all information in specified location."
275 " Default is to write one index per directory where files are located.",
276)
277@click.pass_context
278def write_index(
279 ctx: click.Context, files: Sequence[str], hdrnum: int, regex: str, content: str, outpath: str
280) -> None:
281 """Write a header index file."""
282 okay, failed = write_index_files(
283 files, regex, hdrnum, ctx.obj["TRACEBACK"], content_mode=content, outpath=outpath
284 )
286 if failed:
287 click.echo("Files with failed header extraction:", err=True)
288 for f in failed:
289 click.echo(f"\t{f}", err=True)
291 if not okay:
292 # Good status if anything was returned in okay
293 raise click.exceptions.Exit(1)