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 collections.abc import Iterable 

45from dataclasses import dataclass 

46from typing import TYPE_CHECKING 

47 

48from healsparse.healSparseMap import HealSparseMap 

49 

50import lsst.pex.config as pexConfig 

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

52 

53from ..contexts import ContextApplier 

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

55 

56if TYPE_CHECKING: 

57 from matplotlib.axes import Axes 

58 

59 

60class AnalysisAction(ConfigurableAction): 

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

62 

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

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

65 """ 

66 

67 def __init_subclass__(cls, **kwargs): 

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

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

70 

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

72 # and can be treated as such 

73 applyContext = ContextApplier() 

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

75 

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

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

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

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

80 execution environment like a python shell or notebook. 

81 

82 Parameters 

83 ---------- 

84 context : `Context` 

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

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

87 """ 

88 

89 @abstractmethod 

90 def getInputSchema(self) -> KeyedDataSchema: 

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

92 arguments supplied to the __call__ method. 

93 

94 Returns 

95 ------- 

96 result : `KeyedDataSchema` 

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

98 action, keys are unformatted. 

99 """ 

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

101 

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

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

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

105 None. 

106 

107 Returns 

108 ------- 

109 result : `KeyedDataSchema` or None 

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

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

112 None if action does not return `KeyedData`. 

113 """ 

114 return None 

115 

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

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

118 by kwargs passed to this method. 

119 

120 Returns 

121 ------- 

122 result : `KeyedDataSchema` 

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

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

125 """ 

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

127 yield key.format_map(kwargs), typ 

128 

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

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

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

132 ``getInputSchema``. 

133 

134 Parameters 

135 ---------- 

136 inputSchema : `KeyedDataSchema` 

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

138 call to ``getInputSchema`` is made. 

139 """ 

140 warnings.warn( 

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

142 "this may be expected", 

143 RuntimeWarning, 

144 ) 

145 

146 

147class HealSparseMapAction(AnalysisAction): 

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

149 `HealSparseMap` when called. 

150 """ 

151 

152 @abstractmethod 

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

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

155 

156 

157class KeyedDataAction(AnalysisAction): 

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

159 called. 

160 """ 

161 

162 @abstractmethod 

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

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

165 

166 

167class VectorAction(AnalysisAction): 

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

169 called. 

170 """ 

171 

172 @abstractmethod 

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

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

175 

176 

177class TensorAction(AnalysisAction): 

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

179 called. 

180 """ 

181 

182 @abstractmethod 

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

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

185 

186 

187class ScalarAction(AnalysisAction): 

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

189 called. 

190 """ 

191 

192 @abstractmethod 

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

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

195 

196 Parameters 

197 ---------- 

198 data 

199 Keyed data to compute a value from. 

200 kwargs 

201 Additional keyword arguments. 

202 

203 Returns 

204 ------- 

205 A scalar value. 

206 """ 

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

208 

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

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

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

212 

213 Returns 

214 ------- 

215 result : `Vector` or `slice` 

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

217 a complete Vector when used in getitem. 

218 """ 

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

220 mask = slice(None) 

221 return mask 

222 

223 

224class MetricAction(AnalysisAction): 

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

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

227 """ 

228 

229 @abstractmethod 

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

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

232 

233 

234class PlotAction(AnalysisAction): 

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

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

237 """ 

238 

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

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

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

242 

243 Parameters 

244 ---------- 

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

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

247 needs to be config-aware. 

248 

249 Returns 

250 ------- 

251 result : `Iterable` of `str` 

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

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

254 """ 

255 return tuple() 

256 

257 @abstractmethod 

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

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

260 

261 def getPlotType(self) -> str: 

262 return type(self).__name__ 

263 

264 

265class PlotElement(AnalysisAction): 

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

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

268 

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

270 `KeyedData` from their call method. 

271 """ 

272 

273 @abstractmethod 

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

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

276 

277 

278class NoPlot(PlotAction): 

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

280 

281 

282class NoMetric(MetricAction): 

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

284 

285 

286@dataclass 

287class JointResults: 

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

289 `JointAction`. 

290 """ 

291 

292 plot: PlotResultType | None 

293 metric: MetricResultType | None 

294 

295 

296class JointAction(AnalysisAction): 

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

298 `PlotAction` and a `MetricAction`. 

299 """ 

300 

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

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

303 

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

305 if isinstance(self.plot, NoPlot): 

306 plots = None 

307 else: 

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

309 if isinstance(self.metric, NoMetric): 

310 metrics = None 

311 else: 

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

313 return JointResults(plots, metrics) 

314 

315 def getInputSchema(self) -> KeyedDataSchema: 

316 yield from self.metric.getInputSchema() 

317 yield from self.plot.getInputSchema() 

318 

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

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

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

322 

323 Parameters 

324 ---------- 

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

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

327 needs to be config-aware. 

328 

329 Returns 

330 ------- 

331 outNames : `Iterable` of `str` 

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

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

334 """ 

335 if config is None: 

336 # `dynamicOutputNames` is set to False. 

337 outNames = self.plot.getOutputNames() 

338 else: 

339 # `dynamicOutputNames` is set to True. 

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

341 return outNames