|
2 | 2 |
|
3 | 3 |
|
4 | 4 | # standard library
|
5 |
| -from dataclasses import Field, dataclass, field, is_dataclass |
6 |
| -from typing import Any, Dict, Hashable, Optional, Type, Union, cast |
| 5 | +from dataclasses import dataclass, field, is_dataclass |
| 6 | +from typing import Any, Dict, Hashable, List, Optional, Tuple, Type, Union, cast |
7 | 7 |
|
8 | 8 |
|
9 | 9 | # dependencies
|
10 | 10 | import numpy as np
|
11 | 11 | import xarray as xr
|
12 |
| -from typing_extensions import ParamSpec, TypedDict, get_args, get_type_hints |
| 12 | +from typing_extensions import Literal, ParamSpec, get_type_hints |
13 | 13 |
|
14 | 14 |
|
15 | 15 | # submodules
|
16 |
| -from .typing import ( |
17 |
| - ArrayLike, |
18 |
| - DataClass, |
19 |
| - DataType, |
20 |
| - Dims, |
21 |
| - Dtype, |
22 |
| - FieldType, |
23 |
| - get_dims, |
24 |
| - get_dtype, |
25 |
| - get_inner, |
26 |
| - unannotate, |
27 |
| -) |
| 16 | +from .typing import ArrayLike, DataClass, DataType, Dims, Dtype |
28 | 17 |
|
29 | 18 |
|
30 | 19 | # type hints
|
31 | 20 | P = ParamSpec("P")
|
32 | 21 | AnyDataClass = Union[Type[DataClass[P]], DataClass[P]]
|
33 |
| -DimsDtype = TypedDict("DimsDtype", dims=Dims, dtype=Dtype) |
| 22 | +AnyEntry = Union["AttrEntry", "DataEntry"] |
34 | 23 |
|
35 | 24 |
|
36 |
| -# field models |
37 |
| -@dataclass(frozen=True) |
38 |
| -class Data: |
39 |
| - """Field model for data-related fields.""" |
| 25 | +# constants |
| 26 | +class MissingType: |
| 27 | + """Singleton that indicates missing data.""" |
40 | 28 |
|
41 |
| - name: Hashable |
42 |
| - """Name of the field.""" |
| 29 | + _instance = None |
43 | 30 |
|
44 |
| - value: Any |
45 |
| - """Value assigned to the field.""" |
| 31 | + def __new__(cls) -> "MissingType": |
| 32 | + if cls._instance is None: |
| 33 | + cls._instance = super().__new__(cls) |
46 | 34 |
|
47 |
| - type: DimsDtype |
48 |
| - """Type (dims and dtype) of the field.""" |
| 35 | + return cls._instance |
49 | 36 |
|
50 |
| - factory: Any = None |
51 |
| - """Factory dataclass to create a DataArray object.""" |
| 37 | + def __repr__(self) -> str: |
| 38 | + return "<MISSING>" |
52 | 39 |
|
53 |
| - def __call__(self, reference: Optional[DataType] = None) -> xr.DataArray: |
54 |
| - """Create a DataArray object from the value and a reference.""" |
55 |
| - from .dataarray import asdataarray |
56 |
| - |
57 |
| - if self.factory is None: |
58 |
| - return typedarray( |
59 |
| - self.value, |
60 |
| - self.type["dims"], |
61 |
| - self.type["dtype"], |
62 |
| - reference, |
63 |
| - ) |
64 |
| - |
65 |
| - if is_dataclass(self.value): |
66 |
| - return asdataarray(self.value, reference) |
67 |
| - else: |
68 |
| - return asdataarray(self.factory(self.value), reference) |
69 | 40 |
|
70 |
| - @classmethod |
71 |
| - def from_field(cls, field: Field[Any], value: Any, of: bool) -> "Data": |
72 |
| - """Create a field model from a dataclass field and a value.""" |
73 |
| - hint = unannotate(field.type) |
| 41 | +MISSING = MissingType() |
| 42 | + |
| 43 | + |
| 44 | +# runtime classes |
| 45 | +@dataclass(frozen=True) |
| 46 | +class AttrEntry: |
| 47 | + """Entry of an attribute (i.e. metadata).""" |
| 48 | + |
| 49 | + name: Hashable |
| 50 | + """Name that the attribute is accessed by.""" |
74 | 51 |
|
75 |
| - if not of: |
76 |
| - type: DimsDtype = { |
77 |
| - "dims": get_dims(get_args(hint)[0]), |
78 |
| - "dtype": get_dtype(get_args(hint)[0]), |
79 |
| - } |
80 |
| - return cls(field.name, value, type) |
| 52 | + tag: Literal["attr", "name"] |
| 53 | + """Function of the attribute (either attr or name).""" |
81 | 54 |
|
82 |
| - dataclass = get_inner(hint, 0) |
83 |
| - model = DataModel.from_dataclass(dataclass) |
84 |
| - data_item = next(iter(model.data.values())) |
| 55 | + type: Any = None |
| 56 | + """Type or type hint of the attribute.""" |
85 | 57 |
|
86 |
| - if not model.name: |
87 |
| - return cls(field.name, value, data_item.type, dataclass) |
88 |
| - else: |
89 |
| - name_item = next(iter(model.name.values())) |
90 |
| - return cls(name_item.value, value, data_item.type, dataclass) |
| 58 | + value: Any = MISSING |
| 59 | + """Actual value of the attribute.""" |
| 60 | + |
| 61 | + cast: bool = False |
| 62 | + """Whether the value is cast to the type.""" |
| 63 | + |
| 64 | + def __call__(self) -> Any: |
| 65 | + """Create an object according to the entry.""" |
| 66 | + ... |
91 | 67 |
|
92 | 68 |
|
93 | 69 | @dataclass(frozen=True)
|
94 |
| -class General: |
95 |
| - """Field model for general fields.""" |
| 70 | +class DataEntry: |
| 71 | + """Entry of a data variable.""" |
96 | 72 |
|
97 | 73 | name: Hashable
|
98 |
| - """Name of the field.""" |
| 74 | + """Name that the attribute is accessed by.""" |
99 | 75 |
|
100 |
| - value: Any |
101 |
| - """Value assigned to the field.""" |
| 76 | + tag: Literal["coord", "data"] |
| 77 | + """Function of the data (either coord or data).""" |
102 | 78 |
|
103 |
| - type: str |
104 |
| - """Type of the field.""" |
| 79 | + dims: Dims = cast(Dims, None) |
| 80 | + """Dimensions of the DataArray that the data is cast to.""" |
105 | 81 |
|
106 |
| - factory: Optional[Type[Any]] = None |
107 |
| - """Factory function to create an object.""" |
| 82 | + dtype: Dtype = cast(Dtype, None) |
| 83 | + """Data type of the DataArray that the data is cast to.""" |
108 | 84 |
|
109 |
| - def __call__(self) -> Any: |
110 |
| - """Create an object from the value.""" |
111 |
| - if self.factory is None: |
112 |
| - return self.value |
113 |
| - else: |
114 |
| - return self.factory(self.value) |
| 85 | + base: Optional[Type[Any]] = None |
| 86 | + """Base dataclass that converts the data to a DataArray.""" |
115 | 87 |
|
116 |
| - @classmethod |
117 |
| - def from_field(cls, field: Field[Any], value: Any) -> "General": |
118 |
| - """Create a field model from a dataclass field and a value.""" |
119 |
| - hint = unannotate(field.type) |
| 88 | + value: Any = MISSING |
| 89 | + """Actual value of the data.""" |
120 | 90 |
|
121 |
| - try: |
122 |
| - return cls(field.name, value, f"{hint.__module__}.{hint.__qualname__}") |
123 |
| - except AttributeError: |
124 |
| - return cls(field.name, value, repr(hint)) |
| 91 | + cast: bool = True |
| 92 | + """Whether the value is cast to the data type.""" |
| 93 | + |
| 94 | + def __call__(self, reference: Optional[DataType] = None) -> xr.DataArray: |
| 95 | + """Create a DataArray object according to the entry.""" |
| 96 | + ... |
125 | 97 |
|
126 | 98 |
|
127 |
| -# data models |
128 | 99 | @dataclass(frozen=True)
|
129 | 100 | class DataModel:
|
130 |
| - """Model for dataclasses or their objects.""" |
| 101 | + """Data representation (data model) inside the package.""" |
| 102 | + |
| 103 | + entries: Dict[str, AnyEntry] = field(default_factory=dict) |
| 104 | + """Entries of data variable(s) and attribute(s).""" |
| 105 | + |
| 106 | + @property |
| 107 | + def attrs(self) -> List[AttrEntry]: |
| 108 | + """Return a list of attribute entries.""" |
| 109 | + ... |
131 | 110 |
|
132 |
| - attr: Dict[str, General] = field(default_factory=dict) |
133 |
| - """Model of the attribute fields.""" |
| 111 | + @property |
| 112 | + def coords(self) -> List[DataEntry]: |
| 113 | + """Return a list of coordinate entries.""" |
| 114 | + ... |
134 | 115 |
|
135 |
| - coord: Dict[str, Data] = field(default_factory=dict) |
136 |
| - """Model of the coordinate fields.""" |
| 116 | + @property |
| 117 | + def data_vars(self) -> List[DataEntry]: |
| 118 | + """Return a list of data variable entries.""" |
| 119 | + ... |
137 | 120 |
|
138 |
| - data: Dict[str, Data] = field(default_factory=dict) |
139 |
| - """Model of the data fields.""" |
| 121 | + @property |
| 122 | + def data_vars_items(self) -> List[Tuple[str, DataEntry]]: |
| 123 | + """Return a list of data variable entries with keys.""" |
| 124 | + ... |
140 | 125 |
|
141 |
| - name: Dict[str, General] = field(default_factory=dict) |
142 |
| - """Model of the name fields.""" |
| 126 | + @property |
| 127 | + def names(self) -> List[AttrEntry]: |
| 128 | + """Return a list of name entries.""" |
| 129 | + ... |
143 | 130 |
|
144 | 131 | @classmethod
|
145 | 132 | def from_dataclass(cls, dataclass: AnyDataClass[P]) -> "DataModel":
|
146 | 133 | """Create a data model from a dataclass or its object."""
|
147 |
| - model = cls() |
148 |
| - eval_dataclass(dataclass) |
149 |
| - |
150 |
| - for field in dataclass.__dataclass_fields__.values(): |
151 |
| - value = getattr(dataclass, field.name, field.default) |
152 |
| - |
153 |
| - if FieldType.ATTR.annotates(field.type): |
154 |
| - model.attr[field.name] = General.from_field(field, value) |
155 |
| - elif FieldType.COORD.annotates(field.type): |
156 |
| - model.coord[field.name] = Data.from_field(field, value, False) |
157 |
| - elif FieldType.COORDOF.annotates(field.type): |
158 |
| - model.coord[field.name] = Data.from_field(field, value, True) |
159 |
| - elif FieldType.DATA.annotates(field.type): |
160 |
| - model.data[field.name] = Data.from_field(field, value, False) |
161 |
| - elif FieldType.DATAOF.annotates(field.type): |
162 |
| - model.data[field.name] = Data.from_field(field, value, True) |
163 |
| - elif FieldType.NAME.annotates(field.type): |
164 |
| - model.name[field.name] = General.from_field(field, value) |
165 |
| - |
166 |
| - return model |
| 134 | + ... |
167 | 135 |
|
168 | 136 |
|
169 | 137 | # runtime functions
|
|
0 commit comments