Coverage for python/lsst/meas/extensions/multiprofit/utils.py: 12%
52 statements
« prev ^ index » next coverage.py v7.16.1, created at 2026-09-23 11:07 +0000
1# This file is part of meas_extensions_multiprofit.
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# This program is free software: you can redistribute it and/or modify
10# it under the terms of the GNU General Public License as published by
11# the Free Software Foundation, either version 3 of the License, or
12# (at your option) any later version.
13#
14# This program is distributed in the hope that it will be useful,
15# but WITHOUT ANY WARRANTY; without even the implied warranty of
16# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
17# GNU General Public License for more details.
18#
19# You should have received a copy of the GNU General Public License
20# along with this program. If not, see <https://www.gnu.org/licenses/>.
22from collections import defaultdict
23from collections.abc import Iterable
25import numpy as np
27import lsst.afw.detection as afwDetect
28import lsst.afw.image as afwImage
29import lsst.geom as geom
32def defaultdictNested():
33 """Get a nested defaultdict with defaultdict default value.
35 Returns
36 -------
37 defaultdict : `defaultdict`
38 A `defaultdict` with `defaultdict` default values.
39 """
40 return defaultdict(defaultdictNested)
43def get_all_subclasses(cls, children_first: bool = True):
44 """Return all subclasses of a class recursively.
46 Parameters
47 ----------
48 cls
49 The class to get subclasses for.
50 children_first
51 If true, return child (direct subclasses) first, followed by their
52 children (recursively). Otherwise, return each direct child followed
53 by its own descendants first.
54 """
55 subclasses = {c: None for c in cls.__subclasses__()}
56 subclasses_return = subclasses.copy() if children_first else {}
57 for subclass in subclasses:
58 if children_first:
59 subclasses_return[subclass] = None
60 for subsubclass in get_all_subclasses(subclass):
61 subclasses_return[subsubclass] = None
62 return list(subclasses_return)
65# TODO: Allow addition to existing image
66def get_spanned_image(
67 exposure: afwImage.Exposure,
68 footprint: afwDetect.Footprint = None,
69 bbox: geom.Box2I | None = None,
70 spans: np.ndarray | None = None,
71 get_sig_inv: bool = False,
72 calibrate: bool = True,
73) -> tuple[np.ndarray, geom.Box2I, np.ndarray]:
74 """Get an image masked by its spanset.
76 Parameters
77 ----------
78 exposure
79 An exposure to extract arrays from.
80 footprint
81 The footprint to get spans/bboxes from. Not needed if both of
82 `bbox` and `spans` are provided.
83 bbox
84 The bounding box to subset the exposure with.
85 Defaults to the footprint's bbox.
86 spans
87 A spanset array (inverse mask/selection).
88 Defaults to the footprint's spans.
89 get_sig_inv
90 Whether to get the inverse variance and return its square root.
91 calibrate
92 Whether to calibrate the image; set to False if already calibrated.
94 Returns
95 -------
96 image
97 The image array, with masked pixels set to zero.
98 bbox
99 The bounding box used to subset the exposure.
100 sig_inv
101 The inverse sigma array, with masked pixels set to zero.
102 Set to None if `get_sig_inv` is False.
103 """
104 bbox_is_none = bbox is None
105 if bbox_is_none:
106 bbox = footprint.getBBox()
107 if not (bbox.getHeight() > 0 and bbox.getWidth() > 0):
108 return None, bbox
109 if spans is None:
110 spans = footprint.getSpans().asArray()
111 sig_inv = afwImage.ImageF(bbox) if get_sig_inv else None
112 img = afwImage.ImageF(bbox)
113 img.array[:] = np.nan
114 if footprint is None:
115 maskedIm = exposure.maskedImage.subset(bbox)
116 if not calibrate:
117 img = maskedIm.image.array
118 sig_inv.array[spans] = 1 / np.sqrt(maskedIm.variance.array[spans])
119 else:
120 img.array[spans] = footprint.getImageArray()
121 if get_sig_inv:
122 # footprint.getVarianceArray() returns zeros
123 variance = exposure.variance[bbox]
124 if not calibrate:
125 sig_inv.array[spans] = 1 / np.sqrt(variance.array[spans])
126 if calibrate:
127 # Have to calibrate with the original image
128 maskedIm = afwImage.MaskedImageF(
129 image=exposure.image[bbox],
130 variance=variance if get_sig_inv else None,
131 )
132 if calibrate:
133 maskedIm = exposure.photoCalib.calibrateImage(maskedIm)
134 if footprint is None:
135 img = maskedIm.image.array
136 else:
137 # Apply the calibration to the deblended footprint
138 # ... hopefully it's multiplicative enough
139 img.array[spans] *= maskedIm.image.array[spans] / exposure.image[bbox].array[spans]
140 img = img.array
141 if get_sig_inv:
142 sig_inv.array[spans] = 1 / np.sqrt(maskedIm.variance.array[spans])
143 # Should not happen but does with footprints having nans
144 sig_inv.array[~(sig_inv.array >= 0)] = 0
146 return np.array(img, dtype="float64"), bbox, np.array(sig_inv.array, dtype="float64")
149def join_and_filter(separator: str, items: Iterable[str], exclusion: str | None = None) -> str:
150 """Join an iterable of items by a separator, filtering out an exclusion.
152 Parameters
153 ----------
154 separator
155 The separator to join items by.
156 items
157 Items to join.
158 exclusion
159 The pattern to exclude.
161 Returns
162 -------
163 joined
164 The joined string.
165 """
166 return separator.join(filter(exclusion, items))