

In [ ]:

    
import pandas as pd
from progressivis.core import Scheduler, Every
from progressivis.core.utils import gather_and_run
from progressivis.table import Table
from progressivis.vis import MCScatterPlot
from progressivis.io import CSVLoader
#from progressivis.datasets import get_dataset
from progressivis.table.constant import Constant
import asyncio as aio
import threading
import os

def _quiet(x): pass

def _filter(df):
    pklon = df['pickup_longitude']
    pklat = df['pickup_latitude']
    dolon = df['dropoff_longitude']
    dolat = df['dropoff_latitude']


    return df[(pklon > -74.08) & (pklon < -73.5) & (pklat > 40.55) & (pklat < 41.00) &
                  (dolon > -74.08) & (dolon < -73.5) & (dolat > 40.55) & (dolat < 41.00)]

try:
    s = scheduler
except NameError:
    s = Scheduler()
PREFIX = '../../../nyc-taxi/'

SUFFIX = '.bz2'

URLS = [
    PREFIX+'yellow_tripdata_2015-01.csv'+SUFFIX,
    PREFIX+'yellow_tripdata_2015-02.csv'+SUFFIX,
    PREFIX+'yellow_tripdata_2015-03.csv'+SUFFIX,
    PREFIX+'yellow_tripdata_2015-04.csv'+SUFFIX,
    PREFIX+'yellow_tripdata_2015-05.csv'+SUFFIX,
    PREFIX+'yellow_tripdata_2015-06.csv'+SUFFIX,
]

FILENAMES = pd.DataFrame({'filename': URLS})
CST = Constant(Table('filenames', data=FILENAMES), scheduler=s)
CSV = CSVLoader(index_col=False, skipinitialspace=True,
                usecols=['pickup_longitude', 'pickup_latitude',
                             'dropoff_longitude', 'dropoff_latitude'],
                filter_=_filter, scheduler=s) # TODO: reimplement filter in read_csv.py

CSV.input.filenames = CST.output.table
PR = Every(scheduler=s, proc=_quiet)
PR.input.df = CSV.output.table


MULTICLASS = MCScatterPlot(scheduler=s, classes=[
    ('pickup', 'pickup_longitude', 'pickup_latitude'),
    ('dropoff', 'dropoff_longitude', 'dropoff_latitude')], approximate=True)
MULTICLASS.create_dependent_modules(CSV, 'table')

# Create the dashboard object
psboard = PsBoard(s)
# Visualisations require registration :
# 1) When widget provides the link_module() method do :
psboard.register_visualisation(Scatterplot(), MULTICLASS)
# 2) When widget is generic or for overloading the link_module() method do:
# psboard.register_visualisation(FooWidget(), bar_module, glue=glue_func)
# NB: glue_func(widget, module)-> [coroutine] must be provided
# Start application :

gather_and_run(s.start(), *psboard.coroutines)