Coverage for python / astro_metadata_translator / cli / astrometadata.py: 31%
129 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-04 02:37 -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.
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.pass_context
175def translate(
176 ctx: click.Context, files: Sequence[str], quiet: bool, hdrnum: int, mode: str, regex: str
177) -> None:
178 """Translate a header."""
179 # For quiet mode we want to translate everything but report nothing.
180 if quiet:
181 mode = "none"
183 okay, failed = translate_or_dump_headers(files, regex, hdrnum, ctx.obj["TRACEBACK"], output_mode=mode)
185 if failed:
186 click.echo("Files with failed translations:", err=True)
187 for f in failed:
188 click.echo(f"\t{f}", err=True)
190 if not okay:
191 # Good status if anything was returned in okay
192 raise click.exceptions.Exit(1)
195@main.command(help="Dump data header to standard out in YAML format.")
196@click.argument("files", nargs=-1)
197@hdrnum_option
198@click.option(
199 "-m",
200 "--mode",
201 default="yaml",
202 type=click.Choice(["yaml", "fixed", "yamlnative", "fixexnative"], case_sensitive=False),
203 help="Output mode. 'yaml' dumps the header in YAML format (this is the default)."
204 " 'fixed' dumps the header in YAML format after applying header corrections."
205 " 'yamlnative' is as for 'yaml' but dumps the native (astropy vs PropertyList) native form."
206 " 'yamlfixed' is as for 'fixed' but dumps the native (astropy vs PropertyList) native form.",
207)
208@regex_option
209@click.pass_context
210def dump(ctx: click.Context, files: Sequence[str], hdrnum: int, mode: str, regex: str) -> None:
211 """Dump a header."""
212 okay, failed = translate_or_dump_headers(files, regex, hdrnum, ctx.obj["TRACEBACK"], output_mode=mode)
214 if failed:
215 click.echo("Files with failed header extraction:", err=True)
216 for f in failed:
217 click.echo(f"\t{f}", err=True)
219 if not okay:
220 # Good status if anything was returned in okay
221 raise click.exceptions.Exit(1)
224@main.command(help="Write JSON sidecar files alongside each data file.")
225@click.argument("files", nargs=-1)
226@hdrnum_option
227@regex_option
228@content_option
229@click.pass_context
230def write_sidecar(ctx: click.Context, files: Sequence[str], hdrnum: int, regex: str, content: str) -> None:
231 """Write a sidecar file with header information."""
232 okay, failed = write_sidecar_files(files, regex, hdrnum, content, ctx.obj["TRACEBACK"])
234 if failed:
235 click.echo("Files with failed header extraction:", err=True)
236 for f in failed:
237 click.echo(f"\t{f}", err=True)
239 if not okay and not failed:
240 # No files found at all.
241 click.echo("Found no files matching regex.")
242 raise click.exceptions.Exit(1)
244 if not okay:
245 # Good status if anything was returned in okay
246 click.echo(f"No files processed successfully. Found {len(failed)}.", err=True)
247 raise click.exceptions.Exit(1)
250@main.command(help="Write JSON index file for entire directory.")
251@click.argument("files", nargs=-1)
252@hdrnum_option
253@regex_option
254@content_option
255@click.option(
256 "-o",
257 "--outpath",
258 type=str,
259 default=None,
260 help="If given, write a single index with all information in specified location."
261 " Default is to write one index per directory where files are located.",
262)
263@click.pass_context
264def write_index(
265 ctx: click.Context, files: Sequence[str], hdrnum: int, regex: str, content: str, outpath: str
266) -> None:
267 """Write a header index file."""
268 okay, failed = write_index_files(
269 files, regex, hdrnum, ctx.obj["TRACEBACK"], content_mode=content, outpath=outpath
270 )
272 if failed:
273 click.echo("Files with failed header extraction:", err=True)
274 for f in failed:
275 click.echo(f"\t{f}", err=True)
277 if not okay:
278 # Good status if anything was returned in okay
279 raise click.exceptions.Exit(1)