Coverage for python / lsst / scarlet / lite / io / source_base.py: 69%

1from __future__ import annotations 

2 

3from abc import ABC, abstractmethod 

4from dataclasses import dataclass 

5from typing import Any, ClassVar 

6 

7import numpy as np 

8from numpy.typing import DTypeLike 

9 

10from ..observation import Observation 

11from ..source import Source 

12from .utils import PersistenceError 

13 

14__all__ = ["ScarletSourceBaseData"] 

15 

16 

17@dataclass(kw_only=True) 

18class ScarletSourceBaseData(ABC): 

19 """Data for a scarlet source 

20 

21 Attributes 

22 ---------- 

23 source_type : 

24 The type of source being stored. 

25 source_registry : 

26 A registry of all known source types used for deserialization. 

27 metadata : 

28 Metadata associated with the source. 

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

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

31 version : 

32 The schema version of the exact data class. 

33 """ 

34 

35 source_type: str 

36 source_registry: ClassVar[dict[str, type[ScarletSourceBaseData]]] = {} 

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

38 version: str 

39 

40 @classmethod 

41 def register(cls) -> None: 

42 """Register a new source type""" 

43 ScarletSourceBaseData.source_registry[cls.source_type] = cls 

44 

45 @abstractmethod 

46 def as_dict(self) -> dict: 

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

48 

49 Returns 

50 ------- 

51 result : 

52 The object encoded as a JSON compatible dict 

53 """ 

54 

55 @staticmethod 

56 def from_dict(data: dict, dtype: DTypeLike = np.float32) -> ScarletSourceBaseData: 

57 """Reconstruct `ScarletSourceData` from JSON compatible 

58 dict. 

59 

60 Parameters 

61 ---------- 

62 data : 

63 Dictionary representation of the object 

64 dtype : 

65 Datatype of the resulting model. 

66 

67 Returns 

68 ------- 

69 result : 

70 The reconstructed object 

71 """ 

72 source_type = data.get("source_type", None) 

73 

74 # Fix legacy data that did not have a source_type 

75 if source_type is None: 

76 source_type = "source" 

77 

78 cls = ScarletSourceBaseData.source_registry.get(source_type, None) 

79 if cls is None: 

80 raise PersistenceError(f"Unknown source type: {source_type}") 

81 

82 return cls.from_dict(data, dtype=dtype) 

83 

84 @abstractmethod 

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

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

87 

88 Parameters 

89 ---------- 

90 observation: 

91 The observation used to render the source. 

92 

93 Returns 

94 ------- 

95 source: 

96 The `Source` representation of this data. 

97 """