Coverage for python / lsst / analysis / tools / actions / scalar / scalarActions.py: 33%
199 statements
« prev ^ index » next coverage.py v7.13.2, created at 2026-02-18 05:21 -0800
1# This file is part of analysis_tools.
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 __future__ import annotations
24__all__ = (
25 "MedianAction",
26 "MeanAction",
27 "StdevAction",
28 "ValueAction",
29 "SigmaMadAction",
30 "CountAction",
31 "CountUniqueAction",
32 "ApproxFloor",
33 "FracThreshold",
34 "MaxAction",
35 "MinAction",
36 "FracInRange",
37 "FracNan",
38 "SumAction",
39 "MedianHistAction",
40 "IqrHistAction",
41 "DivideScalar",
42 "RmsAction",
43)
45import logging
46import operator
47from math import nan
48from typing import cast
50import numpy as np
52from lsst.pex.config import ChoiceField, Field
53from lsst.pex.config.configurableActions import ConfigurableActionField
55from ...interfaces import KeyedData, KeyedDataSchema, Scalar, ScalarAction, Vector
56from ...math import nanMax, nanMean, nanMedian, nanMin, nanSigmaMad, nanStd
58log = logging.getLogger(__name__)
61def _dataToArray(data):
62 """Convert input data into a numpy array using the appropriate
63 protocol. `np.from_dlpack` is used for Tensor-like arrays
64 where possible.
65 """
66 try:
67 return np.from_dlpack(data)
68 except (AttributeError, BufferError):
69 return np.array(data)
72class ScalarFromVectorAction(ScalarAction):
73 """Calculates a statistic from a single vector."""
75 vectorKey = Field[str]("Key of Vector to compute statistic from.")
77 def getInputSchema(self) -> KeyedDataSchema:
78 return ((self.vectorKey, Vector),)
81class MedianAction(ScalarFromVectorAction):
82 """Calculates the median of the given data."""
84 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
85 mask = self.getMask(**kwargs)
86 values = _dataToArray(data[self.vectorKey.format(**kwargs)])[mask]
87 med = nanMedian(values) if values.size else np.nan
89 return med
92class MeanAction(ScalarFromVectorAction):
93 """Calculates the mean of the given data."""
95 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
96 mask = self.getMask(**kwargs)
97 values = _dataToArray(data[self.vectorKey.format(**kwargs)])[mask]
98 mean = nanMean(values) if values.size else np.nan
100 return mean
103class StdevAction(ScalarFromVectorAction):
104 """Calculates the standard deviation of the given data."""
106 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
107 mask = self.getMask(**kwargs)
108 return nanStd(_dataToArray(data[self.vectorKey.format(**kwargs)])[mask])
111class RmsAction(ScalarFromVectorAction):
112 """Calculates the root mean square of the given data (without subtracting
113 the mean as in StdevAction)."""
115 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
116 mask = self.getMask(**kwargs)
117 vector = _dataToArray(data[self.vectorKey.format(**kwargs)])[mask]
118 vector = vector[~np.isnan(vector)]
120 return np.sqrt(np.mean(vector**2))
123class ValueAction(ScalarFromVectorAction):
124 """Extracts the first value from a vector."""
126 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
127 return cast(Scalar, float(data[self.vectorKey.format(**kwargs)][0]))
130class SigmaMadAction(ScalarFromVectorAction):
131 """Calculates the sigma mad of the given data."""
133 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
134 mask = self.getMask(**kwargs)
135 values = _dataToArray(data[self.vectorKey.format(**kwargs)])[mask]
136 return nanSigmaMad(values)
139class CountAction(ScalarAction):
140 """Performs count actions, with threshold-based filtering.
141 The operator is specified as a string, for example, "lt", "le", "ge",
142 "gt", "ne", and "eq" for the mathematical operations <, <=, >=, >, !=,
143 and == respectively. To count non-NaN values, only pass the column name
144 as vector key. To count NaN values, pass threshold = nan (from math.nan).
145 Optionally to configure from a YAML file, pass "threshold: !!float nan".
146 To compute the number of elements with values less than a given threshold,
147 use op="le".
148 """
150 vectorKey = Field[str]("Key of Vector to count")
151 op = ChoiceField[str](
152 doc="Operator name string.",
153 allowed={
154 "lt": "less than threshold",
155 "le": "less than or equal to threshold",
156 "ge": "greater than or equal to threshold",
157 "ne": "not equal to a given value",
158 "eq": "equal to a given value",
159 "gt": "greater than threshold",
160 },
161 default="ne",
162 )
163 threshold = Field[float](doc="Threshold to apply.", default=nan)
165 def getInputSchema(self) -> KeyedDataSchema:
166 return ((self.vectorKey, Vector),)
168 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
169 mask = self.getMask(**kwargs)
170 arr = _dataToArray(data[self.vectorKey.format(**kwargs)])[mask]
172 # Count NaNs and non-NaNs
173 if self.threshold == nan:
174 if self.op == "eq":
175 # Count number of NaNs
176 result = np.isnan(arr).sum()
177 return cast(Scalar, int(result))
178 elif self.op == "ne":
179 # Count number of non-NaNs
180 result = arr.size - np.isnan(arr).sum()
181 return cast(Scalar, int(result))
182 else:
183 raise ValueError("Invalid operator for counting NaNs.")
184 # Count for given threshold ignoring all NaNs
185 else:
186 result = arr[~np.isnan(arr)]
187 result = cast(
188 Scalar,
189 int(np.sum(getattr(operator, self.op)(result, self.threshold))),
190 )
191 return result
194class CountUniqueAction(ScalarFromVectorAction):
195 """Counts the number of unique rows in a given column."""
197 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
198 mask = self.getMask(**kwargs)
199 values = _dataToArray(data[self.vectorKey.format(**kwargs)])[mask]
200 count = np.unique(values).size
201 return cast(Scalar, count)
204class ApproxFloor(ScalarFromVectorAction):
205 """Returns the median of the lowest ten values of the sorted input."""
207 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
208 mask = self.getMask(**kwargs)
209 values = np.sort(_dataToArray(data[self.vectorKey.format(**kwargs)])[mask], axis=None) # type: ignore
210 x = values.size // 10
211 return nanMedian(values[-x:])
214class FracThreshold(ScalarFromVectorAction):
215 """Compute the fraction of a distribution above or below a threshold.
217 The operator is specified as a string, for example,
218 "lt", "le", "ge", "gt" for the mathematical operations <, <=, >=, >. To
219 compute the fraction of elements with values less than a given threshold,
220 use op="le".
221 """
223 op = ChoiceField[str](
224 doc="Operator name string.",
225 allowed={
226 "lt": "less than threshold",
227 "le": "less than or equal to threshold",
228 "ge": "greater than or equal to threshold",
229 "gt": "greater than threshold",
230 },
231 )
232 threshold = Field[float](doc="Threshold to apply.")
233 percent = Field[bool](doc="Express result as percentage", default=False)
234 relative_to_median = Field[bool](doc="Calculate threshold relative to the median?", default=False)
235 use_absolute_value = Field[bool](
236 doc=(
237 "Calculate threshold after taking absolute value. If relative_to_median"
238 " is true the absolute value will be applied after the median is subtracted"
239 ),
240 default=False,
241 )
243 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
244 mask = self.getMask(**kwargs)
245 values = _dataToArray(data[self.vectorKey.format(**kwargs)])[mask]
246 values = values[np.logical_not(np.isnan(values))]
247 n_values = values.size
248 if n_values == 0:
249 return np.nan
250 threshold = self.threshold
251 # If relative_to_median is set, shift the threshold to be median+thresh
252 if self.relative_to_median and values.size > 0:
253 offset = nanMedian(values)
254 if np.isfinite(offset):
255 values -= offset
256 if self.use_absolute_value:
257 values = np.abs(values)
258 result = cast(
259 Scalar,
260 float(np.sum(getattr(operator, self.op)(values, threshold)) / n_values), # type: ignore
261 )
262 if self.percent:
263 return 100.0 * result
264 else:
265 return result
268class MaxAction(ScalarFromVectorAction):
269 """Returns the maximum of the given data."""
271 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
272 mask = self.getMask(**kwargs)
273 return nanMax(_dataToArray(data[self.vectorKey.format(**kwargs)])[mask])
276class MinAction(ScalarFromVectorAction):
277 """Returns the minimum of the given data."""
279 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
280 mask = self.getMask(**kwargs)
281 return nanMin(_dataToArray(data[self.vectorKey.format(**kwargs)])[mask])
284class FracInRange(ScalarFromVectorAction):
285 """Compute the fraction of a distribution that is between specified
286 minimum and maximum values, and is not NaN.
287 """
289 maximum = Field[float](doc="The maximum value", default=np.nextafter(np.inf, 0.0))
290 minimum = Field[float](doc="The minimum value", default=np.nextafter(-np.inf, 0.0))
291 percent = Field[bool](doc="Express result as percentage", default=False)
293 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
294 mask = self.getMask(**kwargs)
295 values = _dataToArray(data[self.vectorKey.format(**kwargs)])[mask]
296 nvalues = values.size
297 values = values[np.logical_not(np.isnan(values))]
298 sel_range = (values >= self.minimum) & (values < self.maximum)
299 result = cast(
300 Scalar,
301 float(values[sel_range].size / nvalues), # type: ignore
302 )
303 if self.percent:
304 return 100.0 * result
305 else:
306 return result
309class FracNan(ScalarFromVectorAction):
310 """Compute the fraction of vector entries that are NaN."""
312 percent = Field[bool](doc="Express result as percentage", default=False)
314 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
315 mask = self.getMask(**kwargs)
316 values = _dataToArray(data[self.vectorKey.format(**kwargs)])[mask]
317 nvalues = values.size
318 values = values[np.isnan(values)]
319 result = cast(
320 Scalar,
321 float(values.size / nvalues), # type: ignore
322 )
323 if self.percent:
324 return 100.0 * result
325 else:
326 return result
329class SumAction(ScalarFromVectorAction):
330 """Returns the sum of all values in the column."""
332 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
333 mask = self.getMask(**kwargs)
334 arr = _dataToArray(data[self.vectorKey.format(**kwargs)])[mask]
335 return cast(Scalar, np.nansum(arr))
338class MedianHistAction(ScalarAction):
339 """Calculates the median of the given histogram data."""
341 histKey = Field[str]("Key of frequency Vector")
342 midKey = Field[str]("Key of bin midpoints Vector")
344 def getInputSchema(self) -> KeyedDataSchema:
345 return (
346 (self.histKey, Vector),
347 (self.midKey, Vector),
348 )
350 def histMedian(self, hist, bin_mid):
351 """Calculates the median of a histogram with binned values
353 Parameters
354 ----------
355 hist : `numpy.ndarray`
356 Frequency array
357 bin_mid : `numpy.ndarray`
358 Bin midpoints array
360 Returns
361 -------
362 median : `float`
363 Median of histogram with binned values
364 """
365 cumulative_sum = np.cumsum(hist)
366 median_index = np.searchsorted(cumulative_sum, cumulative_sum[-1] / 2)
367 median = bin_mid[median_index]
368 return median
370 def __call__(self, data: KeyedData, **kwargs):
371 hist = _dataToArray(data[self.histKey.format(**kwargs)])
372 if hist.size != 0:
373 bin_mid = _dataToArray(data[self.midKey.format(**kwargs)])
374 med = cast(Scalar, float(self.histMedian(hist, bin_mid)))
375 else:
376 med = np.nan
377 return med
380class IqrHistAction(ScalarAction):
381 """Calculates the interquartile range of the given histogram data."""
383 histKey = Field[str]("Key of frequency Vector")
384 midKey = Field[str]("Key of bin midpoints Vector")
386 def getInputSchema(self) -> KeyedDataSchema:
387 return (
388 (self.histKey, Vector),
389 (self.midKey, Vector),
390 )
392 def histIqr(self, hist, bin_mid):
393 """Calculates the interquartile range of a histogram with binned values
395 Parameters
396 ----------
397 hist : `numpy.ndarray`
398 Frequency array
399 bin_mid : `numpy.ndarray`
400 Bin midpoints array
402 Returns
403 -------
404 iqr : `float`
405 Inter-quartile range of histogram with binned values
406 """
407 cumulative_sum = np.cumsum(hist)
408 liqr_index = np.searchsorted(cumulative_sum, cumulative_sum[-1] / 4)
409 uiqr_index = np.searchsorted(cumulative_sum, (3 / 4) * cumulative_sum[-1])
410 liqr = bin_mid[liqr_index]
411 uiqr = bin_mid[uiqr_index]
412 iqr = uiqr - liqr
413 return iqr
415 def __call__(self, data: KeyedData, **kwargs):
416 hist = _dataToArray(data[self.histKey.format(**kwargs)])
417 if hist.size != 0:
418 bin_mid = _dataToArray(data[self.midKey.format(**kwargs)])
419 iqr = cast(Scalar, float(self.histIqr(hist, bin_mid)))
420 else:
421 iqr = np.nan
422 return iqr
425class DivideScalar(ScalarAction):
426 """Calculate (A/B) for scalars."""
428 actionA = ConfigurableActionField[ScalarAction](doc="Action which supplies scalar A")
429 actionB = ConfigurableActionField[ScalarAction](doc="Action which supplies scalar B")
431 def getInputSchema(self) -> KeyedDataSchema:
432 yield from self.actionA.getInputSchema()
433 yield from self.actionB.getInputSchema()
435 def __call__(self, data: KeyedData, **kwargs) -> Scalar:
436 """Return the result of A/B.
438 Parameters
439 ----------
440 data : `KeyedData`
442 Returns
443 -------
444 result : `Scalar`
445 The result of dividing A by B.
446 """
447 scalarA = self.actionA(data, **kwargs)
448 scalarB = self.actionB(data, **kwargs)
449 if scalarB == 0:
450 if scalarA == 0:
451 log.warning("Both numerator and denominator are zero! Returning NaN.")
452 return np.nan
453 else:
454 value = np.sign(scalarA) * np.inf
455 log.warning("Non-zero scalar divided by zero! Returning %f.", value)
456 return value
457 else:
458 return scalarA / scalarB