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