Coverage for python/lsst/images/fields/_sum.py: 50%

1# This file is part of lsst-images. 

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# Use of this source code is governed by a 3-clause BSD-style 

10# license that can be found in the LICENSE file. 

11 

12from __future__ import annotations 

13 

14__all__ = ("SumField", "SumFieldSerializationModel") 

15 

16from collections.abc import Iterable 

17from typing import TYPE_CHECKING, Any, ClassVar, Literal, final 

18 

19import astropy.units 

20import numpy as np 

21import pydantic 

22 

23from .._geom import Bounds, Box 

24from .._image import Image 

25from ..describe import DescribeOptions, FieldRole, Report, ReportField 

26from ..serialization import ArchiveTree, InputArchive, InvalidParameterError, OutputArchive 

27from ._base import BaseField 

28 

29if TYPE_CHECKING: 

30 try: 

31 from lsst.afw.math import BackgroundList as LegacyBackgroundList 

32 except ImportError: 

33 type LegacyBackgroundList = Any # type: ignore[no-redef] 

34 

35 from ._concrete import Field, FieldSerializationModel 

36 

37 

38@final 

39class SumField(BaseField): 

40 """A field that sums other fields lazily. 

41 

42 Parameters 

43 ---------- 

44 operands : `~collections.abc.Iterable` [ `BaseField` ] 

45 The fields to sum together. 

46 """ 

47 

48 def __init__(self, operands: Iterable[Field]) -> None: 

49 self._operands = tuple(operands) 

50 if not self._operands: 50 ↛ 51line 50 didn't jump to line 51 because the condition on line 50 was never true

51 raise ValueError("At least one operand must be provided.") 

52 iterator = iter(self._operands) 

53 first = next(iterator) 

54 self._bounds = first.bounds 

55 self._unit = first.unit 

56 for operand in iterator: 

57 self._bounds = self._bounds.intersection(operand.bounds) 

58 if operand.unit is None: 

59 if self._unit is not None: 

60 raise astropy.units.UnitConversionError( 

61 "Cannot add a field with no units to a field with units." 

62 ) 

63 elif self._unit is None: 

64 raise astropy.units.UnitConversionError( 

65 "Cannot add a field with units to a field with no units." 

66 ) 

67 else: 

68 # Raise if these units are not sum-compatible. 

69 self._unit.to(operand.unit) 

70 

71 def __eq__(self, other: object) -> bool: 

72 if type(other) is not SumField: 

73 return NotImplemented 

74 # ``_bounds`` and ``_unit`` are derived from the operands, so 

75 # comparing the operand tuple is sufficient. 

76 return self._operands == other._operands 

77 

78 __hash__ = None # type: ignore[assignment] 

79 

80 @property 

81 def bounds(self) -> Bounds: 

82 return self._bounds 

83 

84 @property 

85 def unit(self) -> astropy.units.UnitBase | None: 

86 return self._unit 

87 

88 @property 

89 def operands(self) -> tuple[Field, ...]: 

90 """The fields that are summed together (`tuple` [`BaseField`, ...]).""" 

91 return self._operands 

92 

93 @property 

94 def is_constant(self) -> bool: 

95 return all(operand.is_constant for operand in self._operands) 

96 

97 def _describe(self, options: DescribeOptions = DescribeOptions(), /) -> Report: 

98 """Return a `Report` describing this sum field. 

99 

100 Parameters 

101 ---------- 

102 options : `DescribeOptions`, optional 

103 Rendering options; forwarded to the operand children. 

104 """ 

105 return Report( 

106 type_name="SumField", 

107 summary=f"SumField({len(self._operands)} operands) over {self.bounds}", 

108 fields=[ 

109 ReportField(label="bounds", value=self.bounds, role=FieldRole.DERIVED), 

110 ReportField(label="unit", value=self.unit, role=FieldRole.DERIVED), 

111 ], 

112 children=( 

113 {} 

114 if options.brief 

115 else { 

116 str(i): operand._describe(options.for_child()) for i, operand in enumerate(self._operands) 

117 } 

118 ), 

119 ) 

120 

121 def _evaluate( 

122 self, *, x: np.ndarray, y: np.ndarray, quantity: bool = False 

123 ) -> np.ndarray | astropy.units.Quantity: 

