Coverage for tests/test_scatterPlot.py: 89%
102 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-07-16 15:49 -0700
1# This file is part of analysis_drp.
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
32import pandas as pd
33from lsst.analysis.tools.actions.plot.plotUtils import get_and_remove_figure_text
34from lsst.analysis.tools.actions.plot.scatterplotWithTwoHists import (
35 ScatterPlotStatsAction,
36 ScatterPlotWithTwoHists,
37)
38from lsst.analysis.tools.actions.vector.mathActions import ConstantValue, DivideVector, SubtractVector
39from lsst.analysis.tools.actions.vector.selectors import (
40 GalaxySelector,
41 SnSelector,
42 StarSelector,
43 VectorSelector,
44)
45from lsst.analysis.tools.actions.vector.vectorActions import ConvertFluxToMag, DownselectVector, LoadVector
46from lsst.analysis.tools.interfaces import AnalysisTool
47from lsst.analysis.tools.math import sqrt
49matplotlib.use("Agg")
51ROOT = os.path.abspath(os.path.dirname(__file__))
52filename_texts_ref = os.path.join(ROOT, "data", "test_scatterPlot_texts.txt")
53path_lines_ref = os.path.join(ROOT, "data", "test_scatterPlot_lines")
56class ScatterPlotWithTwoHistsTaskTestCase(lsst.utils.tests.TestCase):
57 """ScatterPlotWithTwoHistsTask test case."""
59 def setUp(self):
60 self.testDir = tempfile.mkdtemp(dir=ROOT, prefix="test_output")
62 # Set up a quasi-plausible measurement catalog
63 mag = 12.5 + 2.5 * np.log10(np.arange(10, 100000))
64 flux = 10 ** (-0.4 * (mag - (mag[-1] + 1)))
65 rng = np.random.default_rng(0)
66 extendedness = 0.0 + (rng.uniform(size=len(mag)) < 0.99 * (mag - mag[0]) / (mag[-1] - mag[0]))
67 flux_meas = flux + rng.normal(scale=np.sqrt(flux * (1 + extendedness)))
68 flux_err = sqrt(flux_meas * (1 + extendedness))
69 good = (flux_meas / sqrt(flux * (1 + extendedness))) > 3
70 extendedness = extendedness[good]
71 flux = flux[good]
72 flux_meas = flux_meas[good]
73 flux_err = flux_err[good]
75 suffix_x, suffix_y, suffix_stat = "_x", "_y", "_stat"
77 # Configure the plot to show observed vs true mags
78 action = ScatterPlotWithTwoHists(
79 xAxisLabel="mag",
80 yAxisLabel="mag meas - ref",
81 magLabel="mag",
82 plotTypes=[
83 "galaxies",
84 "stars",
85 ],
86 xLims=(20, 30),
87 yLims=(-1000, 1000),
88 addSummaryPlot=False,
89 # Make sure adding a suffix works to produce multiple plots
90 suffix_x=suffix_x,
91 suffix_y=suffix_y,
92 suffix_stat=suffix_stat,
93 )
94 plot = AnalysisTool()
95 plot.produce.plot = action
97 # Load the relevant columns
98 key_flux = "meas_Flux"
99 plot.process.buildActions.fluxes_meas = LoadVector(vectorKey=key_flux)
100 plot.process.buildActions.fluxes_err = LoadVector(vectorKey=f"{key_flux}Err")
101 plot.process.buildActions.fluxes_ref = LoadVector(vectorKey="ref_Flux")
102 plot.process.buildActions.mags_ref = ConvertFluxToMag(
103 vectorKey=plot.process.buildActions.fluxes_ref.vectorKey
104 )
106 # Compute the y-axis quantity
107 plot.process.buildActions.diff = SubtractVector(
108 actionA=ConvertFluxToMag(
109 vectorKey=plot.process.buildActions.fluxes_meas.vectorKey, returnMillimags=True
110 ),
111 actionB=DivideVector(
112 actionA=plot.process.buildActions.mags_ref,
113 actionB=ConstantValue(value=1e-3),
114 ),
115 )
117 # Filter stars/galaxies, storing quantities separately
118 plot.process.buildActions.galaxySelector = GalaxySelector(vectorKey="refExtendedness")
119 plot.process.buildActions.starSelector = StarSelector(vectorKey="refExtendedness")
120 for singular, plural in (("galaxy", "Galaxies"), ("star", "Stars")):
121 setattr(
122 plot.process.filterActions,
123 f"x{plural}{suffix_x}",
124 DownselectVector(
125 vectorKey="mags_ref", selector=VectorSelector(vectorKey=f"{singular}Selector")
126 ),
127 )
128 setattr(
129 plot.process.filterActions,
130 f"y{plural}{suffix_y}",
131 DownselectVector(vectorKey="diff", selector=VectorSelector(vectorKey=f"{singular}Selector")),
132 )
133 setattr(
134 plot.process.filterActions,
135 f"flux{plural}",
136 DownselectVector(
137 vectorKey="fluxes_meas", selector=VectorSelector(vectorKey=f"{singular}Selector")
138 ),
139 )
140 setattr(
141 plot.process.filterActions,
142 f"fluxErr{plural}",
143 DownselectVector(
144 vectorKey="fluxes_err", selector=VectorSelector(vectorKey=f"{singular}Selector")
145 ),
146 )
148 # Compute low/high SN summary stats
149 statAction = ScatterPlotStatsAction(
150 vectorKey=f"y{plural}{suffix_y}",
151 fluxType=f"flux{plural}",
152 highSNSelector=SnSelector(fluxType=f"flux{plural}", threshold=50),
153 lowSNSelector=SnSelector(fluxType=f"flux{plural}", threshold=20),
154 suffix=suffix_stat,
155 )
156 setattr(plot.process.calculateActions, plural.lower(), statAction)
158 data = {
159 "ref_Flux": flux,
160 key_flux: flux_meas,
161 f"{key_flux}Err": flux_err,
162 "refExtendedness": extendedness,
163 }
165 self.data = pd.DataFrame(data)
166 print(self.data.columns)
167 self.plot = plot
168 self.plot.finalize()
169 plotInfo = {key: "test" for key in ("plotName", "run", "tableName")}
170 plotInfo["bands"] = []
171 self.plotInfo = plotInfo
173 def tearDown(self):
174 if os.path.exists(self.testDir): 174 ↛ 176line 174 didn't jump to line 176 because the condition on line 174 was always true
175 shutil.rmtree(self.testDir, True)
176 del self.data
177 del self.plot
178 del self.plotInfo
179 del self.testDir
181 def test_ScatterPlotWithTwoHistsTask(self):
182 plt.rcParams.update(plt.rcParamsDefault)
183 result = self.plot(
184 data=self.data,
185 skymap=None,
186 plotInfo=self.plotInfo,
187 )
188 # unpack the result from the dictionary
189 result = result[type(self.plot.produce.plot).__name__]
190 self.assertTrue(isinstance(result, plt.Figure))
192 # Set to true to save plots as PNGs
193 # Use matplotlib.testing.compare.compare_images if needed
194 save_images = False
195 if save_images: 195 ↛ 196line 195 didn't jump to line 196 because the condition on line 195 was never true
196 result.savefig(os.path.join(ROOT, "data", "test_scatterPlot.png"))
198 texts, lines = get_and_remove_figure_text(result)
199 if save_images: 199 ↛ 200line 199 didn't jump to line 200 because the condition on line 199 was never true
200 result.savefig(os.path.join(ROOT, "data", "test_scatterPlot_unlabeled.png"))
202 # Set to true to re-generate reference data
203 resave = False
205 # Compare line values
206 for idx, line in enumerate(lines):
207 filename = os.path.join(path_lines_ref, f"line_{idx}.txt")
208 if resave: 208 ↛ 209line 208 didn't jump to line 209 because the condition on line 208 was never true
209 np.savetxt(filename, line)
210 arr = np.loadtxt(filename)
211 # Differences of order 1e-12 possible between MacOS and Linux
212 # Plots are generally not expected to be that precise
213 # Differences to 1e-3 should not be visible with this test data
214 self.assertFloatsAlmostEqual(arr, line, atol=1e-3, rtol=1e-4)
216 # Ensure that newlines within labels are replaced by a sentinel
217 newline = "\n"
218 newline_replace = "[newline]"
219 # Compare text labels
220 if resave: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true
221 with open(filename_texts_ref, "w") as f:
222 f.writelines(f"{text.strip().replace(newline, newline_replace)}\n" for text in texts)
224 with open(filename_texts_ref, "r") as f:
225 texts_ref = set(x.strip() for x in f.readlines())
226 texts_set = set(x.strip().replace(newline, newline_replace) for x in texts)
228 self.assertEqual(texts_ref, texts_set)
231class MemoryTester(lsst.utils.tests.MemoryTestCase):
232 pass
235def setup_module(module):
236 lsst.utils.tests.init()
239if __name__ == "__main__": 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 lsst.utils.tests.init()
241 unittest.main()