Coverage for tests / test_contexts.py: 53%
104 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-05-07 08: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 <http://www.gnu.org/licenses/>.
21from __future__ import annotations
23import warnings
24from typing import cast
25from unittest import TestCase, main
27import astropy.units as apu
28import numpy as np
30import lsst.utils.tests
31from lsst.analysis.tools.actions.scalar import MeanAction, MedianAction
32from lsst.analysis.tools.contexts import Context
33from lsst.analysis.tools.interfaces import (
34 AnalysisAction,
35 AnalysisTool,
36 KeyedData,
37 KeyedDataAction,
38 KeyedDataSchema,
39 KeyedResults,
40 Scalar,
41 ScalarAction,
42 Vector,
43)
44from lsst.pex.config import Field
45from lsst.pex.config.configurableActions import ConfigurableActionField, ConfigurableActionStructField
46from lsst.verify import Measurement
49class MedianContext(Context):
50 """Test Context for median"""
52 pass
55class MeanContext(Context):
56 """Test Context for mean"""
58 pass
61class MultiplyContext(Context):
62 """Test Context to multiply results"""
64 pass
67class DivideContext(Context):
68 """Test Context to divide result"""
70 pass
73class TestAction1(KeyedDataAction):
74 __test__ = False # Tell pytest that this is *not* a test suite
76 multiple = ConfigurableActionStructField[ScalarAction](doc="Multiple Actions")
78 def getInputSchema(self) -> KeyedDataSchema:
79 return (("a", Vector),)
81 def medianContext(self) -> None:
82 self.multiple.a = MedianAction(vectorKey="a")
83 self.multiple.b = MedianAction(vectorKey="a")
84 self.multiple.c = MedianAction(vectorKey="a")
86 def meanContext(self) -> None:
87 self.multiple.a = MeanAction(vectorKey="a")
88 self.multiple.b = MeanAction(vectorKey="a")
90 def __call__(self, data: KeyedData, **kwargs) -> KeyedData:
91 result = np.array([action(data, **kwargs) for action in self.multiple])
92 return cast(KeyedData, {"b": result})
95class TestAction2(KeyedDataAction):
96 __test__ = False # Tell pytest that this is *not* a test suite
98 single = ConfigurableActionField[ScalarAction](doc="Single Action")
99 multiplier = ConfigurableActionField[AnalysisAction](doc="Multiplier")
101 def getInputSchema(self) -> KeyedDataSchema:
102 return (("b", Vector),)
104 def setDefaults(self) -> None:
105 super().setDefaults()
106 # This action remains constant in every context, but it will be
107 # recursively configured with any contexts
108 self.multiplier = TestAction3()
110 def medianContext(self) -> None:
111 self.single = MedianAction(vectorKey="b")
113 def meanContext(self) -> None:
114 self.single = MeanAction(vectorKey="b")
116 def __call__(self, data: KeyedData, **kwargs) -> KeyedData:
117 result = self.single(data, **kwargs)
118 intermediate = self.multiplier(cast(KeyedData, {"c": result}), **kwargs)["d"]
119 intermediate = cast(Measurement, intermediate)
120 assert intermediate.quantity is not None
121 result *= cast(Scalar, intermediate.quantity.value)
122 return {"c": result}
125class TestAction3(KeyedDataAction):
126 __test__ = False # Tell pytest that this is *not* a test suite
128 multiplier = Field[float](doc="Number to multiply result by")
130 def getInputSchema(self) -> KeyedDataSchema:
131 return (("c", Vector),)
133 def multiplyContext(self):
134 self.multiplier = 2
136 def divideContext(self):
137 self.multiplier = 0.5
139 def __call__(self, data: KeyedData, **kwargs) -> KeyedResults:
140 result = Measurement("TestMeasurement", cast(Scalar, data["c"]) * self.multiplier * apu.Unit("count"))
141 return {"d": result}
144class TestAnalysisTool(AnalysisTool):
145 __test__ = False # Tell pytest that this is *not* a test suite
147 def setDefaults(self) -> None:
148 self.prep = TestAction1()
149 self.process = TestAction2()
150 self.produce = TestAction3()
153class ContextTestCase(TestCase):
154 def setUp(self) -> None:
155 super().setUp()
156 self.array = np.arange(20)
157 self.input = cast(KeyedData, {"a": self.array})
159 def testContext1(self):
160 tester = TestAnalysisTool()
161 # test applying contexts serially
162 tester.applyContext(MultiplyContext())
164 # verify assignment syntax works to support Yaml
165 # normally this should be called as a function in python
166 tester.applyContext = MedianContext
167 with warnings.catch_warnings():
168 warnings.simplefilter("ignore")
169 result = cast(Measurement, tester(self.input)["d"])
170 assert result.quantity is not None
171 self.assertEqual(result.quantity.value, 361)
173 def testContext2(self):
174 tester2 = TestAnalysisTool()
175 compound = MeanContext | DivideContext
176 tester2.applyContext(compound)
177 with warnings.catch_warnings():
178 warnings.simplefilter("ignore")
179 result = cast(Measurement, tester2(self.input)["d"])
180 assert result.quantity is not None
181 self.assertEqual(result.quantity.value, 22.5625)
184class MyMemoryTestCase(lsst.utils.tests.MemoryTestCase):
185 pass
188def setup_module(module):
189 lsst.utils.tests.init()
192if __name__ == "__main__": 192 ↛ 193line 192 didn't jump to line 193 because the condition on line 192 was never true
193 lsst.utils.tests.init()
194 main()