Coverage for python / lsst / scarlet / lite / io / source.py: 62%

1from __future__ import annotations 

2 

3import logging 

4from dataclasses import dataclass 

5from typing import Any 

6 

7import numpy as np 

8from deprecated.sphinx import deprecated # type: ignore 

9from numpy.typing import DTypeLike 

10 

11from ..component import Component 

12from ..observation import Observation 

13from ..source import Source 

14from .component import ScarletComponentBaseData 

15from .migration import PRE_SCHEMA, MigrationRegistry, migration 

16from .source_base import ScarletSourceBaseData 

17from .utils import decode_metadata, encode_metadata 

18 

19__all__ = ["ScarletSourceData"] 

20 

21CURRENT_SCHEMA = "1.0.0" 

22SOURCE_TYPE = "source" 

23MigrationRegistry.set_current(SOURCE_TYPE, CURRENT_SCHEMA) 

24logger = logging.getLogger(__name__) 

25 

26 

27@dataclass(kw_only=True) 

28class ScarletSourceData(ScarletSourceBaseData): 

29 """Data for a scarlet source 

30 

31 Attributes 

32 ---------- 

33 components : 

34 The components contained in the source that are not factorized. 

35 metadata : 

36 Metadata associated with the source. 

37 If `metadata` contains the `id` key, it is used as the 

38 key for the source in a `Blend`'s dictionary of sources. 

39 source_type : 

40 The type of source being stored 

41 version : 

42 The schema version of the ScarletSourceData. 

43 """ 

44 

45 source_type: str = SOURCE_TYPE 

46 components: list[ScarletComponentBaseData] 

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

48 version: str = CURRENT_SCHEMA 

49 

50 def as_dict(self) -> dict: 

51 """Return the object encoded into a dict for JSON serialization 

52 

53 Returns 

54 ------- 

55 result : 

56 The object encoded as a JSON compatible dict 

57 """ 

58 result: dict[str, Any] = { 

59 "source_type": self.source_type, 

60 "components": [component.as_dict() for component in self.components], 

61 "version": self.version, 

62 } 

63 if self.metadata is not None: 

64 result["metadata"] = encode_metadata(self.metadata) 

65 return result 

66 

67 @classmethod 

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

69 """Reconstruct `ScarletSourceData` from JSON compatible 

70 dict. 

71 

72 Parameters 

73 ---------- 

74 data : 

75 Dictionary representation of the object 

76 dtype : 

77 Datatype of the resulting model. 

78 

79 Returns 

80 ------- 

81 result : 

82 The reconstructed object 

83 """ 

84 data = MigrationRegistry.migrate(SOURCE_TYPE, data) 

85 metadata = data.get("metadata", None) 

86 components = [ 

87 ScarletComponentBaseData.from_dict(component, dtype=dtype) for component in data["components"] 

88 ] 

89 return cls(components=components, metadata=decode_metadata(metadata)) 

90 

91 def to_source(self, observation: Observation) -> Source: 

92 """Convert to a `Source` for use in scarlet 

93 

94 Parameters 

95 ---------- 

96 observation: 

97 The observation used to render the source. 

98 

99 Returns 

100 ------- 

101 source: 

102 The `Source` representation of this data. 

103 """ 

104 components: list[Component] = [component.to_component(observation) for component in self.components] 

105 return Source(components=components, metadata=self.metadata) 

106 

107 @staticmethod 

108 @deprecated( 

109 reason="from_source is deprecated and will be removed in a future release.", 

110 version="v30.0", 

111 category=FutureWarning, 

112 ) 

113 def from_source(source: Source) -> ScarletSourceData: 

114 """Deprecated: Create a `ScarletSourceData` from a scarlet `Source` 

115 

116 Parameters 

117 ---------- 

118 source: 

119 The scarlet `Source` to convert. 

120 

121 Returns 

122 ------- 

123 result: 

124 The `ScarletSourceData` representation of the source. 

125 """ 

126 return source.to_data() 

127 

128 

129ScarletSourceData.register() 

130 

131 

132@migration(SOURCE_TYPE, PRE_SCHEMA) 

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

134 """Migrate a pre-schema source to schema version 1.0.0 

135 

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

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

138 

139 Parameters 

140 ---------- 

141 data : 

142 The data to migrate. 

143 Returns 

144 ------- 

145 result : 

146 The migrated data. 

147 """ 

148 # Check for legacy models 

149 if "factorized" in data: 

150 data["data_type"] = "factorized" 

151 data["components"] = data["factorized"] 

152 del data["factorized"] 

153 data["version"] = "1.0.0" 

154 return data