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 "source_type": self.source_type, 

83 "version": self.version, 

84 } 

85 if self.metadata is not None: 

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

87 return result 

88 

89 @classmethod 

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

91 """Reconstruct `IsolatedSourceData` from JSON compatible 

92 dict. 

93 

94 Parameters 

95 ---------- 

96 data : dict 

97 Dictionary representation of the object 

98 dtype : DTypeLike 

99 Datatype of the resulting model. 

100 

101 Returns 

102 ------- 

103 result : IsolatedSourceData 

104 The reconstructed object. 

105 """ 

106 data = scl.io.migration.MigrationRegistry.migrate(SOURCE_TYPE, data) 

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

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

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

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

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

112 return cls( 

113 span_array=span_array, 

114 origin=origin, 

115 peak=peak, 

116 metadata=metadata, 

117 ) 

118 

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

120 """Convert to a scarlet Source object 

121 

122 Parameters 

123 ---------- 

124 observation : scl.Observation 

125 The observation of the source. 

126 

127 Returns 

128 ------- 

129 result : IsolatedSourceData 

130 The scarlet Source object. 

131 """ 

132 # Extract the image data that overlaps with the Footprint 

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

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

135 

136 # Mask the image data with the footprint spans 

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

138 

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

140 model = scl.Image( 

141 model_data, 

142 yx0=bbox.origin, 

143 bands=observation.bands, 

144 ) 

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

146 

147 

148IsolatedSourceData.register()