Coverage for python/lsst/images/cli/_convert.py: 69%
94 statements
« prev ^ index » next coverage.py v7.15.0, created at 2026-07-04 17:17 +0000
1# This file is part of lsst-images.
2#
3# Developed for the LSST Data Management System.
4# This product includes software developed by the LSST Project
5# (https://www.lsst.org).
6# See the COPYRIGHT 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.
11from __future__ import annotations
13__all__ = ("convert",)
15import os
16import tempfile
17from typing import TYPE_CHECKING, Any
19import astropy.io.fits
20import click
21from click.core import ParameterSource
23from ..serialization import backend_for_path, write
25if TYPE_CHECKING:
26 from .. import VisitImage
27 from ..cells import CellCoadd
29_LEGACY_TYPES = ("visit_image", "cell_coadd")
32def detect_legacy_type(path: str) -> str | None:
33 """Return ``"visit_image"`` or ``"cell_coadd"`` from a legacy FITS file's
34 ``HIERARCH LSST BUTLER DATASETTYPE`` header, or `None` if it cannot be
35 determined.
37 A dataset type ending in ``visit_image`` (e.g. ``visit_image``,
38 ``preliminary_visit_image``, difference images) is a `VisitImage`; one
39 containing ``coadd`` is a `CellCoadd`. astropy exposes the
40 ``HIERARCH LSST BUTLER DATASETTYPE`` card as
41 ``header["LSST BUTLER DATASETTYPE"]``.
43 Parameters
44 ----------
45 path
46 Path to the legacy FITS file to inspect.
47 """
48 dataset_type: str | None = None
49 with astropy.io.fits.open(path) as hdul:
50 for hdu in hdul:
51 value = hdu.header.get("LSST BUTLER DATASETTYPE")
52 if value:
53 dataset_type = str(value)
54 break
55 if dataset_type is None:
56 return None
57 if dataset_type.endswith("visit_image"):
58 return "visit_image"
59 if dataset_type == "difference_image": 59 ↛ 60line 59 didn't jump to line 60 because the condition on line 59 was never true
60 return "difference_image"
61 if "coadd" in dataset_type:
62 return "cell_coadd"
63 return None
66def _load_skymap(skymap: str | None, butler: str | None, collection: str | None, skymap_name: str) -> Any:
67 """Load a skymap object from a pickle path or a butler repository."""
68 if skymap is not None:
69 import pickle
71 with open(skymap, "rb") as stream:
72 return pickle.load(stream)
73 if butler is not None:
74 if collection is None:
75 raise click.ClickException("--butler also requires --collection (the skymap's collection).")
76 from lsst.daf.butler import Butler
78 with Butler.from_config(butler) as repo:
79 return repo.get("skyMap", skymap=skymap_name, collections=collection)
80 raise click.ClickException("Converting a cell coadd requires --skymap (a pickled skymap) or --butler.")
83def _read_legacy(
84 input: str,
85 legacy_type: str,
86 skymap: str | None,
87 butler: str | None,
88 collection: str | None,
89 preserve_quantization: bool = False,
90) -> VisitImage | CellCoadd:
91 """Read a legacy FITS file into the corresponding lsst.images object."""
92 if legacy_type == "visit_image":
93 from .. import VisitImage
95 return VisitImage.read_legacy(input, preserve_quantization=preserve_quantization)
96 if legacy_type == "difference_image": 96 ↛ 97line 96 didn't jump to line 97 because the condition on line 96 was never true
97 from .. import DifferenceImage
99 return DifferenceImage.read_legacy(input, preserve_quantization=preserve_quantization)
100 if legacy_type == "cell_coadd": 100 ↛ 114line 100 didn't jump to line 114 because the condition on line 100 was always true
101 from lsst.cell_coadds import MultipleCellCoadd
103 from .. import get_legacy_deep_coadd_mask_planes
104 from ..cells import CellCoadd
106 legacy = MultipleCellCoadd.read_fits(input)
107 sky = _load_skymap(skymap, butler, collection, legacy.identifiers.skymap)
108 tract_info = sky[legacy.identifiers.tract]
109 return CellCoadd.from_legacy(
110 legacy,
111 tract_info=tract_info,
112 plane_map=get_legacy_deep_coadd_mask_planes(),
113 )
114 raise click.ClickException(f"Conversion of {legacy_type!r} is not yet implemented.")
117@click.command(name="convert")
118@click.argument("input", type=click.Path(exists=True, dir_okay=False))
119@click.argument("output", type=click.Path(dir_okay=False))
120@click.option(
121 "--type",
122 "type_",
123 type=click.Choice(_LEGACY_TYPES),
124 default=None,
125 help="Legacy input type; overrides auto-detection.",
126)
127@click.option(
128 "--skymap",
129 type=click.Path(exists=True, dir_okay=False),
130 default=None,
131 help="Pickled skymap (required for cell coadds unless --butler is given).",
132)
133@click.option(
134 "--butler",
135 default=None,
136 help="Butler repository to resolve the skymap (cell coadds only).",
137)
138@click.option(
139 "--collection",
140 default=None,
141 help="Butler collection holding the skymap (required with --butler).",
142)
143@click.option("--overwrite", is_flag=True, default=False, help="Overwrite OUTPUT if it exists.")
144@click.option(
145 "--preserve-quantization/--no-preserve-quantization",
146 default=True,
147 help=(
148 "Preserve quantization-compressed pixel values so they can be written "
149 "back out losslessly (visit images only). On by default."
150 ),
151)
152def convert(
153 input: str,
154 output: str,
155 type_: str | None,
156 skymap: str | None,
157 butler: str | None,
158 collection: str | None,
159 overwrite: bool,
160 preserve_quantization: bool,
161) -> None: # numpydoc ignore=PR01
162 """Convert a legacy FITS file to a new lsst.images format.
164 The output format is chosen from OUTPUT's extension.
165 """
166 try:
167 backend = backend_for_path(output)
168 except ValueError as err:
169 raise click.ClickException(str(err)) from None
171 legacy_type = type_ or detect_legacy_type(input)
172 if legacy_type is None: 172 ↛ 173line 172 didn't jump to line 173 because the condition on line 172 was never true
173 raise click.ClickException(f"Could not determine the legacy type of {input!r}; pass --type.")
175 # The flag is on by default, so only object when the user explicitly set
176 # it for a type that does not support it (only visit images do).
177 if legacy_type != "visit_image":
178 source = click.get_current_context().get_parameter_source("preserve_quantization")
179 if source == ParameterSource.COMMANDLINE:
180 raise click.ClickException(
181 f"--preserve-quantization is only valid for visit images, not {legacy_type!r}."
182 )
183 preserve_quantization = False
185 output_abs = os.path.realpath(output)
186 if os.path.realpath(input) == output_abs:
187 raise click.ClickException("INPUT and OUTPUT must be different paths.")
189 if os.path.exists(output_abs) and not overwrite: 189 ↛ 190line 189 didn't jump to line 190 because the condition on line 189 was never true
190 raise click.ClickException(f"{output!r} already exists; pass --overwrite to replace it.")
192 try:
193 obj = _read_legacy(input, legacy_type, skymap, butler, collection, preserve_quantization)
194 except click.ClickException:
195 raise
196 except ImportError as err:
197 raise click.ClickException(
198 f"Reading a legacy {legacy_type} requires Rubin packages that are not installed: {err}"
199 ) from None
201 # Write to a temporary file in the output's directory and move it into
202 # place only after a successful write, so a read or write failure never
203 # destroys an existing OUTPUT (the backends refuse to overwrite, so the
204 # temporary path must not already exist).
205 output_dir = os.path.dirname(output_abs)
206 with tempfile.TemporaryDirectory(dir=output_dir) as staging:
207 staged = os.path.join(staging, os.path.basename(output_abs))
208 write(obj, staged)
209 os.replace(staged, output_abs)
210 click.echo(f"Wrote {output} ({backend.name}, {legacy_type}).")