Coverage for python/lsst/images/fields/_product.py: 51%

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__ = ("ProductField", "ProductFieldSerializationModel") 

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 ProductBoundedField as LegacyProductBoundedField 

32 except ImportError: 

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

34 

35 from ._concrete import Field, FieldSerializationModel 

36 

37 

38@final 

39class ProductField(BaseField): 

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

41 

42 Parameters 

43 ---------- 

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

45 The fields to multiply 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 not None: 

59 if self._unit is None: 

60 self._unit = operand.unit 

61 else: 

62 self._unit *= operand.unit 

63 

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

65 if type(other) is not ProductField: 

66 return NotImplemented 

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

68 # comparing the operand tuple is sufficient. 

69 return self._operands == other._operands 

70 

71 __hash__ = None # type: ignore[assignment] 

72 

73 @property 

74 def bounds(self) -> Bounds: 

75 return self._bounds 

76 

77 @property 

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

79 return self._unit 

80 

81 @property 

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

83 """The fields that are multiplied together 

84 (`tuple` [`BaseField`, ...]). 

85 """ 

86 return self._operands 

87 

88 @property 

89 def is_constant(self) -> bool: 

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

91 

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

93 """Return a `Report` describing this product field. 

94 

95 Parameters 

96 ---------- 

97 options : `DescribeOptions`, optional 

98 Rendering options; forwarded to the operand children. 

99 """ 

100 return Report( 

101 type_name="ProductField", 

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

103 fields=[ 

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

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

106 ], 

107 children=( 

108 {} 

109 if options.brief 

110 else { 

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

112 } 

113 ), 

114 ) 

115 

116 def _evaluate( 

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

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

119 iterator = iter(self._operands) 

120 first = next(iterator) 

121 result = first(x=x, y=y, quantity=False) 

122 for operand in iterator: 

123 result *= operand(x=x, y=y, quantity=False) 

124 if quantity: 

125 return result * self.unit 

126 return result 

127 

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

129 if bbox is None: 

130 bbox = self.bounds.bbox 

131 result = Image(1.0, bbox=bbox, dtype=dtype, unit=self.unit) 

132 for operand in self._operands: 

133 result.array *= operand.render(bbox, dtype=dtype).array 

134 return result 

135 

136 def _multiply_constant( 

137 self, factor: float | astropy.units.Quantity | astropy.units.UnitBase 

138 ) -> ProductField: 

139 new_operands = list(self._operands[:-1]) 

140 new_operands.append(self._operands[-1] * factor) 

141 return ProductField(new_operands) 

142 

143 def serialize(self, archive: OutputArchive[Any]) -> ProductFieldSerializationModel: 

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

145 

146 Parameters 

147 ---------- 

148 archive 

149 Archive to write to. 

150 """ 

151 return ProductFieldSerializationModel( 

152 operands=[operand.serialize(archive) for operand in self._operands] 

153 ) 

154 

155 @staticmethod 

156 def _get_archive_tree_type( 

157 pointer_type: type[Any], 

158 ) -> type[ProductFieldSerializationModel]: 

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

160 type that uses the given pointer type. 

161 """ 

162 return ProductFieldSerializationModel 

163 

164 @staticmethod 

165 def from_legacy( 

166 legacy: LegacyProductBoundedField, 

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

168 bounds: Bounds | None = None, 

169 ) -> ProductField: 

170 """Convert from a legacy `lsst.afw.math.ProductBoundedField`. 

171 

172 Parameters 

173 ---------- 

174 legacy 

175 Legacy field to convert. 

176 unit 

177 The units of the returned field (`lsst.afw.math.BoundedField` 

178 objects do not know their units). 

179 bounds 

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

181 the bounding box of ``legacy``. 

182 """ 

183 from ._concrete import field_from_legacy 

184 

185 legacy_factors = legacy.getFactors() 

186 operands = [field_from_legacy(f, bounds=bounds) for f in legacy_factors[:-1]] 

187 operands.append(field_from_legacy(legacy_factors[-1], unit=unit, bounds=bounds)) 

188 return ProductField(operands) 

189 

190 def to_legacy(self) -> LegacyProductBoundedField: 

191 """Convert to a legacy `lsst.afw.math.ProductBoundedField`.""" 

192 from lsst.afw.math import ProductBoundedField 

193 

194 # Not all Field types have a to_legacy, since they don't all have an 

195 # afw analog. But we just let that "no method" exception propagate. 

196 return ProductBoundedField([operand.to_legacy() for operand in self._operands]) 

197 

198 

199class ProductFieldSerializationModel(ArchiveTree): 

200 """Serialization model for `ProductField`.""" 

201 

202 SCHEMA_NAME: ClassVar[str] = "product_field" 

203 SCHEMA_VERSION: ClassVar[str] = "1.0.0" 

204 MIN_READ_VERSION: ClassVar[int] = 1 

205 PUBLIC_TYPE: ClassVar[type] = ProductField 

206 

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

208 

209 field_type: Literal["PRODUCT"] = "PRODUCT" 

210 

211 def deserialize(self, archive: InputArchive, **kwargs: Any) -> ProductField: 

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

213 

214 Parameters 

215 ---------- 

216 archive 

217 Archive to read from. 

218 **kwargs 

219 Unsupported keyword arguments are accepted only to provide 

220 better error messages (raising 

221 `.serialization.InvalidParameterError`). 

222 """ 

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

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

225 return ProductField([operand.deserialize(archive) for operand in self.operands])