Coverage for python / astro_metadata_translator / bin / writesidecar.py: 22%
38 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-14 23:38 +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__ = ("write_sidecar_file", "write_sidecar_files")
16import logging
17import traceback
18from collections.abc import Sequence
19from typing import IO, TYPE_CHECKING
21from lsst.resources import ResourcePath
23from ..file_helpers import find_files, read_file_info
25if TYPE_CHECKING:
26 from lsst.resources import ResourcePathExpression
28log = logging.getLogger(__name__)
31def write_sidecar_file(
32 file: ResourcePathExpression,
33 hdrnum: int,
34 content_mode: str,
35 print_trace: bool,
36 outstream: IO | None = None,
37) -> bool:
38 """Write JSON summary to sidecar file.
40 Parameters
41 ----------
42 file : `str` or `lsst.resources.ResourcePathExpression`
43 The file from which the header is to be read.
44 hdrnum : `int`
45 The HDU number to read. The primary header is always read and
46 merged with the header from this HDU.
47 content_mode : `str`
48 Content mode to use when writing JSON to sidecar. Options are:
49 ``metadata`` to write the unmodified header;
50 ``translated`` to write the translated ObservationInfo.
51 print_trace : `bool`
52 If there is an error reading the file and this parameter is `True`,
53 a full traceback of the exception will be reported. If `False` prints
54 a one line summary of the error condition. If `None` the exception
55 will be allowed.
56 outstream : `io.StringIO`, optional
57 Output stream to use for standard messages. Defaults to `None` which
58 uses the default output stream.
60 Returns
61 -------
62 success : `bool`
63 `True` if the file was handled successfully, `False` if the file
64 could not be processed.
65 """
66 if content_mode not in ("metadata", "translated"):
67 raise ValueError(f"Specified content mode '{content_mode}' is not understood.")
69 try:
70 # Calculate the JSON from the file
71 json_str = read_file_info(
72 file,
73 hdrnum,
74 content_mode=content_mode,
75 content_type="json",
76 print_trace=print_trace,
77 outstream=outstream,
78 )
79 if json_str is None:
80 return False
82 # Calculate sidecar file name derived from this file.
83 # .fits.gz should be replaced
84 # with .json, and not resulting in .fits.json.
85 uri = ResourcePath(file)
86 newfile = uri.updatedExtension(".json")
87 newfile.write(str(json_str).encode())
88 log.debug("Writing sidecar file %s", newfile)
90 except Exception as e:
91 if print_trace is None:
92 raise e
93 if print_trace:
94 traceback.print_exc(file=outstream)
95 else:
96 print(repr(e), file=outstream)
97 return False
98 return True
101def write_sidecar_files(
102 files: Sequence[str],
103 regex: str,
104 hdrnum: int,
105 content_mode: str,
106 print_trace: bool,
107 outstream: IO | None = None,
108) -> tuple[list[str], list[str]]:
109 """Process each file and create sidecar file.
111 Parameters
112 ----------
113 files : iterable of `str`
114 The files or directories from which the headers are to be read.
115 regex : `str`
116 Regular expression string used to filter files when a directory is
117 scanned.
118 hdrnum : `int`
119 The HDU number to read. The primary header is always read and
120 merged with the header from this HDU.
121 content_mode : `str`
122 Content mode to use when writing JSON to sidecar. Options are:
123 ``metadata`` to write the unmodified header;
124 ``translated`` to write the translated ObservationInfo.
125 print_trace : `bool`
126 If there is an error reading the file and this parameter is `True`,
127 a full traceback of the exception will be reported. If `False` prints
128 a one line summary of the error condition.
129 outstream : `io.StringIO`, optional
130 Output stream to use for standard messages. Defaults to `None` which
131 uses the default output stream.
133 Returns
134 -------
135 okay : `list` of `str`
136 All the files that were processed successfully.
137 failed : `list` of `str`
138 All the files that could not be processed.
139 """
140 found_files = find_files(files, regex)
142 # Process each file
143 failed = []
144 okay = []
145 for path in sorted(found_files):
146 isok = write_sidecar_file(path, hdrnum, content_mode, print_trace, outstream)
147 if isok:
148 okay.append(str(path))
149 else:
150 failed.append(str(path))
152 return okay, failed