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
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193 | @dataclass
class Record:
id: str
name: str
datetime: datetime
tags: Dict[str, str]
aoi_id: str
_aoi: geometry.MultiPolygon = geometry.MultiPolygon()
@classmethod
def from_pb(cls, pb: records_pb2.Record):
return cls(
id=pb.id,
name=pb.name,
tags={key: value for key, value in pb.tags.items()},
datetime=pb.time.ToDatetime(),
aoi_id=pb.aoi_id,
_aoi=aoi_from_pb(pb.aoi)
)
@classmethod
def from_geodataframe(cls, gdf: gpd.GeoDataFrame):
return cls(
id=gdf["id"].iloc[0],
name=gdf["name"].iloc[0],
tags={k: v for k, v in gdf["tags"].iloc[0]},
datetime=gdf["datetime"].iloc[0],
aoi_id=gdf["aoi_id"].iloc[0],
_aoi=gdf.geometry.iloc[0]
)
def to_pb(self) -> records_pb2.Record:
pb = records_pb2.Record(
id=self.id,
name=self.name,
tags={key: value for key, value in self.tags.items()},
aoi_id=self.aoi_id
)
pb.time.FromDatetime(self.datetime)
return pb
def geodataframe(self) -> gpd.GeoDataFrame:
self._check_aoi()
return gpd.GeoDataFrame(
{
"id": [self.id],
"name": [self.name],
"tags": [[(k, v) for k, v in self.tags.items()]],
"datetime": [str(self.datetime)],
"aoi_id": [self.aoi_id],
"geometry": [self.aoi]
},
crs='epsg:4326'
)
@property
@utils.catch_rpc_error
def aoi(self) -> geometry.MultiPolygon:
self._check_aoi()
return self._aoi
@aoi.setter
def aoi(self, aoi: geometry.MultiPolygon):
self._aoi = aoi
def __repr__(self):
return "Record {} ({})".format(self.name, self.id)
def __str__(self):
return "Record {} ({})\n" \
" datetime {}\n" \
" tags \n{}\n"\
" aoi_id {}\n".format(self.name, self.id, self.datetime,
pprint.pformat(self.tags, indent=6, width=1), self.aoi_id) + \
" aoi {}\n".format(self.aoi if not self._aoi.is_empty else "(not loaded)")
@staticmethod
def key_date(r: Record):
""" Returns the date of the record (without time).
It's a GroupByKeyFunc, thus it can be used to group_by."""
return r.datetime.date()
@staticmethod
def key_datetime(r: Record):
""" Returns the datetime of the record.
It's a GroupByKeyFunc, thus it can be used to group_by."""
return r.datetime
@staticmethod
def group_by(records: List[Union[Record, GroupedRecords]], func_key: GroupByKeyFunc) -> List[GroupedRecords]:
"""
group_by groups the records of the list by the key provided by the func_key(Record)
Returns a list of lists of records
Args:
records : list of records to group.
If records is a list of list, records is flattened.
func_key : function taking a record and returning a key
(e.g. entities.Record.key_date, lambda r:r.datetime)
Returns:
A list of grouped records (which is actually a list of records)
"""
if len(records) == 0:
return records
if isinstance(records[0], list):
records = [r for rs in records for r in rs]
return Record.group_by(records, func_key)
dict_rs = {}
for r in records:
k = func_key(r)
if k not in dict_rs:
dict_rs[k] = [r]
else:
dict_rs[k].append(r)
return list(dict_rs.values())
@staticmethod
def list_to_geodataframe(records: List[Record]) -> gpd.GeoDataFrame:
return gpd.GeoDataFrame(
{
"id": [r.id for r in records],
"name": [r.name for r in records],
"tags": [[(k, v) for k, v in r.tags.items()] for r in records],
"datetime": [str(r.datetime) for r in records],
"aoi_id": [r.aoi_id for r in records],
"geometry": [r.aoi for r in records],
},
crs='epsg:4326'
)
def _check_aoi(self, raise_if_empty=True) -> bool:
if self._aoi.is_empty:
if raise_if_empty:
raise ReferenceError("AOI is not loaded. Call 'client.load_aoi(record)'")
return False
return True
|