-
-
Notifications
You must be signed in to change notification settings - Fork 473
/
datamodel.py
233 lines (192 loc) · 7.68 KB
/
datamodel.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
import weakref
from functools import partial
import bokeh
import bokeh.core.properties as bp
import param as pm
from bokeh.model import DataModel
from bokeh.models import ColumnDataSource
from ..reactive import Syncable
from .document import unlocked
from .notebook import push
from .state import state
class Parameterized(bokeh.core.property.bases.Property):
""" Accept a Parameterized object.
This property only exists to support type validation, e.g. for "accepts"
clauses. It is not serializable itself, and is not useful to add to
Bokeh models directly.
"""
def validate(self, value, detail=True):
super().validate(value, detail)
if isinstance(value, pm.Parameterized):
return
msg = "" if not detail else f"expected param.Parameterized, got {value!r}"
raise ValueError(msg)
class ParameterizedList(bokeh.core.property.bases.Property):
""" Accept a list of Parameterized objects.
This property only exists to support type validation, e.g. for "accepts"
clauses. It is not serializable itself, and is not useful to add to
Bokeh models directly.
"""
def validate(self, value, detail=True):
super().validate(value, detail)
if isinstance(value, list) and all(isinstance(v, pm.Parameterized) for v in value):
return
msg = "" if not detail else f"expected list of param.Parameterized, got {value!r}"
raise ValueError(msg)
_DATA_MODELS = weakref.WeakKeyDictionary()
# The Bokeh Color property has `_default_help` set which causes
# an error to be raise when Nullable is called on it. This converter
# overrides the Bokeh _help to set it to None and avoid the error.
# See https://github.com/holoviz/panel/issues/3058
def color_param_to_ppt(p, kwargs):
ppt = bp.Color(**kwargs)
ppt._help = None
return ppt
def list_param_to_ppt(p, kwargs):
if isinstance(p.item_type, type) and issubclass(p.item_type, pm.Parameterized):
return bp.List(bp.Instance(DataModel)), [(ParameterizedList, lambda ps: [create_linked_datamodel(p) for p in ps])]
return bp.List(bp.Any, **kwargs)
PARAM_MAPPING = {
pm.Array: lambda p, kwargs: bp.Array(bp.Any, **kwargs),
pm.Boolean: lambda p, kwargs: bp.Bool(**kwargs),
pm.CalendarDate: lambda p, kwargs: bp.Date(**kwargs),
pm.CalendarDateRange: lambda p, kwargs: bp.Tuple(bp.Date, bp.Date, **kwargs),
pm.ClassSelector: lambda p, kwargs: (
(bp.Instance(DataModel, **kwargs), [(Parameterized, create_linked_datamodel)])
if isinstance(p.class_, type) and issubclass(p.class_, pm.Parameterized) else
bp.Any(**kwargs)
),
pm.Color: color_param_to_ppt,
pm.DataFrame: lambda p, kwargs: (
bp.ColumnData(bp.Any, bp.Seq(bp.Any), **kwargs),
[(bp.PandasDataFrame, lambda x: ColumnDataSource._data_from_df(x))]
),
pm.DateRange: lambda p, kwargs: bp.Tuple(bp.Datetime, bp.Datetime, **kwargs),
pm.Date: lambda p, kwargs: bp.Datetime(**kwargs),
pm.Dict: lambda p, kwargs: bp.Dict(bp.String, bp.Any, **kwargs),
pm.Event: lambda p, kwargs: bp.Bool(**kwargs),
pm.Integer: lambda p, kwargs: bp.Int(**kwargs),
pm.List: list_param_to_ppt,
pm.Number: lambda p, kwargs: bp.Float(**kwargs),
pm.NumericTuple: lambda p, kwargs: bp.Tuple(*(bp.Float for p in range(p.length)), **kwargs),
pm.Range: lambda p, kwargs: bp.Tuple(bp.Float, bp.Float, **kwargs),
pm.String: lambda p, kwargs: bp.String(**kwargs),
pm.Tuple: lambda p, kwargs: bp.Tuple(*(bp.Any for p in range(p.length)), **kwargs),
}
def construct_data_model(parameterized, name=None, ignore=[], types={}):
"""
Dynamically creates a Bokeh DataModel class from a Parameterized
object.
Arguments
---------
parameterized: param.Parameterized
The Parameterized class or instance from which to create the
DataModel
name: str or None
Name of the dynamically created DataModel class
ignore: list(str)
List of parameters to ignore.
types: dict
A dictionary mapping from parameter name to a Parameter type,
making it possible to override the default parameter types.
Returns
-------
DataModel
"""
properties = {}
for pname in parameterized.param:
if pname in ignore:
continue
p = parameterized.param[pname]
if p.precedence and p.precedence < 0:
continue
ptype = types.get(pname, type(p))
prop = PARAM_MAPPING.get(ptype)
if isinstance(parameterized, Syncable):
pname = parameterized._rename.get(pname, pname)
if pname == 'name' or pname is None:
continue
nullable = getattr(p, 'allow_None', False)
kwargs = {'default': p.default, 'help': p.doc}
if prop is None:
bk_prop, accepts = bp.Any(**kwargs), []
else:
bkp = prop(p, {} if nullable else kwargs)
bk_prop, accepts = bkp if isinstance(bkp, tuple) else (bkp, [])
if nullable:
bk_prop = bp.Nullable(bk_prop, **kwargs)
for bkp, convert in accepts:
bk_prop = bk_prop.accepts(bkp, convert)
properties[pname] = bk_prop
name = name or parameterized.name
return type(name, (DataModel,), properties)
def create_linked_datamodel(obj, root=None):
"""
Creates a Bokeh DataModel from a Parameterized class or instance
which automatically links the parameters bi-directionally.
Arguments
---------
obj: param.Parameterized
The Parameterized class to create a linked DataModel for.
Returns
-------
DataModel instance linked to the Parameterized object.
"""
if isinstance(obj, type) and issubclass(obj, pm.Parameterized):
cls = obj
elif isinstance(obj, pm.Parameterized):
cls = type(obj)
else:
raise TypeError('Can only create DataModel for Parameterized class or instance.')
if cls in _DATA_MODELS:
model = _DATA_MODELS[cls]
else:
_DATA_MODELS[cls] = model = construct_data_model(obj)
properties = model.properties()
model = model(**{k: v for k, v in obj.param.values().items() if k in properties})
_changing = []
def cb_bokeh(attr, old, new):
if attr in _changing:
return
try:
_changing.append(attr)
obj.param.update(**{attr: new})
finally:
_changing.remove(attr)
def cb_param(*events):
update = {
event.name: event.new for event in events
if event.name not in _changing
}
try:
_changing.extend(list(update))
tags = [tag for tag in model.tags if tag.startswith('__ref:')]
if root:
ref = root.ref['id']
elif tags:
ref = tags[0].split('__ref:')[-1]
else:
ref = None
if ref and ref in state._views:
_, root_model, doc, comm = state._views[ref]
if comm or state._unblocked(doc):
with unlocked():
model.update(**update)
if comm and 'embedded' not in root_model.tags:
push(doc, comm)
else:
cb = partial(model.update, **update)
if doc.session_context:
doc.add_next_tick_callback(cb)
else:
cb()
else:
model.update(**update)
finally:
for attr in update:
_changing.remove(attr)
for p in obj.param:
if p in properties:
model.on_change(p, cb_bokeh)
obj.param.watch(cb_param, list(set(properties) & set(obj.param)))
return model