Coverage for python / lsst / analysis / tools / interfaces / _actions.py: 59%

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/>. 

21 

22from __future__ import annotations 

23 

24__all__ = ( 

25 "AnalysisAction", 

26 "HealSparseMapAction", 

27 "KeyedDataAction", 

28 "VectorAction", 

29 "TensorAction", 

30 "ScalarAction", 

31 "MetricResultType", 

32 "MetricAction", 

33 "PlotResultType", 

34 "PlotElement", 

35 "PlotAction", 

36 "JointResults", 

37 "JointAction", 

38 "NoMetric", 

39 "NoPlot", 

40) 

41 

42import warnings 

43from abc import abstractmethod 

44from dataclasses import dataclass 

45from typing import TYPE_CHECKING, Iterable 

46 

47import lsst.pex.config as pexConfig 

48from healsparse.healSparseMap import HealSparseMap 

49from lsst.pex.config.configurableActions import ConfigurableAction, ConfigurableActionField 

50 

51from ..contexts import ContextApplier 

52from ._interfaces import KeyedData, KeyedDataSchema, MetricResultType, PlotResultType, Scalar, Tensor, Vector 

53 

54if TYPE_CHECKING: 

55 from matplotlib.axes import Axes 

56 

57 

58class AnalysisAction(ConfigurableAction): 

59 """Base class interface for the various actions used in analysis tools. 

60 

61 This extends the basic `ConfigurableAction` class to include interfaces for 

62 defining what an action expects to consume, and what it expects to produce. 

63 """ 

64 

65 def __init_subclass__(cls, **kwargs): 

66 if "getInputSchema" not in dir(cls): 66 ↛ 67line 66 didn't jump to line 67 because the condition on line 66 was never true

67 raise NotImplementedError(f"Class {cls} must implement method getInputSchema") 

68 

69 # This is a descriptor that functions like a function in most contexts 

70 # and can be treated as such 

71 applyContext = ContextApplier() 

72 r"""Apply a `Context` to an `AnalysisAction` recursively. 

73 

74 Generally this method is called from within an `AnalysisTool` to 

75 configure all `AnalysisAction`\ s at one time to make sure that they 

76 all are consistently configured. However, it is permitted to call this 

77 method if you are aware of the effects, or from within a specific 

78 execution environment like a python shell or notebook. 

79 

80 Parameters 

81 ---------- 

82 context : `Context` 

83 The specific execution context, this may be a single context or 

84 a joint context, see `Context` for more info. 

85 """ 

86 

87 @abstractmethod 

88 def getInputSchema(self) -> KeyedDataSchema: 

89 """Return the schema an `AnalysisAction` expects to be present in the 

90 arguments supplied to the __call__ method. 

91 

92 Returns 

93 ------- 

94 result : `KeyedDataSchema` 

95 The schema this action requires to be present when calling this 

96 action, keys are unformatted. 

97 """ 

98 raise NotImplementedError("This is not implemented on the base class") 

99 

100 def getOutputSchema(self) -> KeyedDataSchema | None: 

101 """Return the schema an `AnalysisAction` will produce, if the 

102 ``__call__`` method returns `KeyedData`, otherwise this may return 

103 None. 

104 

105 Returns 

106 ------- 

107 result : `KeyedDataSchema` or None 

108 The schema this action will produce when returning from call. This 

109 will be unformatted if any templates are present. Should return 

110 None if action does not return `KeyedData`. 

111 """ 

112 return None 

113 

114 def getFormattedInputSchema(self, **kwargs) -> KeyedDataSchema: 

115 """Return input schema, with keys formatted with any arguments supplied 

116 by kwargs passed to this method. 

117 

118 Returns 

119 ------- 

120 result : `KeyedDataSchema` 

121 The schema this action requires to be present when calling this 

122 action, formatted with any input arguments (e.g. band='i') 

123 """ 

124 for key, typ in self.getInputSchema(): 

