-
Notifications
You must be signed in to change notification settings - Fork 49
/
util.py
233 lines (186 loc) · 7.54 KB
/
util.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
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
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
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
#
# Copyright (c) 2019. JetBrains s.r.o.
# Use of this source code is governed by the MIT license that can be found in the LICENSE file.
#
from collections.abc import Iterable
from datetime import datetime
from typing import Any, Tuple, Sequence, Optional, Dict
from lets_plot._type_utils import is_dict_or_dataframe, is_polars_dataframe
from lets_plot.geo_data_internals.utils import find_geo_names
from lets_plot.mapping import MappingMeta
from lets_plot.plot.core import aes
def as_boolean(val, *, default):
if val is None:
return default
return bool(val) and val != 'False'
def as_annotated_data(raw_data: Any, raw_mapping: Any) -> Tuple:
data_meta = {}
# data
data = raw_data
if is_data_pub_stream(data):
data = {}
for col_name in raw_data.col_names:
data[col_name] = []
data_meta.update({'pubsub': {'channel_id': raw_data.channel_id, 'col_names': raw_data.col_names}})
# mapping
mapping = {}
mapping_meta = []
# series annotations
series_meta = []
class VariableMeta:
def __init__(self):
self.levels = None
self.aesthetics = []
self.order = None
variables_meta: Dict[str, VariableMeta] = {}
if is_data_frame(data):
dtypes = data.dtypes.to_dict().items()
for column_name, dtype in dtypes:
if dtype.name == 'category' and dtype.ordered:
var_meta = VariableMeta()
var_meta.levels = dtype.categories.to_list()
variables_meta[column_name] = var_meta
if raw_mapping is not None:
for aesthetic, variable in raw_mapping.as_dict().items():
if aesthetic == 'name': # ignore FeatureSpec.name property
continue
if isinstance(variable, MappingMeta):
mapping[aesthetic] = variable.variable
if variable.variable in variables_meta:
var_meta = variables_meta[variable.variable]
else:
var_meta = VariableMeta()
var_meta.aesthetics.append(aesthetic)
if variable.levels is not None:
var_meta.levels = variable.levels
order = variable.parameters.get('order')
if order is not None:
var_meta.order = order
variables_meta[variable.variable] = var_meta
else:
mapping[aesthetic] = variable
for variableName, settings in variables_meta.items():
if settings.levels is not None:
# series annotations
series_meta.append({
'column': variableName,
'factor_levels': settings.levels,
'order': settings.order
})
else:
# mapping annotations
for aesthetic in settings.aesthetics:
value = raw_mapping.as_dict().get(aesthetic)
if value is not None and isinstance(value, MappingMeta):
mapping_meta.append({
'aes': aesthetic,
'annotation': value.annotation,
'parameters': value.parameters
})
data_as_dict = None
if is_dict_or_dataframe(data):
data_as_dict = data
elif is_polars_dataframe(data):
data_as_dict = data.to_dict()
if data_as_dict is not None:
for column_name, values in data_as_dict.items():
if isinstance(values, Iterable):
not_empty_series = any(True for _ in values)
if not_empty_series and all(isinstance(val, datetime) for val in values):
series_meta.append({
'column': column_name,
'type': 'datetime'
})
if len(series_meta) > 0:
data_meta.update({'series_annotations': series_meta})
if len(mapping_meta) > 0:
data_meta.update({'mapping_annotations': mapping_meta})
return data, aes(**mapping), {'data_meta': data_meta}
def is_data_pub_stream(data: Any) -> bool:
# try:
# from lets_plot.display import DataPubStream
# return isinstance(data, DataPubStream)
# except ImportError:
# return False # no pub-sub in standalone deployment
return False
def normalize_map_join(map_join):
if map_join is None:
return None
def invalid_map_join_format():
return ValueError("map_join must be a str, list[str] or pair of list[str]")
if isinstance(map_join, str): # 'foo' -> [['foo'], None]
data_names = [map_join]
map_names = None
elif isinstance(map_join, Sequence):
if all(isinstance(v, str) for v in map_join): # all items are strings
if len(map_join) == 1: # ['foo'] -> [['foo'], None]
data_names = map_join
map_names = None
elif len(map_join) == 2: # ['foo', 'bar'] -> [['foo'], ['bar']]
data_names = [map_join[0]]
map_names = [map_join[1]]
elif len(map_join) > 2: # ['foo', 'bar', 'baz'] -> error
raise ValueError("map_join of type list[str] expected to have 1 or 2 items, but was {}".format(len(map_join)))
else:
raise invalid_map_join_format()
elif all(isinstance(v, Sequence) and not isinstance(v, str) for v in map_join): # all items are lists
if len(map_join) == 1: # [['foo', 'bar']] -> [['foo', 'bar'], None]
data_names = map_join[0]
map_names = None
elif len(map_join) == 2: # [['foo', 'bar'], ['baz', 'qux']] -> [['foo', 'bar'], ['baz', 'qux']]
data_names = map_join[0]
map_names = map_join[1]
else:
raise invalid_map_join_format()
else:
raise invalid_map_join_format()
else:
raise invalid_map_join_format()
return [data_names, map_names]
def auto_join_geo_names(map_join: Any, gdf):
if map_join is None:
return None
data_names = map_join[0]
map_names = map_join[1]
if map_names is None:
map_names = find_geo_names(gdf)
if len(map_names) == 0:
raise ValueError(
"Can't deduce joining keys.\n"
"Define both data and map key columns in map_join "
"explicitly: map_join=[['data_column'], ['map_column']]."
)
if len(data_names) > len(map_names):
raise ValueError(
"Data key columns count exceeds map key columns count: {} > {}".format(len(data_names), len(map_names))
)
map_names = map_names[:len(data_names)] # use same number of key columns
return [data_names, map_names]
def is_geo_data_frame(data: Any) -> bool:
try:
from geopandas import GeoDataFrame
return isinstance(data, GeoDataFrame)
except ImportError:
return False
def get_geo_data_frame_meta(geo_data_frame) -> dict:
return {
'geodataframe': {
'geometry': geo_data_frame.geometry.name
}
}
def geo_data_frame_to_crs(gdf: 'GeoDataFrame', use_crs: Optional[str]):
if gdf.crs is None:
return gdf
return gdf.to_crs('EPSG:4326' if use_crs is None else use_crs)
def is_ndarray(data) -> bool:
try:
from numpy import ndarray
return isinstance(data, ndarray)
except ImportError:
return False
def is_data_frame(data: Any) -> bool:
try:
from pandas import DataFrame
return isinstance(data, DataFrame)
except ImportError:
return False