import pandas as pd
import xarray as xr
import netCDF4 as nc
from pathlib import Path
from datetime import datetime as dt
import geoviews as gv
import numpy as np
import param
from holoviews.operation.datashader import datashade
import panel as pn
import holoviews as hv
from holoviews import opts
hv.extension('bokeh')
pn.extension()
data_dir = Path('../COVID-19/csse_covid_19_data')
time_series_path = data_dir / 'csse_covid_19_time_series' / 'time_series_19-covid-{VAR}.csv'
daily_reports_path = data_dir / 'csse_covid_19_daily_reports' / '{DATE}.csv'
VARS = ['Confirmed', 'Deaths', 'Recovered']
data_vars = list()
dims = ['Aggrigation', 'Quantity', 'Region', 'Date']
for var in VARS:
df = pd.read_csv(time_series_path.as_posix().format(VAR=var))
data_vars.append(df.iloc[:,4:].values)
data_vars.append(data_vars[0] - data_vars[1] - data_vars[2]) # active cases
VARS.append('Active')
data_vars = np.stack(data_vars)
data_vars = np.stack((data_vars, np.diff(data_vars, axis=-1, n=1, prepend=0))) # daily changes
data_vars={'counts': (dims, np.stack(data_vars))}
coords = dict(
Country=('Region', df['Country/Region']),
Province_State=('Region', df['Province/State']),
Date=df.columns.tolist()[4:],
longitude=('Region', df['Long']),
latitude=('Region', df['Lat']),
Quantity=VARS,
Aggrigation=['Totals', 'Daily']
)
data = xr.Dataset(data_vars=data_vars, coords=coords)
data
cdata = data.groupby('Country').sum('Region')
ds = hv.Dataset(cdata, kdims=['Country', 'Date', 'Quantity', 'Aggrigation'],
vdims=['counts'])
dates = data['Date'].values.tolist()
dsel = ds.select(Aggrigation='Daily').aggregate(dimensions=['Quantity', 'Date'], function=np.sum)
curves = dsel.to(hv.Curve, 'Date', ).options(tools=['hover'])
curves.overlay(['Quantity']).options(
legend_position='top_left',
width=800,
height=400,
xrotation=60,
)
dsel = ds.select(Aggrigation='Totals').aggregate(dimensions=['Quantity', 'Date'], function=np.sum)
curves = dsel.to(hv.Curve, 'Date', ).options(tools=['hover'])
curves.overlay(['Quantity']).options(
legend_position='top_left',
width=800,
height=400,
xrotation=60,
)
def curves(countries, quantity):
dsel = ds.select(Country=countries, Quantity=quantity, Aggrigation='Totals')
curves = dsel.to(hv.Curve, 'Date', ).options(tools=['hover'])
return curves.overlay(['Country', 'Quantity']).options(
legend_position='top_left',
width=800,
height=400,
xrotation=60,
)
curves(['US', 'China'], 'Active')
dsel = ds.select(Aggrigation='Daily', Quantity='Active')
dsel = dsel.aggregate(dimensions=['Date'], function=np.sum)
def bars(quantity):
dsel = ds.select(Aggrigation='Daily', Quantity=quantity).aggregate(dimensions=['Date'], function=np.sum)
bars = hv.Bars(dsel, kdims=['Date'])
return bars.opts(width=800, height=400, stacked=False, show_legend=False, xrotation=60)
bars('Active')
class CovidStateCurves(param.Parameterized):
states = param.ListSelector(default=['Mississippi', 'Louisiana'], objects=usdata['Province_State'].values)
@param.depends('states')
def curves(self):
dsel = ds.select(Province_State=self.states, Quantity='Active')
curves = dsel.to(hv.Curve, 'Date', )
# curves = hv.Curve(dsel, kdims=['Date'])
return curves.overlay(['Province_State', 'Quantity']).opts(
legend_position='top_left',
width=800,
height=400,
xrotation=60 )
@param.depends('states')
def bars(self):
dsel = ds.select(Province_State=self.states, Date=data['Date'][-1])
bars = hv.Bars(dsel, kdims=['Province_State', 'Quantity'])
return bars.opts(width=800, height=400, stacked=False, show_legend=False, xrotation=60)
def panel(self):
return pn.Row(
pn.Param(self.param.states, widgets={'states': {'height': 800}}),
pn.Column(
self.curves,
self.bars,
),
)
ccc = CovidStateCurves()
ccc.panel()
dsel = ds.select(Country=ccc.countries, Date='3/15/20')
# bars = dsel.to(hv.Bars, kdims=['Country', 'Date'])
bars = hv.Bars(dsel, kdims=['Country', 'Quantity'])
bars.opts(width=800, height=400, stacked=False, show_legend=False, xrotation=60)
ccc.bars()
tiles = gv.tile_sources.CartoDark
dates = data['Date'].values.tolist()
class CovidViewer(param.Parameterized):
date = param.ObjectSelector(default=dates[1], objects=dates)
variable = param.ObjectSelector(default='Confirmed', objects=VARS)
@param.depends('date', 'variable')
def infection_map(self):
seldata = {
'longitude': data['longitude'],
'latitude': data['latitude'],
'Confirmed': data.sel(Quantity='Confirmed', Date=self.date)['counts'],
'Deaths': data.sel(Quantity='Deaths', Date=self.date)['counts'],
'Recovered': data.sel(Quantity='Recovered', Date=self.date)['counts'],
'Active': data.sel(Quantity='Active', Date=self.date)['counts'],
'Country': data['Country'],
'State/Province': data['Province_State'],
}
# gvdata = gv.Dataset(seldata, kdims=['latitude', 'longitude'], vdims=['Variable'])
# heatmap = gvdata.to(hv.HeatMap)
# points = gvdata.to(gv.Points, ['Lat', 'Lon'])
points = gv.Points(seldata, kdims=['longitude', 'latitude'], vdims=VARS+['Country', 'State/Province'])
return tiles * points.opts(
size=gv.dim(self.variable).log()*4,
logz=True,
fill_color=self.variable,
fill_alpha=.5,
line_color='gray',
cmap='hot',
width=800, height=600, global_extent=True, tools=['hover'], show_legend=False)
@param.depends('date')
def date_label(self):
return f'## {self.date}'
def panel(self):
player = pn.widgets.Player(name='Player', start=0, end=100, value=32, loop_policy='loop')
return pn.Column(
self.param.variable,
pn.Param(self.param.date, widgets={'date': pn.widgets.DiscretePlayer}),
self.date_label,
self.infection_map)
# hv.DynamicMap(infection_map, kdims=['date', 'variable']
# ).redim.values(variable=VARS, date=data['Date'].values)
cv = CovidViewer()
cv.panel()