124 iterator = iter(self._operands) 

125 first = next(iterator) 

126 # We have to add quantities if this is a unit-aware field, as the 

127 # terms in the sum might have different-but-compatible units. 

128 result = first(x=x, y=y, quantity=(self.unit is not None)) 

129 for operand in iterator: 

130 result += operand(x=x, y=y, quantity=(self.unit is not None)) 

131 if self.unit is not None and not quantity: 

132 # Caller doesn't want a Quantity back. 

133 assert isinstance(result, astropy.units.Quantity) 

134 return result.to_value(self.unit) 

135 if self.unit is None and quantity: 135 ↛ 137line 135 didn't jump to line 137 because the condition on line 135 was never true

136 # Caller wants a Quantity back even though there's no units. 

137 return astropy.units.Quantity(result) 

138 return result 

139 

140 def render(self, bbox: Box | None = None, *, dtype: np.typing.DTypeLike | None = None) -> Image: 

141 if bbox is None: 

142 bbox = self.bounds.bbox 

143 result = Image(0.0, bbox=bbox, dtype=dtype, unit=self.unit) 

144 for operand in self._operands: 

145 result.quantity += operand.render(bbox, dtype=dtype).quantity 

146 return result 

147 

148 def _multiply_constant(self, factor: float | astropy.units.Quantity | astropy.units.UnitBase) -> SumField: 

149 return SumField([operand * factor for operand in self._operands]) 

150 

151 def serialize(self, archive: OutputArchive[Any]) -> SumFieldSerializationModel: 

152 """Serialize the field to an output archive. 

153 

154 Parameters 

155 ---------- 

156 archive 

157 Archive to write to. 

158 """ 

159 return SumFieldSerializationModel(operands=[operand.serialize(archive) for operand in self._operands]) 

160 

161 @staticmethod 

162 def _get_archive_tree_type( 

163 pointer_type: type[Any], 

164 ) -> type[SumFieldSerializationModel]: 

165 """Return the serialization model type for this object for an archive 

166 type that uses the given pointer type. 

167 """ 

168 return SumFieldSerializationModel 

169 

170 @staticmethod 

171 def from_legacy_background( 

172 legacy_background: LegacyBackgroundList, 

173 bounds: Bounds | None = None, 

174 unit: astropy.units.UnitBase | None = None, 

175 ) -> SumField: 

176 """Convert from a legacy `lsst.afw.math.BackgroundList` instance. 

177 

178 Parameters 

179 ---------- 

180 legacy_background 

181 Legacy background object to convert. 

182 bounds 

183 The bounds of the returned field, if they should be different from 

184 the bounding box of ``legacy_background``. 

185 unit 

186 The units of the returned field (`lsst.afw.math.BackgroundList` 

187 objects do not know their units). 

188 """ 

189 from ._concrete import field_from_legacy_background 

190 

191 return SumField( 

192 [field_from_legacy_background(b, bounds=bounds, unit=unit) for b, *_ in legacy_background] 

193 ) 

194 

195 

196class SumFieldSerializationModel(ArchiveTree): 

197 """Serialization model for `SumField`.""" 

198 

199 SCHEMA_NAME: ClassVar[str] = "sum_field" 

200 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

201 MIN_READ_VERSION: ClassVar[int] = 1 

202 PUBLIC_TYPE: ClassVar[type] = SumField 

203 

204 operands: list[FieldSerializationModel] = pydantic.Field(default_factory=list) 

205 

206 field_type: Literal["SUM"] = "SUM" 

207 

208 def deserialize(self, archive: InputArchive, **kwargs: Any) -> SumField: 

209 """Deserialize the field from an input archive. 

210 

211 Parameters 

212 ---------- 

213 archive 

214 Archive to read from. 

215 **kwargs 

216 Unsupported keyword arguments are accepted only to provide 

217 better error messages (raising 

218 `.serialization.InvalidParameterError`). 

219 """ 

220 if kwargs: 220 ↛ 221line 220 didn't jump to line 221 because the condition on line 220 was never true

221 raise InvalidParameterError(f"Unrecognized parameters for SumField: {set(kwargs.keys())}.") 

222 return SumField([operand.deserialize(archive) for operand in self.operands])