-
-
Notifications
You must be signed in to change notification settings - Fork 99
/
fugue.py
62 lines (50 loc) · 1.84 KB
/
fugue.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
"""
Experimental support for fugue.
"""
from typing import Any, Dict, Tuple
import panel as _pn
from . import hvPlotTabular, post_patch
from .util import _fugue_ipython
def patch(name="hvplot", extension="bokeh", logo=False):
try:
from fugue import DataFrames, Outputter
from fugue.extensions import namespace_candidate, parse_outputter
except:
raise ImportError(
'Could not add fugue support as it could not be imported. '
'Please make sure you have installed fugue in your environment.'
)
import hvplot.pandas # noqa: F401
class _Visualize(Outputter):
def __init__(self, func: str) -> None:
super().__init__()
self._func = func
getattr(hvPlotTabular, func) # ensure the func exists
def process(self, dfs: DataFrames) -> None:
"""
Process the dataframes and output the result as
a pn.Column.
Parameters:
-----------
dfs: fugue.DataFrames
"""
charts = []
for df in dfs.values():
params = dict(self.params)
opts: Dict[str, Any] = params.pop("opts", {})
chart = getattr(df.as_pandas().hvplot, self._func)(**params).opts(**opts)
charts.append(chart)
col = _pn.Column(*charts)
try:
if not _fugue_ipython:
get_ipython()
except NameError:
col.show() # in script
else:
from IPython.display import display
display(col) # in notebook
@parse_outputter.candidate(namespace_candidate(name, lambda x: isinstance(x, str)))
def _parse_hvplot(obj: Tuple[str, str]) -> Outputter:
return _Visualize(obj[1])
post_patch(extension, logo)
patch()