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

1from __future__ import annotations 

2 

3import logging 

4from dataclasses import dataclass 

5from typing import Any 

6 

7import numpy as np 

8from numpy.typing import DTypeLike 

9 

10from ..component import Component 

11from ..observation import Observation 

12from ..source import Source 

13from .component import ScarletComponentBaseData 

14from .migration import PRE_SCHEMA, MigrationRegistry, migration 

15from .source_base import ScarletSourceBaseData 

16from .utils import decode_metadata, encode_metadata 

17 

18__all__ = ["ScarletSourceData"] 

19 

20CURRENT_SCHEMA = "1.0.0" 

21SOURCE_TYPE = "source" 

22MigrationRegistry.set_current(SOURCE_TYPE, CURRENT_SCHEMA) 

23logger = logging.getLogger(__name__) 

24 

25 

26@dataclass(kw_only=True) 

27class ScarletSourceData(ScarletSourceBaseData): 

28 """Data for a scarlet source 

29 

30 Attributes 

31 ---------- 

32 components : 

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

34 metadata : 

35 Metadata associated with the source. 

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

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

38 source_type : 

39 The type of source being stored 

40 version : 

41 The schema version of the ScarletSourceData. 

42 """ 

43 

44 source_type: str = SOURCE_TYPE 

45 components: list[ScarletComponentBaseData] 

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

47 version: str = CURRENT_SCHEMA 

48 

49 def as_dict(self) -> dict: 

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

51 

52 Returns 

53 ------- 

54 result : 

55 The object encoded as a JSON compatible dict 

56 """ 

57 result: dict[str, Any] = { 

58 "source_type": self.source_type, 

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

60 "version": self.version, 

61 } 

62 if self.metadata is not None: 

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

64 return result 

65 

66 @classmethod 

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

68 """Reconstruct `ScarletSourceData` from JSON compatible 

69 dict. 

70 

71 Parameters 

72 ---------- 

73 data : 

74 Dictionary representation of the object 

75 dtype : 

76 Datatype of the resulting model. 

77 

78 Returns 

79 ------- 

80 result : 

81 The reconstructed object 

82 """ 

83 data = MigrationRegistry.migrate(SOURCE_TYPE, data) 

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

85 components = [ 

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

87 ] 

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

89 

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

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

92 

93 Parameters 

94 ---------- 

95 observation: 

96 The observation used to render the source. 

97 

98 Returns 

99 ------- 

100 source: 

101 The `Source` representation of this data. 

102 """ 

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

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

105 

106 @staticmethod 

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

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

109 

110 Parameters 

111 ---------- 

112 source: 

113 The scarlet `Source` to convert. 

114 

115 Returns 

116 ------- 

117 result: 

118 The `ScarletSourceData` representation of the source. 

119 """ 

120 logger.warning("from_source is deprecated and will be removed in a future release.") 

121 return source.to_data() 

122 

123 

124ScarletSourceData.register() 

125 

126 

127@migration(SOURCE_TYPE, PRE_SCHEMA) 

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

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

130 

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

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

133 

134 Parameters 

135 ---------- 

136 data : 

137 The data to migrate. 

138 Returns 

139 ------- 

140 result : 

141 The migrated data. 

142 """ 

143 # Check for legacy models 

144 if "factorized" in data: 

145 data["data_type"] = "factorized" 

146 data["components"] = data["factorized"] 

147 del data["factorized"] 

148 data["version"] = "1.0.0" 

149 return data