<s> """<STR_LIT>""" <EOL> import math <EOL> import warnings <EOL> from . client import InfluxDBClient <EOL> class DataFrameClient ( InfluxDBClient ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ignore_nan = True , * args , ** kwargs ) : <EOL> super ( DataFrameClient , self ) . __init__ ( * args , ** kwargs ) <EOL> try : <EOL> global pd <EOL> import pandas as pd <EOL> except ImportError as ex : <EOL> raise ImportError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( ex = str ( ex ) ) ) <EOL> self . EPOCH = pd . Timestamp ( '<STR_LIT>' ) <EOL> self . ignore_nan = ignore_nan <EOL> def write_points ( self , data , * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> batch_size = kwargs . get ( '<STR_LIT>' ) <EOL> time_precision = kwargs . get ( '<STR_LIT>' , '<STR_LIT:s>' ) <EOL> if batch_size : <EOL> kwargs . pop ( '<STR_LIT>' ) <EOL> for key , data_frame in data . items ( ) : <EOL> number_batches = int ( math . ceil ( <EOL> len ( data_frame ) / float ( batch_size ) ) ) <EOL> for batch in range ( number_batches ) : <EOL> start_index = batch * batch_size <EOL> end_index = ( batch + <NUM_LIT:1> ) * batch_size <EOL> data = [ self . _convert_dataframe_to_json ( <EOL> name = key , <EOL> dataframe = data_frame . ix [ start_index : end_index ] . copy ( ) , <EOL> time_precision = time_precision ) ] <EOL> InfluxDBClient . write_points ( self , data , * args , ** kwargs ) <EOL> return True <EOL> else : <EOL> data = [ self . _convert_dataframe_to_json ( <EOL> name = key , dataframe = dataframe , time_precision = time_precision ) <EOL> for key , dataframe in data . items ( ) ] <EOL> return InfluxDBClient . write_points ( self , data , * args , ** kwargs ) <EOL> def write_points_with_precision ( self , data , time_precision = '<STR_LIT:s>' ) : <EOL> """<STR_LIT>""" <EOL> warnings . warn ( <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" <EOL> "<STR_LIT>" , <EOL> FutureWarning ) <EOL> return self . write_points ( data , time_precision = '<STR_LIT:s>' ) <EOL> def query ( self , query , time_precision = '<STR_LIT:s>' , chunked = False ) : <EOL> """<STR_LIT>""" <EOL> result = InfluxDBClient . query ( self , query = query , <EOL> time_precision = time_precision , <EOL> chunked = chunked ) <EOL> if len ( result ) == <NUM_LIT:0> : <EOL> return result <EOL> elif len ( result ) == <NUM_LIT:1> : <EOL> return self . _to_dataframe ( result [ <NUM_LIT:0> ] , time_precision ) <EOL> else : <EOL> ret = { } <EOL> for time_series in result : <EOL> ret [ time_series [ '<STR_LIT:name>' ] ] = self . _to_dataframe ( time_series , <EOL> time_precision ) <EOL> return ret <EOL> def _to_dataframe ( self , json_result , time_precision ) : <EOL> dataframe = pd . DataFrame ( data = json_result [ '<STR_LIT>' ] , <EOL> columns = json_result [ '<STR_LIT>' ] ) <EOL> if '<STR_LIT>' in dataframe . keys ( ) : <EOL> dataframe . sort_values ( [ '<STR_LIT:time>' , '<STR_LIT>' ] , inplace = True ) <EOL> else : <EOL> dataframe . sort_values ( [ '<STR_LIT:time>' ] , inplace = True ) <EOL> pandas_time_unit = time_precision <EOL> if time_precision == '<STR_LIT:m>' : <EOL> pandas_time_unit = '<STR_LIT>' <EOL> elif time_precision == '<STR_LIT:u>' : <EOL> pandas_time_unit = '<STR_LIT>' <EOL> dataframe . index = pd . to_datetime ( list ( dataframe [ '<STR_LIT:time>' ] ) , <EOL> unit = pandas_time_unit , <EOL> utc = True ) <EOL> del dataframe [ '<STR_LIT:time>' ] <EOL> return dataframe <EOL> def _convert_dataframe_to_json ( self , dataframe , name , time_precision = '<STR_LIT:s>' ) : <EOL> if not isinstance ( dataframe , pd . DataFrame ) : <EOL> raise TypeError ( '<STR_LIT>' <EOL> . format ( type ( dataframe ) ) ) <EOL> if not ( isinstance ( dataframe . index , pd . tseries . period . PeriodIndex ) or <EOL> isinstance ( dataframe . index , pd . tseries . index . DatetimeIndex ) ) : <EOL> raise TypeError ( '<STR_LIT>' ) <EOL> dataframe . index = dataframe . index . to_datetime ( ) <EOL> if dataframe . index . tzinfo is None : <EOL> dataframe . index = dataframe . index . tz_localize ( '<STR_LIT>' ) <EOL> dataframe [ '<STR_LIT:time>' ] = [ self . _datetime_to_epoch ( dt , time_precision ) <EOL> for dt in dataframe . index ] <EOL> data = { '<STR_LIT:name>' : name , <EOL> '<STR_LIT>' : [ str ( column ) for column in dataframe . columns ] , <EOL> '<STR_LIT>' : [ self . _convert_array ( x ) for x in dataframe . values ] } <EOL> return data <EOL> def _convert_array ( self , array ) : <EOL> try : <EOL> global np <EOL> import numpy as np <EOL> except ImportError as ex : <EOL> raise ImportError ( '<STR_LIT>' <EOL> '<STR_LIT>' . format ( ex = str ( ex ) ) ) <EOL> if self . ignore_nan : <EOL> number_types = ( int , float , np . number ) <EOL> condition = ( all ( isinstance ( el , number_types ) for el in array ) and <EOL> np . isnan ( array ) ) <EOL> return list ( np . where ( condition , None , array ) ) <EOL> else : <EOL> return list ( array ) <EOL> def _datetime_to_epoch ( self , datetime , time_precision = '<STR_LIT:s>' ) : <EOL> seconds = ( datetime - self . EPOCH ) . total_seconds ( ) <EOL> if time_precision == '<STR_LIT:s>' : <EOL> return seconds <EOL> elif time_precision == '<STR_LIT:m>' or time_precision == '<STR_LIT>' : <EOL> return seconds * <NUM_LIT:1000> <EOL> elif time_precision == '<STR_LIT:u>' : <EOL> return seconds * <NUM_LIT> </s>
