15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131 | @dataclass
class CubeParams:
_instance_id: str
_records_id: Union[List[entities.GroupedRecordIds], None]
tile: entities.Tile
tags: Union[Dict[str, str], None]
from_time: Union[datetime, None]
to_time: Union[datetime, None]
@classmethod
def from_tags(cls, crs: str, transform: Union[affine.Affine, Tuple6Float],
shape: Tuple[int, int],
instance: Union[str, entities.VariableInstance, None],
tags: Dict[str, str], from_time: datetime = None, to_time: datetime = None) -> entities.CubeParams:
"""
Create a set of parameters to get a cube from record tags
Args:
crs: of the output images (images will be reprojected on the fly if necessary)
transform: of the requested cube (images will be rescaled on the fly if necessary)
shape: of the requested cube (@warning shape is the transpose of numpy shape)
instance: of the requested data
tags: of the records to be requested
from_time: (optional) to filter the records
to_time: (optional) to filter the records
Returns:
A CubeParams to be passed as a parameter of a get_cube request
"""
return cls.from_tile(entities.Tile.from_geotransform(transform, crs, shape), instance,
tags=tags, from_time=from_time, to_time=to_time)
@classmethod
def from_records(cls, records: List[entities.RecordIdentifiers],
crs: str, transform: Union[affine.Affine, Tuple6Float],
shape: Tuple[int, int],
instance: Union[str, entities.VariableInstance, None]) -> entities.CubeParams:
"""
Create a set of parameters to get a cube from a list of records
Args:
crs: of the output images (images will be reprojected on the fly if necessary)
transform: of the requested cube (images will be rescaled on the fly if necessary)
shape: of the requested cube (@warning shape is the transpose of numpy shape)
instance: of the requested data
records: to be retrieved
Returns:
A CubeParams to be passed as a parameter of a get_cube request
"""
return cls.from_tile(entities.Tile.from_geotransform(transform, crs, shape), instance, records=records)
@classmethod
def from_tile(cls, tile: entities.Tile,
instance: Union[str, entities.VariableInstance, None],
records: List[entities.RecordIdentifiers] = None,
tags: Dict[str, str] = None, from_time: datetime = None, to_time: datetime = None) \
-> entities.CubeParams:
"""
Args:
tile: defining the Cube to be retrieved (images will be reprojected on the fly if necessary)
instance: of the requested data
records: (optional) to be retrieved
tags: of the records to be requested
from_time: (optional) to filter the records
to_time: (optional) to filter the records
Returns:
A CubeParams to be passed as a parameter of a get_cube request
"""
return cls(tile=tile,
_instance_id=entities.get_id(instance) if instance else None,
_records_id=CubeParams._parse_grouped_record_ids(records),
tags=tags, from_time=from_time, to_time=to_time)
@property
def crs(self) -> str:
return self.tile.crs
@property
def transform(self) -> affine.Affine:
return self.tile.transform
@property
def shape(self) -> Tuple[int, int]:
return self.tile.shape
@property
def records(self) -> List[entities.GroupedRecordIds]:
return self._records_id
@records.setter
def records(self, records: List[entities.RecordIdentifiers]):
self._records_id = CubeParams._parse_grouped_record_ids(records)
@property
def instance(self) -> str:
return self._instance_id
@instance.setter
def instance(self, instance: Union[str, entities.VariableInstance]):
self._instance_id = entities.get_id(instance)
@staticmethod
def _parse_grouped_record_ids(records: List[entities.RecordIdentifiers])\
-> Union[List[entities.GroupedRecordIds], None]:
return [CubeParams._parse_record_ids(rs) for rs in records] if records is not None else None
@staticmethod
def _parse_record_ids(records: entities.RecordIdentifiers) -> List[str]:
if isinstance(records, gpd.GeoDataFrame):
return list(records['id'])
return entities.get_ids(records)
|