Coverage for python / lsst / meas / extensions / scarlet / io / model_data.py: 30%

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 typing import Any 

25 

26import numpy as np 

27from numpy.typing import DTypeLike 

28 

29import lsst.scarlet.lite as scl 

30 

31from .source_data import IsolatedSourceData 

32 

33CURRENT_SCHEMA = "1.0.1" 

34SCARLET_LITE_SCHEMA = "1.0.0" 

35MODEL_TYPE = "lsst" 

36scl.io.migration.MigrationRegistry.set_current(MODEL_TYPE, CURRENT_SCHEMA) 

37 

38# Ensure that the ScarletModelData from scarlet lite hasn't changed. 

39if scl.io.model_data.CURRENT_SCHEMA != SCARLET_LITE_SCHEMA: 39 ↛ 40line 39 didn't jump to line 40 because the condition on line 39 was never true

40 scarletVersion = SCARLET_LITE_SCHEMA.split("."), scl.io.model_data.CURRENT_SCHEMA.split(".") 

41 lsstVersion = CURRENT_SCHEMA.split(".") 

42 

43 outdated = False 

44 if scarletVersion[0] != lsstVersion[0]: 

45 if int(scarletVersion[0]) > int(lsstVersion[0]): 

46 outdated = True 

47 elif scarletVersion[1] != lsstVersion[1]: 

48 if int(scarletVersion[1]) > int(lsstVersion[1]): 

49 outdated = True 

50 elif scarletVersion[2] != lsstVersion[2]: 

51 if int(scarletVersion[2]) > int(lsstVersion[2]): 

52 outdated = True 

53 

54 if outdated: 

55 raise RuntimeError( 

56 "Version mismatch between meas_extensions_scarlet and scarlet lite. " 

57 "This requires updating SCARLET_LITE_SCHEMA, CURRENT_SCHEMA, and a migration step " 

58 f"to match the ScarletModelData schema version {scl.io.model_data.CURRENT_SCHEMA}." 

59 ) 

60 

61 

62class LsstScarletModelData(scl.io.ScarletModelData): 

63 """A ScarletModelData that includes isolated sources. 

64 

65 Attributes 

66 ---------- 

67 isolated : dict[int, IsolatedSourceData] 

68 A mapping of isolated source IDs to their data. 

69 version : dict[int, scl.io.ScarletBlendBaseData] 

70 The schema version of the serialized data. 

71 """ 

72 model_type: str = MODEL_TYPE 

73 isolated: dict[int, IsolatedSourceData] 

74 version: str = CURRENT_SCHEMA 

75 

76 def __init__( 

77 self, 

78 isolated: dict[int, IsolatedSourceData] | None = None, 

79 blends: dict[int, scl.io.ScarletBlendBaseData] | None = None, 

80 metadata: dict[str, Any] | None = None, 

81 ): 

82 super().__init__(blends=blends, metadata=metadata) 

83 self.isolated = isolated if isolated is not None else {} 

84 

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

86 """Convert to a dictionary for serialization 

87 

88 Returns 

89 ------- 

90 result : dict[str, Any] 

91 The object encoded as a JSON-compatible dictionary. 

92 """ 

93 data = super().as_dict() 

94 data.update( 

95 { 

96 "model_type": MODEL_TYPE, 

97 "isolated": {k: v.as_dict() for k, v in self.isolated.items()}, 

98 "version": self.version, 

99 } 

100 ) 

101 return data 

102 

103 @classmethod 

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

105 """Reconstruct `LsstScarletModelData` from JSON compatible dict. 

106 

107 Parameters 

108 ---------- 

109 data : dict 

110 Dictionary representation of the object 

111 dtype : DTypeLike 

112 Datatype of the resulting model. 

113 

114 Returns 

115 ------- 

116 result : LsstScarletModelData 

117 The reconstructed object 

118 """ 

119 data = scl.io.migration.MigrationRegistry.migrate(MODEL_TYPE, data) 

120 isolated: dict[int, IsolatedSourceData] = {} 

121 for sid, source_data in data.get("isolated", {}).items(): 

122 isolated[int(sid)] = IsolatedSourceData.from_dict(source_data, dtype=dtype) 

123 if "metadata" not in data: 

124 data["metadata"] = None 

125 return super().from_dict(data, dtype=dtype, isolated=isolated) 

126 

127 

128@scl.io.migration.migration(MODEL_TYPE, scl.io.migration.PRE_SCHEMA) 

129def _to_1_0_0(data: dict) -> dict: 

130 """Migrate a pre-schema model to schema version 1.0.0 

131 

132 There were no changes to this data model in v1.0.0 but we need 

133 to provide a way to migrate pre-schema data. 

134 

135 Parameters 

136 ---------- 

137 data : dict 

138 The data to migrate. 

139 Returns 

140 ------- 

141 result : dict 

142 The migrated data. 

143 """ 

144 # Ensure that the model type and version are set and add an 

145 # empty isolated sources dictionary. 

146 if "model_type" not in data: 

147 data["model_type"] = MODEL_TYPE 

148 data["isolated"] = {} 

149 data["version"] = "1.0.0" 

150 return data 

151 

152 

153@scl.io.migration.migration(MODEL_TYPE, "1.0.0") 

154def _to_1_0_1(data: dict) -> dict: 

155 """Migrate a schema version 1.0.0 model to schema version 1.0.1 

156 

157 There were no changes to this data model in v1.0.1 but we need 

158 to provide a way to migrate 1.0.0 data. 

159 

160 Parameters 

161 ---------- 

162 data : dict 

163 The data to migrate. 

164 Returns 

165 ------- 

166 result : dict 

167 The migrated data. 

168 """ 

169 data["version"] = "1.0.1" 

170 data.setdefault("metadata", {}).setdefault("footprint", None) 

171 return data