Coverage for python/lsst/meas/extensions/scarlet/io/source_data.py: 56%

1# This file is part of meas_extensions_scarlet. 

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 

24from dataclasses import dataclass 

25from typing import Any 

26 

27import numpy as np 

28from numpy.typing import DTypeLike 

29 

30import lsst.scarlet.lite as scl 

31from ..source import IsolatedSource 

32 

33__all__ = ["IsolatedSourceData"] 

34 

35CURRENT_SCHEMA = "1.0.0" 

36SOURCE_TYPE = "isolated" 

37scl.io.migration.MigrationRegistry.set_current(SOURCE_TYPE, CURRENT_SCHEMA) 

38 

39 

40@dataclass(kw_only=True) 

41class IsolatedSourceData(scl.io.blend.ScarletSourceBaseData): 

42 """A source data instance of an isolated source. 

43 

44 This is used to represent sources that were not blended with any 

45 other sources, and therefore do not have any deblending information. 

46 

47 Attributes 

48 ---------- 

49 source_type : str 

50 The type of source data. 

51 version : str 

52 The schema version of the serialized data. 

53 span_array : np.ndarray 

54 The span mask of the source. 

55 origin : tuple[int, int] 

56 The (y, x) origin of the footprint in the observation. 

57 peak : tuple[int, int] 

58 The (y, x) coordinates of the source peak in the observation. 

59 metadata : dict | None 

60 Additional metadata associated with the source. 

61 """ 

62 

63 source_type: str = SOURCE_TYPE 

64 version: str = CURRENT_SCHEMA 

65 span_array: np.ndarray 

66 origin: tuple[int, int] 

67 peak: tuple[int, int] 

68 

69 def as_dict(self) -> dict[str, Any]: 

70 """Convert to a dictionary for serialization 

71 

72 Returns 

73 ------- 

74 result : dict[str, Any] 

75 The object encoded as a JSON-compatible dictionary. 

76 """ 

77 result: dict[str, Any] = { 

78 "origin": tuple(int(o) for o in self.origin), 

79 "shape": tuple(int(s) for s in self.span_array.shape), 

80 "peak": tuple(float(p) for p in self.peak), 

81 "span_array": tuple(self.span_array.flatten().astype(float)), 

82 "version": self.version, 

83 } 

84 if self.metadata is not None: 

85 result["metadata"] = scl.io.utils.encode_metadata(self.metadata) 

86 return result 

87 

88 @classmethod 

89 def from_dict(cls, data: dict, dtype: DTypeLike = np.float32) -> IsolatedSourceData: 

90 """Reconstruct `IsolatedSourceData` from JSON compatible 

91 dict. 

92 

93 Parameters 

94 ---------- 

95 data : dict 

96 Dictionary representation of the object 

97 dtype : DTypeLike 

98 Datatype of the resulting model. 

99 

100 Returns 

101 ------- 

102 result : IsolatedSourceData 

103 The reconstructed object. 

104 """ 

105 data = scl.io.MigrationRegistry.migrate(SOURCE_TYPE, data) 

106 shape = tuple(int(s) for s in data["shape"]) 

107 origin = tuple(int(o) for o in data["origin"]) 

108 span_array = np.array(data["span_array"], dtype=dtype).reshape(shape) 

109 peak = tuple(int(p) for p in data["peak"]) 

110 metadata = scl.io.utils.decode_metadata(data.get("metadata", None)) 

111 return cls( 

112 span_array=span_array, 

113 origin=origin, 

114 peak=peak, 

115 metadata=metadata, 

116 ) 

117 

118 def to_source(self, observation: scl.Observation) -> IsolatedSourceData: 

119 """Convert to a scarlet Source object 

120 

121 Parameters 

122 ---------- 

123 observation : scl.Observation 

124 The observation of the source. 

125 

126 Returns 

127 ------- 

128 result : IsolatedSourceData 

129 The scarlet Source object. 

130 """ 

131 # Extract the image data that overlaps with the Footprint 

132 bbox = scl.Box(self.span_array.shape, origin=self.origin) 

133 image_data = observation.images[:, bbox].data 

134 

135 # Mask the image data with the footprint spans 

136 model_data = image_data * self.span_array[None, :, :] 

137 

138 # Convert the array and bounding box into a scarlet Image 

139 model = scl.Image( 

140 model_data, 

141 yx0=bbox.origin, 

142 bands=observation.bands, 

143 ) 

144 return IsolatedSource(model=model, peak=self.peak, metadata=self.metadata) 

145 

146 

147IsolatedSourceData.register()