Coverage for python / lsst / analysis / tools / actions / vector / mathActions.py: 59%

116 statements  

« prev     ^ index     » next       coverage.py v7.13.5, created at 2026-05-06 09:07 +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 <https://www.gnu.org/licenses/>. 

21from __future__ import annotations 

22 

23__all__ = ( 

24 "ConstantValue", 

25 "AddVector", 

26 "SubtractVector", 

27 "MultiplyVector", 

28 "DivideVector", 

29 "SquareVector", 

30 "SqrtVector", 

31 "RaiseFromBaseVector", 

32 "RaiseToPowerVector", 

33 "Log10Vector", 

34 "FractionalDifference", 

35 "CosVector", 

36 "SinVector", 

37) 

38 

39import logging 

40 

41import numpy as np 

42 

43from lsst.pex.config import Field 

44from lsst.pex.config.configurableActions import ConfigurableActionField 

45 

46from ...interfaces import KeyedData, KeyedDataSchema, Vector, VectorAction 

47from ...math import cos, divide, log10, sin, sqrt 

48 

49_LOG = logging.getLogger(__name__) 

50 

51 

52class ConstantValue(VectorAction): 

53 """Return a constant scalar value.""" 

54 

55 value = Field[float](doc="A single constant value", optional=False) 

56 

57 def getInputSchema(self) -> KeyedDataSchema: 

58 return () 

59 

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

61 return np.array([self.value]) 

62 

63 

64class AddVector(VectorAction): 

65 """Calculate (A+B).""" 

66 

67 actionA = ConfigurableActionField[VectorAction](doc="Action which supplies vector A") 

68 actionB = ConfigurableActionField[VectorAction](doc="Action which supplies vector B") 

69 

70 def getInputSchema(self) -> KeyedDataSchema: 

71 yield from self.actionA.getInputSchema() # type: ignore 

72 yield from self.actionB.getInputSchema() # type: ignore 

73 

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

75 vecA = self.actionA(data, **kwargs) # type: ignore 

76 vecB = self.actionB(data, **kwargs) # type: ignore 

77 return vecA + vecB 

78 

79 

80class SubtractVector(VectorAction): 

81 """Calculate (A-B).""" 

82 

83 actionA = ConfigurableActionField[VectorAction](doc="Action which supplies vector A") 

84 actionB = ConfigurableActionField[VectorAction](doc="Action which supplies vector B") 

85 

86 def getInputSchema(self) -> KeyedDataSchema: 

87 yield from self.actionA.getInputSchema() # type: ignore 

88 yield from self.actionB.getInputSchema() # type: ignore 

89 

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

91 vecA = self.actionA(data, **kwargs) # type: ignore 

92 vecB = self.actionB(data, **kwargs) # type: ignore 

93 return vecA - vecB 

94 

95 

96class MultiplyVector(VectorAction): 

97 """Calculate (A*B)""" 

98 

99 actionA = ConfigurableActionField[VectorAction](doc="Action which supplies vector A") 

100 actionB = ConfigurableActionField[VectorAction](doc="Action which supplies vector B") 

101 

102 def getInputSchema(self) -> KeyedDataSchema: 

103 yield from self.actionA.getInputSchema() # type: ignore 

104 yield from self.actionB.getInputSchema() # type: ignore 

105 

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

107 vecA = self.actionA(data, **kwargs) # type: ignore 

108 vecB = self.actionB(data, **kwargs) # type: ignore 

109 return vecA * vecB 

110 

111 

112class DivideVector(VectorAction): 

113 """Calculate (A/B)""" 

114 

115 actionA = ConfigurableActionField[VectorAction](doc="Action which supplies vector A") 

116 actionB = ConfigurableActionField[VectorAction](doc="Action which supplies vector B") 

117 

118 def getInputSchema(self) -> KeyedDataSchema: 

119 yield from self.actionA.getInputSchema() # type: ignore 

120 yield from self.actionB.getInputSchema() # type: ignore 

121 

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

123 vecA = self.actionA(data, **kwargs) # type: ignore 

124 vecB = self.actionB(data, **kwargs) # type: ignore 