125 yield key.format_map(kwargs), typ 

126 

127 def addInputSchema(self, inputSchema: KeyedDataSchema) -> None: 

128 """Add the supplied inputSchema argument to the class such that it will 

129 be returned along side any other arguments in a call to 

130 ``getInputSchema``. 

131 

132 Parameters 

133 ---------- 

134 inputSchema : `KeyedDataSchema` 

135 A schema that is to be merged in with any existing schema when a 

136 call to ``getInputSchema`` is made. 

137 """ 

138 warnings.warn( 

139 f"{type(self)} does not implement adding input schemas, call will do nothing, " 

140 "this may be expected", 

141 RuntimeWarning, 

142 ) 

143 

144 

145class HealSparseMapAction(AnalysisAction): 

146 """A `HealSparseMapAction` is an `AnalysisAction` that returns a 

147 `HealSparseMap` when called. 

148 """ 

149 

150 @abstractmethod 

151 def __call__(self, data: KeyedData, **kwargs) -> HealSparseMap: 

152 raise NotImplementedError("This is not implemented on the base class") 

153 

154 

155class KeyedDataAction(AnalysisAction): 

156 """A `KeyedDataAction` is an `AnalysisAction` that returns `KeyedData` when 

157 called. 

158 """ 

159 

160 @abstractmethod 

161 def __call__(self, data: KeyedData, **kwargs) -> KeyedData: 

162 raise NotImplementedError("This is not implemented on the base class") 

163 

164 

165class VectorAction(AnalysisAction): 

166 """A `VectorAction` is an `AnalysisAction` that returns a `Vector` when 

167 called. 

168 """ 

169 

170 @abstractmethod 

171 def __call__(self, data: KeyedData, **kwargs) -> Vector: 

172 raise NotImplementedError("This is not implemented on the base class") 

173 

174 

175class TensorAction(AnalysisAction): 

176 """A `TensorAction` is an `AnalysisAction` that returns a `Tensor` when 

177 called. 

178 """ 

179 

180 @abstractmethod 

181 def __call__(self, data: KeyedData, **kwargs) -> Tensor: 

182 raise NotImplementedError("This is not implemented on the base class") 

183 

184 

185class ScalarAction(AnalysisAction): 

186 """A `ScalarAction` is an `AnalysisAction` that returns a `Scalar` when 

187 called. 

188 """ 

189 

190 @abstractmethod 

191 def __call__(self, data: KeyedData, **kwargs) -> Scalar: 

192 """Compute a scalar value from keyed data. 

193 

194 Parameters 

195 ---------- 

196 data 

197 Keyed data to compute a value from. 

198 kwargs 

199 Additional keyword arguments. 

200 

201 Returns 

202 ------- 

203 A scalar value. 

204 """ 

205 raise NotImplementedError("This is not implemented on the base class") 

206 

207 def getMask(self, **kwargs) -> Vector | slice: 

208 """Extract a mask if one is passed as key word args, otherwise return 

209 an empty slice object that can still be used in a getitem call. 

210 

211 Returns 

212 ------- 

213 result : `Vector` or `slice` 

214 The mask passed as a keyword, or a slice object that will return 

215 a complete Vector when used in getitem. 

216 """ 

217 if (mask := kwargs.get("mask")) is None: 

218 mask = slice(None) 

219 return mask 

220 

221 

222class MetricAction(AnalysisAction): 

223 """A `MetricAction` is an `AnalysisAction` that returns a `Measurement` or 

224 a `Mapping` of `str` to `Measurement` when called. 

225 """ 

226 

227 @abstractmethod 

228 def __call__(self, data: KeyedData, **kwargs) -> MetricResultType: 

229 raise NotImplementedError("This is not implemented on the base class") 

230 

231 

232class PlotAction(AnalysisAction): 

233 """A `PlotAction` is an `AnalysisAction` that returns a `PlotType` or 

234 a `Mapping` of `str` to `PlotType` when called. 

235 """ 

