Coverage for tests / test_completenessPlot.py: 21%
108 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-05 18:53 +0000
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/>.
23import os
24import shutil
25import tempfile
26import unittest
28import lsst.utils.tests
29import matplotlib
30import matplotlib.pyplot as plt
31import numpy as np
32from lsst.analysis.tools.actions.plot import CompletenessHist
33from lsst.analysis.tools.actions.plot.plotUtils import get_and_remove_figure_text
34from lsst.analysis.tools.math import divide, sqrt
36# Set to True to debug plot test(s)
37debug_plot = False
38if not debug_plot: 38 ↛ 41line 38 didn't jump to line 41 because the condition on line 38 was always true
39 matplotlib.use("Agg")
41ROOT = os.path.abspath(os.path.dirname(__file__))
42filename_texts_ref = os.path.join(ROOT, "data", "test_completenessPlot_texts.txt")
43path_lines_ref = os.path.join(ROOT, "data", "test_completenessPlot_lines")
46class CompletenessPlotTestCase(lsst.utils.tests.TestCase):
47 """CompletenessHist test case."""
49 def setUp(self):
50 self.testDir = tempfile.mkdtemp(dir=ROOT, prefix="test_output_completeness")
52 plot = CompletenessHist()
54 plot.action.action.selector_range_ref.vectorKey = "mag_ref"
55 plot.action.action.selector_range_target.vectorKey = "mag_target"
56 plot.action.bins.mag_low_max = 25500
57 plot.action.bins.mag_interval = 100
58 plot.action.bins.mag_width = 200
60 band = "r"
62 mag = 12.5 + 2.5 * np.log10(np.arange(10, 100000))
63 zeropoint = mag[-1] + 1
64 flux = 10 ** (-0.4 * (mag - zeropoint))
65 rng = np.random.default_rng(0)
66 # See usage below if changing this to allow nan/bad extendedness
67 extendedness = 0.0 + (rng.uniform(size=len(mag)) < 0.99 * (mag - mag[0]) / (mag[-1] - mag[0]))
68 flux_meas = flux + rng.normal(scale=sqrt(flux * (1 + extendedness)))
69 flux_pos = flux_meas > 0
70 flux_err = np.empty_like(flux_meas)
71 flux_err[flux_pos] = sqrt(flux_meas[flux_pos])
72 flux_err[~flux_pos] = 0
73 detect_sn = divide(flux_meas, flux_err)
74 good = detect_sn > 3
75 match_distance = np.full_like(detect_sn, np.nan)
76 rand_norm_abs = np.abs(rng.normal(size=match_distance.size))
77 # make some unmatched as a function of S/N
78 matched = rand_norm_abs < 0.2 * detect_sn
79 match_distance[matched] = rand_norm_abs[matched]
81 flux_meas[~good] = np.nan
83 # estimated from a DC2 tract
84 prob_ps = 0.5 * (1 + np.tanh((19.2 - mag) / 2.8))
85 is_ps = rng.uniform(size=len(prob_ps)) < prob_ps
86 is_extended = rng.uniform(size=len(prob_ps)) > prob_ps
87 is_extended[~good] = np.nan
89 # add some false detections
90 false_positive = np.abs(rng.normal(size=match_distance.size)) > 0.5 * detect_sn
91 flux_meas_extra = flux_meas[false_positive]
92 nan_extra = np.full_like(flux_meas_extra, np.nan)
93 action_hist = plot.action.action
94 action_hist.key_mask_ref = "select_is_ref"
95 action_hist.key_mask_target = "select_is_target"
96 is_ps = np.concatenate((is_ps, nan_extra))
97 extendedness = np.concatenate((extendedness, extendedness[false_positive]))
98 is_extended = np.concatenate((is_extended, is_extended[false_positive]))
100 data = {
101 "mag_ref": np.concatenate((mag, nan_extra)),
102 "mag_target": -2.5 * np.log10(np.concatenate((flux_meas, flux_meas_extra))) + zeropoint,
103 "refcat_is_pointsource": is_ps,
104 action_hist.key_match_distance: np.concatenate((match_distance, nan_extra)),
105 action_hist.key_matched_class: is_ps == ~is_extended,
106 action_hist.key_mask_ref: np.isfinite(is_ps),
107 action_hist.key_mask_target: np.isfinite(extendedness),
108 }
110 results = plot.action(data, band="r")
111 data.update(results)
112 self.band = "r"
113 self.data = data
114 self.plot = plot
115 self.plotInfo = {"plotName": "compurity test", "run": "", "tableName": "", "bands": band}
117 def tearDown(self):
118 if os.path.exists(self.testDir):
119 shutil.rmtree(self.testDir, True)
120 del self.data
121 del self.plot
122 del self.plotInfo
123 del self.testDir
125 def test_completenessPlotWithTwoHistsTask(self):
126 plt.rcParams.update(plt.rcParamsDefault)
127 result = self.plot(
128 data=self.data,
129 band=self.band,
130 plotInfo=self.plotInfo,
131 )
132 self.assertTrue(isinstance(result, plt.Figure))
134 to_compare = True
135 if to_compare:
136 # Set to true to save plots as PNGs
137 # Use matplotlib.testing.compare.compare_images if needed
138 save_images = False
139 if save_images:
140 result.savefig(os.path.join(ROOT, "data", "test_completenessPlot.png"))
142 texts, lines = get_and_remove_figure_text(result)
143 if save_images:
144 result.savefig(os.path.join(ROOT, "data", "test_completenessPlot_unlabeled.png"))
146 # Set to true to re-generate reference data
147 resave = False
149 # Compare line values
150 for idx, line in enumerate(lines):
151 filename = os.path.join(path_lines_ref, f"line_{idx}.txt")
152 if resave:
153 np.savetxt(filename, line)
154 arr = np.loadtxt(filename)
155 # Differences of order 1e-12 possible between MacOS and Linux
156 # Plots are generally not expected to be that precise
157 # Differences to 1e-3 should not be visible with this test data
158 # nans are possible here (divide by zero)
159 self.assertFloatsAlmostEqual(arr, line, atol=1e-3, rtol=1e-4, ignoreNaNs=True)
161 # Ensure that newlines within labels are replaced by a sentinel
162 newline = "\n"
163 newline_replace = "[newline]"
164 # Compare text labels
165 if resave:
166 with open(filename_texts_ref, "w") as f:
167 f.writelines(f"{text.strip().replace(newline, newline_replace)}\n" for text in texts)
169 with open(filename_texts_ref, "r") as f:
170 texts_ref = set(x.strip() for x in f.readlines())
171 texts_set = set(x.strip().replace(newline, newline_replace) for x in texts)
173 self.assertEqual(texts_ref, texts_set)
174 if debug_plot:
175 plt.show()
178class MemoryTester(lsst.utils.tests.MemoryTestCase):
179 pass
182def setup_module(module):
183 lsst.utils.tests.init()
186if __name__ == "__main__": 186 ↛ 187line 186 didn't jump to line 187 because the condition on line 186 was never true
187 lsst.utils.tests.init()
188 unittest.main()