125 return divide(vecA, vecB) 

126 

127 

128class SqrtVector(VectorAction): 

129 """Calculate sqrt(A)""" 

130 

131 actionA = ConfigurableActionField(doc="Action which supplies vector A", dtype=VectorAction) 

132 

133 def getInputSchema(self) -> KeyedDataSchema: 

134 yield from self.actionA.getInputSchema() # type: ignore 

135 

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

137 vecA = self.actionA(data, **kwargs) # type: ignore 

138 return sqrt(vecA) 

139 

140 

141class SquareVector(VectorAction): 

142 """Calculate A**2""" 

143 

144 actionA = ConfigurableActionField(doc="Action which supplies vector A", dtype=VectorAction) 

145 

146 def getInputSchema(self) -> KeyedDataSchema: 

147 yield from self.actionA.getInputSchema() # type: ignore 

148 

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

150 vecA = self.actionA(data, **kwargs) # type: ignore 

151 return vecA * vecA 

152 

153 

154class RaiseFromBaseVector(VectorAction): 

155 """Calculate n**A""" 

156 

157 actionA = ConfigurableActionField(doc="Action which supplies vector A", dtype=VectorAction) 

158 base = Field[float](doc="The base value to raise to the power of vector values") 

159 

160 def getInputSchema(self) -> KeyedDataSchema: 

161 yield from self.actionA.getInputSchema() # type: ignore 

162 

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

164 vecA = self.actionA(data, **kwargs) # type: ignore 

165 return self.base**vecA 

166 

167 

168class RaiseToPowerVector(VectorAction): 

169 """Calculate A**n""" 

170 

171 actionA = ConfigurableActionField(doc="Action which supplies vector A", dtype=VectorAction) 

172 power = Field[float](doc="The power to raise the vector to") 

173 

174 def getInputSchema(self) -> KeyedDataSchema: 

175 yield from self.actionA.getInputSchema() # type: ignore 

176 

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

178 vecA = self.actionA(data, **kwargs) # type: ignore 

179 return vecA**self.power 

180 

181 

182class Log10Vector(VectorAction): 

183 """Calculate log10(A)""" 

184 

185 actionA = ConfigurableActionField(doc="Action which supplies vector A", dtype=VectorAction) 

186 

187 def getInputSchema(self) -> KeyedDataSchema: 

188 yield from self.actionA.getInputSchema() # type: ignore 

189 

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

191 vecA = self.actionA(data, **kwargs) # type: ignore 

192 return log10(vecA) 

193 

194 

195class FractionalDifference(VectorAction): 

196 """Calculate (A-B)/B.""" 

197 

198 actionA = ConfigurableActionField[VectorAction](doc="Action which supplies vector A") 

199 actionB = ConfigurableActionField[VectorAction](doc="Action which supplies vector B") 

200 

201 def getInputSchema(self) -> KeyedDataSchema: 

202 yield from self.actionA.getInputSchema() # type: ignore 

203 yield from self.actionB.getInputSchema() # type: ignore 

204 

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

206 vecA = self.actionA(data, **kwargs) # type: ignore 

207 vecB = self.actionB(data, **kwargs) # type: ignore 

208 return divide(vecA - vecB, vecB) 

209 

210 

211class CosVector(VectorAction): 

212 """Calculate cos(A)""" 

213 

214 actionA = ConfigurableActionField(doc="Action which supplies vector A", dtype=VectorAction) 

215 

216 def getInputSchema(self) -> KeyedDataSchema: 

217 yield from self.actionA.getInputSchema() # type: ignore 

218 

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

220 vecA = self.actionA(data, **kwargs) # type: ignore 

221 return cos(vecA) 

222 

223 

224class SinVector(VectorAction): 

225 """Calculate sin(A)""" 

226 

227 actionA = ConfigurableActionField(doc="Action which supplies vector A", dtype=VectorAction) 

228 

229 def getInputSchema(self) -> KeyedDataSchema: 

230 yield from self.actionA.getInputSchema() # type: ignore 

231 

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

233 vecA = self.actionA(data, **kwargs) # type: ignore 

234 return sin(vecA)