236 

237 def getOutputNames(self, config: pexConfig.Config | None = None) -> Iterable[str]: 

238 """Returns a list of names that will be used as keys if this action's 

239 call method returns a mapping. Otherwise return an empty Iterable. 

240 

241 Parameters 

242 ---------- 

243 config : `lsst.pex.config.Config`, optional 

244 Configuration of the task. This is only used if the output naming 

245 needs to be config-aware. 

246 

247 Returns 

248 ------- 

249 result : `Iterable` of `str` 

250 If a `PlotAction` produces more than one plot, this should be the 

251 keys the action will use in the returned `Mapping`. 

252 """ 

253 return tuple() 

254 

255 @abstractmethod 

256 def __call__(self, data: KeyedData, **kwargs) -> PlotResultType: 

257 raise NotImplementedError("This is not implemented on the base class") 

258 

259 def getPlotType(self) -> str: 

260 return type(self).__name__ 

261 

262 

263class PlotElement(AnalysisAction): 

264 """PlotElements are the most basic components of a plot. They can be 

265 composed together within a `PlotAction` to create rich plots. 

266 

267 Plot elements may return metadata about creating their element by returning 

268 `KeyedData` from their call method. 

269 """ 

270 

271 @abstractmethod 

272 def __call__(self, data: KeyedData, ax: Axes, **kwargs) -> KeyedData: 

273 raise NotImplementedError("This is not implemented on the base class") 

274 

275 

276class NoPlot(PlotAction): 

277 """This is a sentinel class to indicate that there is no plotting action""" 

278 

279 

280class NoMetric(MetricAction): 

281 """This is a sentinel class to indicate that there is no Metric action""" 

282 

283 

284@dataclass 

285class JointResults: 

286 """The `JointResults` dataclass is a container for the results of a 

287 `JointAction`. 

288 """ 

289 

290 plot: PlotResultType | None 

291 metric: MetricResultType | None 

292 

293 

294class JointAction(AnalysisAction): 

295 """A `JointAction` is an `AnalysisAction` that is a composite of a 

296 `PlotAction` and a `MetricAction`. 

297 """ 

298 

299 metric = ConfigurableActionField[MetricAction](doc="Action to run that will produce one or more metrics") 

300 plot = ConfigurableActionField[PlotAction](doc="Action to run that will produce one or more plots") 

301 

302 def __call__(self, data: KeyedData, **kwargs) -> JointResults: 

303 if isinstance(self.plot, NoPlot): 

304 plots = None 

305 else: 

306 plots = self.plot(data, **kwargs) 

307 if isinstance(self.metric, NoMetric): 

308 metrics = None 

309 else: 

310 metrics = self.metric(data, **kwargs) 

311 return JointResults(plots, metrics) 

312 

313 def getInputSchema(self) -> KeyedDataSchema: 

314 yield from self.metric.getInputSchema() 

315 yield from self.plot.getInputSchema() 

316 

317 def getOutputNames(self, config: pexConfig.Config | None = None) -> Iterable[str]: 

318 """Returns a list of names that will be used as keys if this action's 

319 call method returns a mapping. Otherwise return an empty Iterable. 

320 

321 Parameters 

322 ---------- 

323 config : `lsst.pex.config.Config`, optional 

324 Configuration of the task. This is only used if the output naming 

325 needs to be config-aware. 

326 

327 Returns 

328 ------- 

329 outNames : `Iterable` of `str` 

330 If a `PlotAction` produces more than one plot, this should be the 

331 keys the action will use in the returned `Mapping`. 

332 """ 

333 if config is None: 

334 # `dynamicOutputNames` is set to False. 

335 outNames = self.plot.getOutputNames() 

336 else: 

337 # `dynamicOutputNames` is set to True. 

338 outNames = self.plot.getOutputNames(config=config) 

339 return outNames