Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def __init__(self,**kwargs):
# Collapse all colormap aliases into a single string, and then sort by it
inverse = {}
inv_rev = {}
for k,v in cc.cm.items():
if not k[-2:] == "_r":
inverse[v] = inverse.get(v, [])
inverse[v].insert(0,k)
else:
inv_rev[v] = inverse.get(v, [])
inv_rev[v].insert(0,k)
all_cms = {', '.join(reversed(v)):k for (k,v) in inverse.items()}.items()
rev_cms = {', '.join(reversed(v)):k for (k,v) in inv_rev.items()}.items()
self._all_cmaps = dict([i for i in all_cms])
self._rev_cmaps = dict([i for i in rev_cms])
self._div_cms = dict([i for i in all_cms if "diverging" in i[0]])
self._iso_cms = dict([i for i in all_cms if "isoluminant" in i[0]])
self._linear_cms = dict([i for i in all_cms if "linear" in i[0]])
self._cycle_cms = dict([i for i in all_cms if "cyclic" in i[0]])
self._alias_cms = dict([(x.split(',')[1].lstrip(' '),v) for x,v in self._all_cmaps.items() if ',' in x])
control_box = ['c$N=seaborn_palette',
def update(self,val):
if not isinstance(val['new'],(bool,int)):
self._curr_cmap = val['new'] if ',' not in val['new'] else val['new'].split(',')[0]
self._cmap_name = self._curr_cmap
if self.reversed.value:
self._cmap_name = self._curr_cmap + '_r'
elif self._cmap_name[-2:]=='_r':
self._cmap_name = self._curr_cmap
self.cmap = cc.cm[self._cmap_name]
x = np.linspace(0,1,num=self.n_colors.value)
self.pal[:] = ListedColormap(self.cmap(x))(x)
self.update_fig_widget()
from bokeh.models import WMTSTileSource
url = 'https://server.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{Z}/{Y}/{X}.jpg'
wmts = gv.WMTS(WMTSTileSource(url=url))
stream = hv.streams.Stream.define('HourSelect', hour=0)()
points = hv.Points(ddf, kdims=['dropoff_x', 'dropoff_y'])
dmap = hv.util.Dynamic(points, operation=lambda obj, hour: obj.select(dropoff_hour=hour).relabel('Hour of Day: %d' % hour),
streams=[stream])
# Apply aggregation
aggregated = aggregate(dmap, link_inputs=True, streams=[hv.streams.RangeXY], width=1200, height=600)
# Shade the data
class ColormapPicker(hv.streams.Stream):
colormap = param.ObjectSelector(default=cm["fire"],
objects=[cm[k] for k in cm.keys() if not '_' in k])
cmap_picker = ColormapPicker(rename={'colormap': 'cmap'}, name='')
shaded = shade(aggregated, link_inputs=True, streams=[cmap_picker])
# Define PointerX stream, attach to points and declare DynamicMap for cross-section and VLine
pointer = hv.streams.PointerX(x=ddf.dropoff_x.loc[0].compute().iloc[0], source=points)
section = hv.util.Dynamic(aggregated, operation=lambda obj, x: obj.sample(dropoff_x=x),
streams=[pointer], link_inputs=False).relabel('')
vline = hv.DynamicMap(lambda x: hv.VLine(x), streams=[pointer])
# Define options
hv.opts("RGB [width=1200 height=600 xaxis=None yaxis=None fontsize={'title': '14pt'}] VLine (color='white' line_width=2)")
hv.opts("Curve [width=150 yaxis=None show_frame=False] (color='black') {+framewise} Layout [shared_axes=False]")
# Combine it all into a complex layout
hvobj = (wmts * shaded * vline) << section
from holoviews import opts
from holoviews.operation.datashader import rasterize, shade
from holoviews.element.tiles import EsriImagery
from colorcet import cm
hv.extension('bokeh', logo=False)
df = dd.read_parquet('../../data/osm-1billion.snappy.parq', engine='fastparquet').persist()
cmaps = ['fire','bgy','bgyw','bmy','gray','kbc']
topts = opts.Tiles(width=900, height=600, xaxis=None, yaxis=None, bgcolor='black', show_grid=False)
tiles = EsriImagery().opts(topts)
class OSM(param.Parameterized):
alpha = param.Magnitude(default=0.75, doc="Map tile opacity")
cmap = param.ObjectSelector(cm['fire'], objects={c:cm[c] for c in cmaps})
@param.depends('alpha')
def tiles(self):
return tiles.opts(alpha=self.alpha)
@param.depends()
def view(self):
points = hv.Points(df, kdims=['x', 'y'])
raster = rasterize(points, x_sampling=1, y_sampling=1, width=900, height=600)
return hv.DynamicMap(self.tiles) * shade(raster, streams=[hv.streams.Params(self, ['cmap'])])
osm = OSM(name="Open Street Map GPS")
pn.Row(osm.param, osm.view).servable()
from colorcet import cm
from holoviews.operation.datashader import rasterize, shade
from holoviews.element.tiles import StamenTerrain
hv.extension('bokeh', logo=False)
usecols = ['dropoff_x','dropoff_y','pickup_x','pickup_y','dropoff_hour','pickup_hour','passenger_count']
df = dd.read_parquet('../../data/nyc_taxi_wide.parq', engine='fastparquet')[usecols].persist()
opts = dict(width=1000,height=600,xaxis=None,yaxis=None,bgcolor='black',show_grid=False)
cmaps = ['fire','bgy','bgyw','bmy','gray','kbc']
tiles = StamenTerrain()
class NYCTaxiExplorer(param.Parameterized):
alpha = param.Magnitude(default=0.75, doc="Alpha value for the map opacity")
cmap = param.ObjectSelector(cm['fire'], objects={c:cm[c] for c in cmaps})
hour = param.Range(default=(0, 24), bounds=(0, 24))
location = param.ObjectSelector(default='dropoff', objects=['dropoff', 'pickup'])
@param.depends('location', 'hour')
def points(self):
points = hv.Points(df, kdims=[self.location+'_x', self.location+'_y'], vdims=['dropoff_hour'])
if self.hour != (0, 24): points = points.select(dropoff_hour=self.hour)
return points
@param.depends('alpha')
def tiles(self):
return tiles.options(alpha=self.alpha, **opts)
def view(self,**kwargs):
points = hv.DynamicMap(self.points)
agg = rasterize(points, x_sampling=1, y_sampling=1, width=600, height=400)
# TODO: make this more general - not reliant on '_r' versions of cmaps
if reverse:
if name[-2:] != '_r':
name = name + '_r'
# 'un'-reverse
else:
name = name[:-2]
# Custom xyzpy colormaps
if name in _XYZ_CMAPS:
return _XYZ_CMAPS[name]
# special cases with name conflicts --> prefer matplotlib
if name not in {'inferno', 'coolwarm', 'blues'}:
try:
return colorcet.cm[name]
except (NameError, KeyError):
pass
try:
return getattr(cmocean.cm, name)
except (NameError, AttributeError):
pass
# matplotlib colormaps
return getattr(cm, name)
from colorcet import cm
from holoviews.operation.datashader import rasterize, shade
from holoviews.streams import RangeXY
hv.extension('bokeh', logo=False)
usecols = ['dropoff_x','dropoff_y','pickup_x','pickup_y','dropoff_hour','pickup_hour','passenger_count']
df = dd.read_parquet('../data/nyc_taxi_wide.parq')[usecols].persist()
opts = dict(width=1000,height=600,xaxis=None,yaxis=None,bgcolor='black',show_grid=False)
cmaps = ['fire','bgy','bgyw','bmy','gray','kbc']
class NYCTaxiExplorer(param.Parameterized):
alpha = param.Magnitude(default=0.75, doc="Alpha value for the map opacity")
cmap = param.ObjectSelector(cm['fire'], objects={c:cm[c] for c in cmaps})
hour = param.Range(default=(0, 24), bounds=(0, 24))
location = param.ObjectSelector(default='dropoff', objects=['dropoff', 'pickup'])
@param.depends('location', 'hour')
def points(self):
points = hv.Points(df, kdims=[self.location+'_x', self.location+'_y'], vdims=['dropoff_hour'])
if self.hour != (0, 24): points = points.select(dropoff_hour=self.hour)
return points
@param.depends('alpha')
def tiles(self):
return gv.tile_sources.StamenTerrain.options(alpha=self.alpha, **opts)
def view(self):
points = hv.DynamicMap(self.points)
agg = rasterize(points, x_sampling=1, y_sampling=1, width=600, height=400)
def get_cmap_safe(cmap):
"""Fetches a colormap by name from matplotlib, colorcet, or cmocean"""
try:
from matplotlib.cm import get_cmap
except ImportError:
raise Exception('cmap requires matplotlib')
if isinstance(cmap, str):
# Try colorcet first
try:
import colorcet
cmap = colorcet.cm[cmap]
except (ImportError, KeyError):
pass
else:
return cmap
# Try cmocean second
try:
import cmocean
cmap = getattr(cmocean.cm, cmap)
except (ImportError, AttributeError):
pass
else:
return cmap
# Else use Matplotlib
cmap = get_cmap(cmap)
elif isinstance(cmap, list):
for item in cmap:
import holoviews as hv, geoviews as gv, param, dask.dataframe as dd, panel as pp
from holoviews.operation.datashader import datashade, rasterize, shade
from holoviews.streams import RangeXY
from colorcet import cm
hv.extension('bokeh', logo=False)
df = dd.read_parquet('../data/osm-1billion.snappy.parq').persist()
cmaps = ['fire','bgy','bgyw','bmy','gray','kbc']
topts = dict(width=900,height=600,xaxis=None,yaxis=None,bgcolor='black',show_grid=False)
class OSM(param.Parameterized):
alpha = param.Magnitude(default=0.75, doc="Map tile opacity")
cmap = param.ObjectSelector(cm['fire'], objects={c:cm[c] for c in cmaps})
@param.depends('alpha')
def tiles(self):
return gv.tile_sources.EsriImagery.options(alpha=self.alpha, **topts)
@param.depends()
def view(self):
points = hv.DynamicMap(hv.Points(df, kdims=['x', 'y']))
raster = rasterize(points, x_sampling=1, y_sampling=1, width=900, height=600)
return hv.DynamicMap(self.tiles) * shade(raster, streams=[hv.streams.Params(self, ['cmap'])])
osm = OSM(name="Open Street Map GPS")
panel = pp.Row(osm, osm.view)
panel.server_doc()
from bokeh.models import WMTSTileSource
url = 'https://server.arcgisonline.com/ArcGIS/rest/services/World_Imagery/MapServer/tile/{Z}/{Y}/{X}.jpg'
wmts = gv.WMTS(WMTSTileSource(url=url))
stream = hv.streams.Stream.define('HourSelect', hour=0)()
points = hv.Points(ddf, kdims=['dropoff_x', 'dropoff_y'])
dmap = hv.util.Dynamic(points, operation=lambda obj, hour: obj.select(dropoff_hour=hour).relabel('Hour of Day: %d' % hour),
streams=[stream])
# Apply aggregation
aggregated = aggregate(dmap, link_inputs=True, streams=[hv.streams.RangeXY], width=1200, height=600)
# Shade the data
class ColormapPicker(hv.streams.Stream):
colormap = param.ObjectSelector(default=cm["fire"],
objects=[cm[k] for k in cm.keys() if not '_' in k])
cmap_picker = ColormapPicker(rename={'colormap': 'cmap'}, name='')
shaded = shade(aggregated, link_inputs=True, streams=[cmap_picker])
# Define PointerX stream, attach to points and declare DynamicMap for cross-section and VLine
pointer = hv.streams.PointerX(x=ddf.dropoff_x.loc[0].compute().iloc[0], source=points)
section = hv.util.Dynamic(aggregated, operation=lambda obj, x: obj.sample(dropoff_x=x),
streams=[pointer], link_inputs=False).relabel('')
vline = hv.DynamicMap(lambda x: hv.VLine(x), streams=[pointer])
# Define options
hv.opts("RGB [width=1200 height=600 xaxis=None yaxis=None fontsize={'title': '14pt'}] VLine (color='white' line_width=2)")
hv.opts("Curve [width=150 yaxis=None show_frame=False] (color='black') {+framewise} Layout [shared_axes=False]")
# Combine it all into a complex layout