<s> """<STR_LIT>""" <EOL> from __future__ import absolute_import <EOL> import numpy as np <EOL> import pandas as pd <EOL> from bokeh . models . sources import ColumnDataSource <EOL> from bokeh . core . properties import ( HasProps , Float , Either , String , Date , Datetime , Int , <EOL> Bool , List , Instance ) <EOL> from . properties import Column , EitherColumn , ColumnLabel <EOL> class Stat ( HasProps ) : <EOL> """<STR_LIT>""" <EOL> column = ColumnLabel ( help = """<STR_LIT>""" ) <EOL> source = Instance ( ColumnDataSource , help = """<STR_LIT>""" ) <EOL> values = EitherColumn ( Column ( Float ) , Column ( Int ) , Column ( String ) , <EOL> Column ( Date ) , Column ( Datetime ) , Column ( Bool ) , default = None , help = """<STR_LIT>""" ) <EOL> value = Float ( help = """<STR_LIT>""" ) <EOL> def __init__ ( self , ** properties ) : <EOL> source = properties . pop ( '<STR_LIT:source>' , None ) <EOL> if source is not None : <EOL> if isinstance ( source , pd . DataFrame ) : <EOL> source = ColumnDataSource ( source ) <EOL> properties [ '<STR_LIT:source>' ] = source <EOL> super ( Stat , self ) . __init__ ( ** properties ) <EOL> self . _refresh ( ) <EOL> def _refresh ( self ) : <EOL> """<STR_LIT>""" <EOL> if self . get_data ( ) is not None : <EOL> self . update ( ) <EOL> self . calculate ( ) <EOL> def set_data ( self , data , column = None ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( data , pd . DataFrame ) : <EOL> data = ColumnDataSource ( data ) <EOL> if isinstance ( data , ColumnDataSource ) : <EOL> self . source = data <EOL> if column is not None : <EOL> self . column = column <EOL> else : <EOL> self . values = data <EOL> self . update ( ) <EOL> self . calculate ( ) <EOL> def get_data ( self , column = None ) : <EOL> """<STR_LIT>""" <EOL> if self . source is not None and ( self . column is not None or column is not None ) : <EOL> if column is not None : <EOL> col = column <EOL> else : <EOL> col = self . column <EOL> return pd . Series ( self . source . data [ col ] ) <EOL> elif self . values is None and self . source is not None : <EOL> return pd . Series ( self . source . to_df ( ) . index ) <EOL> elif self . values is not None : <EOL> return self . values <EOL> else : <EOL> return None <EOL> def calculate ( self ) : <EOL> """<STR_LIT>""" <EOL> raise NotImplementedError ( '<STR_LIT>' <EOL> '<STR_LIT>' ) <EOL> def update ( self ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class Sum ( Stat ) : <EOL> def calculate ( self ) : <EOL> self . value = self . get_data ( ) . sum ( ) <EOL> class Mean ( Stat ) : <EOL> def calculate ( self ) : <EOL> self . value = self . get_data ( ) . mean ( ) <EOL> class Count ( Stat ) : <EOL> def calculate ( self ) : <EOL> self . value = self . get_data ( ) . count ( ) <EOL> class CountDistinct ( Stat ) : <EOL> def calculate ( self ) : <EOL> self . value = self . get_data ( ) . nunique ( ) <EOL> class Median ( Stat ) : <EOL> def calculate ( self ) : <EOL> self . value = self . get_data ( ) . median ( ) <EOL> class StdDeviation ( Stat ) : <EOL> def calculate ( self ) : <EOL> self . value = self . get_data ( ) . std ( ) <EOL> class Min ( Stat ) : <EOL> def calculate ( self ) : <EOL> self . value = self . get_data ( ) . min ( ) <EOL> class Max ( Stat ) : <EOL> def calculate ( self ) : <EOL> self . value = self . get_data ( ) . max ( ) <EOL> class Quantile ( Stat ) : <EOL> """<STR_LIT>""" <EOL> interval = Float ( default = <NUM_LIT:0.5> ) <EOL> def calculate ( self ) : <EOL> self . value = self . get_data ( ) . quantile ( self . interval ) <EOL> class Bin ( Stat ) : <EOL> """<STR_LIT>""" <EOL> label = Either ( String , List ( String ) ) <EOL> start = Either ( Float , List ( Float ) ) <EOL> stop = Either ( Float , List ( Float ) ) <EOL> start_label = String ( ) <EOL> stop_label = String ( ) <EOL> center = Either ( Float , List ( Float ) ) <EOL> stat = Instance ( Stat , default = Count ( ) ) <EOL> def __init__ ( self , bin_label , values = None , source = None , ** properties ) : <EOL> if isinstance ( bin_label , tuple ) : <EOL> bin_label = list ( bin_label ) <EOL> else : <EOL> bin_label = [ bin_label ] <EOL> properties [ '<STR_LIT:label>' ] = bin_label <EOL> bounds = self . process_bounds ( bin_label ) <EOL> starts , stops = zip ( * bounds ) <EOL> centers = [ ( start + stop ) / <NUM_LIT> for start , stop in zip ( starts , stops ) ] <EOL> if len ( starts ) == <NUM_LIT:1> : <EOL> starts = starts [ <NUM_LIT:0> ] <EOL> stops = stops [ <NUM_LIT:0> ] <EOL> centers = centers [ <NUM_LIT:0> ] <EOL> else : <EOL> starts = list ( starts ) <EOL> stops = list ( stops ) <EOL> centers = list ( centers ) <EOL> properties [ '<STR_LIT:start>' ] = starts <EOL> properties [ '<STR_LIT>' ] = stops <EOL> properties [ '<STR_LIT>' ] = centers <EOL> properties [ '<STR_LIT>' ] = values <EOL> super ( Bin , self ) . __init__ ( ** properties ) <EOL> @ staticmethod <EOL> def binstr_to_list ( bins ) : <EOL> """<STR_LIT>""" <EOL> value_chunks = bins . split ( '<STR_LIT:U+002C>' ) <EOL> value_chunks = [ val . replace ( '<STR_LIT:[>' , '<STR_LIT>' ) . replace ( '<STR_LIT:]>' , '<STR_LIT>' ) . replace ( '<STR_LIT:(>' , '<STR_LIT>' ) . replace ( '<STR_LIT:)>' , '<STR_LIT>' ) for val in value_chunks ] <EOL> bin_values = [ float ( value ) for value in value_chunks ] <EOL> return bin_values [ <NUM_LIT:0> ] , bin_values [ <NUM_LIT:1> ] <EOL> def process_bounds ( self , bin_label ) : <EOL> if isinstance ( bin_label , list ) : <EOL> return [ self . binstr_to_list ( dim ) for dim in bin_label ] <EOL> else : <EOL> return [ self . binstr_to_list ( bin_label ) ] <EOL> def update ( self ) : <EOL> self . stat . set_data ( self . values ) <EOL> def calculate ( self ) : <EOL> self . value = self . stat . value <EOL> class BinStats ( Stat ) : <EOL> """<STR_LIT>""" <EOL> bin_count = Either ( Int , Float ) <EOL> bin_width = Float ( default = None , help = '<STR_LIT>' ) <EOL> q1 = Quantile ( interval = <NUM_LIT> ) <EOL> q3 = Quantile ( interval = <NUM_LIT> ) <EOL> labels = List ( String ) <EOL> def __init__ ( self , values = None , column = None , ** properties ) : <EOL> properties [ '<STR_LIT>' ] = values <EOL> properties [ '<STR_LIT>' ] = column or '<STR_LIT>' <EOL> super ( BinStats , self ) . __init__ ( ** properties ) <EOL> def update ( self ) : <EOL> values = self . get_data ( ) <EOL> self . q1 . set_data ( values ) <EOL> self . q3 . set_data ( values ) <EOL> if self . bin_count is None : <EOL> self . calc_num_bins ( values ) <EOL> def calc_num_bins ( self , values ) : <EOL> """<STR_LIT>""" <EOL> iqr = self . q3 . value - self . q1 . value <EOL> if iqr == <NUM_LIT:0> : <EOL> self . bin_width = np . sqrt ( values . size ) <EOL> else : <EOL> self . bin_width = <NUM_LIT:2> * iqr * ( len ( values ) ** - ( <NUM_LIT:1.> / <NUM_LIT> ) ) <EOL> self . bin_count = int ( np . ceil ( ( values . max ( ) - values . min ( ) ) / self . bin_width ) ) <EOL> if self . bin_count <= <NUM_LIT:1> : <EOL> self . bin_count = <NUM_LIT:3> <EOL> def calculate ( self ) : <EOL> pass <EOL> class Bins ( Stat ) : <EOL> """<STR_LIT>""" <EOL> bin_stat = Instance ( BinStats , help = """<STR_LIT>""" ) <EOL> bins = List ( Instance ( Bin ) , help = """<STR_LIT>""" ) <EOL> stat = Instance ( Stat , default = Count ( ) , help = """<STR_LIT>""" ) <EOL> bin_count = Int ( help = """<STR_LIT>""" ) <EOL> bin_column = String ( ) <EOL> centers_column = String ( ) <EOL> aggregate = Bool ( default = True ) <EOL> bin_values = Bool ( default = False ) <EOL> bin_width = Float ( ) <EOL> def __init__ ( self , values = None , column = None , bins = None , <EOL> stat = '<STR_LIT:count>' , source = None , ** properties ) : <EOL> if isinstance ( stat , str ) : <EOL> stat = stats [ stat ] ( ) <EOL> properties [ '<STR_LIT>' ] = column or '<STR_LIT>' <EOL> properties [ '<STR_LIT>' ] = bins <EOL> properties [ '<STR_LIT>' ] = stat <EOL> properties [ '<STR_LIT>' ] = values <EOL> properties [ '<STR_LIT:source>' ] = source <EOL> super ( Bins , self ) . __init__ ( ** properties ) <EOL> def _get_stat ( self ) : <EOL> stat_kwargs = { } <EOL> if self . source is not None : <EOL> stat_kwargs [ '<STR_LIT:source>' ] = self . source <EOL> stat_kwargs [ '<STR_LIT>' ] = self . column <EOL> elif self . values is not None : <EOL> stat_kwargs [ '<STR_LIT>' ] = self . values <EOL> stat_kwargs [ '<STR_LIT>' ] = self . bin_count <EOL> return BinStats ( ** stat_kwargs ) <EOL> def update ( self ) : <EOL> self . bin_stat = self . _get_stat ( ) <EOL> self . bin_stat . update ( ) <EOL> def calculate ( self ) : <EOL> bin_str = '<STR_LIT>' <EOL> self . bin_column = self . column + bin_str <EOL> bin_models = [ ] <EOL> data = self . bin_stat . get_data ( ) <EOL> bins = self . bin_stat . bin_count <EOL> if data . size < <NUM_LIT:2> : <EOL> raise ValueError ( '<STR_LIT>' ) <EOL> if data . ndim == <NUM_LIT:1> and data . std ( ) == <NUM_LIT:0> : <EOL> margin = <NUM_LIT> * abs ( float ( data [ <NUM_LIT:0> ] ) ) or <NUM_LIT> <EOL> bins = np . linspace ( data [ <NUM_LIT:0> ] - margin , data [ <NUM_LIT:0> ] + margin , bins + <NUM_LIT:1> ) <EOL> binned , bin_bounds = pd . cut ( data , bins , <EOL> retbins = True , include_lowest = True , precision = <NUM_LIT:0> ) <EOL> self . bin_width = np . round ( bin_bounds [ <NUM_LIT:2> ] - bin_bounds [ <NUM_LIT:1> ] , <NUM_LIT:1> ) <EOL> if self . source is not None : <EOL> self . source . add ( binned . tolist ( ) , name = self . bin_column ) <EOL> df = self . source . to_df ( ) <EOL> else : <EOL> df = pd . DataFrame ( { self . column : self . values , self . bin_column : binned } ) <EOL> for name , group in df . groupby ( self . bin_column ) : <EOL> bin_models . append ( Bin ( bin_label = name , values = group [ self . column ] , <EOL> stat = self . stat ) ) <EOL> self . bins = bin_models <EOL> centers = binned . copy ( ) <EOL> centers = centers . astype ( str ) <EOL> for bin in self . bins : <EOL> centers [ binned == bin . label ] = bin . center <EOL> self . centers_column = self . column + '<STR_LIT>' <EOL> if self . source is not None : <EOL> self . source . add ( centers . tolist ( ) , name = self . centers_column ) <EOL> else : <EOL> df [ self . centers_column ] = centers <EOL> def __getitem__ ( self , item ) : <EOL> return self . bins [ item ] <EOL> def apply ( self , data ) : <EOL> self . set_data ( data . source ) <EOL> return self . source . to_df ( ) <EOL> def sort ( self , ascending = True ) : <EOL> if self . bins is not None : <EOL> self . bins = list ( sorted ( self . bins , key = lambda x : x . center , <EOL> reverse = ~ ascending ) ) <EOL> def bins ( data , values = None , column = None , bin_count = None , labels = None , <EOL> ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( data , str ) : <EOL> column = data <EOL> values = None <EOL> else : <EOL> column = None <EOL> return Bins ( values = values , column = column , bin_count = bin_count , ** kwargs ) <EOL> stats = { <EOL> '<STR_LIT>' : Sum , <EOL> '<STR_LIT>' : Mean , <EOL> '<STR_LIT:count>' : Count , <EOL> '<STR_LIT>' : CountDistinct , <EOL> '<STR_LIT>' : Median , <EOL> '<STR_LIT>' : StdDeviation , <EOL> '<STR_LIT>' : Min , <EOL> '<STR_LIT>' : Max , <EOL> '<STR_LIT>' : Quantile <EOL> } </s>
