Coverage for tests / test_contexts.py: 53%
104 statements
« prev ^ index » next coverage.py v7.13.5, created at 2026-04-22 09:09 +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 lsst.utils.tests
29import numpy as np
30from lsst.analysis.tools.actions.scalar import MeanAction, MedianAction
31from lsst.analysis.tools.contexts import Context
32from lsst.analysis.tools.interfaces import (
33 AnalysisAction,
34 AnalysisTool,
35 KeyedData,
36 KeyedDataAction,
37 KeyedDataSchema,
38 KeyedResults,
39 Scalar,
40 ScalarAction,
41 Vector,
42)
43from lsst.pex.config import Field
44from lsst.pex.config.configurableActions import ConfigurableActionField, ConfigurableActionStructField
45from lsst.verify import Measurement
48class MedianContext(Context):
49 """Test Context for median"""
51 pass
54class MeanContext(Context):
55 """Test Context for mean"""
57 pass
60class MultiplyContext(Context):
61 """Test Context to multiply results"""
63 pass
66class DivideContext(Context):
67 """Test Context to divide result"""
69 pass
72class TestAction1(KeyedDataAction):
73 __test__ = False # Tell pytest that this is *not* a test suite
75 multiple = ConfigurableActionStructField[ScalarAction](doc="Multiple Actions")
77 def getInputSchema(self) -> KeyedDataSchema:
78 return (("a", Vector),)
80 def medianContext(self) -> None:
81 self.multiple.a = MedianAction(vectorKey="a")
82 self.multiple.b = MedianAction(vectorKey="a")
83 self.multiple.c = MedianAction(vectorKey="a")
85 def meanContext(self) -> None:
86 self.multiple.a = MeanAction(vectorKey="a")
87 self.multiple.b = MeanAction(vectorKey="a")
89 def __call__(self, data: KeyedData, **kwargs) -> KeyedData:
90 result = np.array([action(data, **kwargs) for action in self.multiple])
91 return cast(KeyedData, {"b": result})
94class TestAction2(KeyedDataAction):
95 __test__ = False # Tell pytest that this is *not* a test suite
97 single = ConfigurableActionField[ScalarAction](doc="Single Action")
98 multiplier = ConfigurableActionField[AnalysisAction](doc="Multiplier")
100 def getInputSchema(self) -> KeyedDataSchema:
101 return (("b", Vector),)
103 def setDefaults(self) -> None:
104 super().setDefaults()
105 # This action remains constant in every context, but it will be
106 # recursively configured with any contexts
107 self.multiplier = TestAction3()
109 def medianContext(self) -> None:
110 self.single = MedianAction(vectorKey="b")
112 def meanContext(self) -> None:
113 self.single = MeanAction(vectorKey="b")
115 def __call__(self, data: KeyedData, **kwargs) -> KeyedData:
116 result = self.single(data, **kwargs)
117 intermediate = self.multiplier(cast(KeyedData, {"c": result}), **kwargs)["d"]
118 intermediate = cast(Measurement, intermediate)
119 assert intermediate.quantity is not None
120 result *= cast(Scalar, intermediate.quantity.value)
121 return {"c": result}
124class TestAction3(KeyedDataAction):
125 __test__ = False # Tell pytest that this is *not* a test suite
127 multiplier = Field[float](doc="Number to multiply result by")
129 def getInputSchema(self) -> KeyedDataSchema:
130 return (("c", Vector),)
132 def multiplyContext(self):
133 self.multiplier = 2
135 def divideContext(self):
136 self.multiplier = 0.5
138 def __call__(self, data: KeyedData, **kwargs) -> KeyedResults:
139 result = Measurement("TestMeasurement", cast(Scalar, data["c"]) * self.multiplier * apu.Unit("count"))
140 return {"d": result}
143class TestAnalysisTool(AnalysisTool):
144 __test__ = False # Tell pytest that this is *not* a test suite
146 def setDefaults(self) -> None:
147 self.prep = TestAction1()
148 self.process = TestAction2()
149 self.produce = TestAction3()
152class ContextTestCase(TestCase):
153 def setUp(self) -> None:
154 super().setUp()
155 self.array = np.arange(20)
156 self.input = cast(KeyedData, {"a": self.array})
158 def testContext1(self):
159 tester = TestAnalysisTool()
160 # test applying contexts serially
161 tester.applyContext(MultiplyContext())
163 # verify assignment syntax works to support Yaml
164 # normally this should be called as a function in python
165 tester.applyContext = MedianContext
166 with warnings.catch_warnings():
167 warnings.simplefilter("ignore")
168 result = cast(Measurement, tester(self.input)["d"])
169 assert result.quantity is not None
170 self.assertEqual(result.quantity.value, 361)
172 def testContext2(self):
173 tester2 = TestAnalysisTool()
174 compound = MeanContext | DivideContext
175 tester2.applyContext(compound)
176 with warnings.catch_warnings():
177 warnings.simplefilter("ignore")
178 result = cast(Measurement, tester2(self.input)["d"])
179 assert result.quantity is not None
180 self.assertEqual(result.quantity.value, 22.5625)
183class MyMemoryTestCase(lsst.utils.tests.MemoryTestCase):
184 pass
187def setup_module(module):
188 lsst.utils.tests.init()
191if __name__ == "__main__": 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 lsst.utils.tests.init()
193 main()