<s> '''<STR_LIT>''' <EOL> import argparse <EOL> import csv <EOL> import json <EOL> import logging <EOL> import sys <EOL> import time <EOL> from datetime import datetime , timedelta , tzinfo <EOL> from prelert . engineApiClient import EngineApiClient <EOL> HOST = '<STR_LIT:localhost>' <EOL> PORT = <NUM_LIT> <EOL> BASE_URL = '<STR_LIT>' <EOL> ZERO_OFFSET = timedelta ( <NUM_LIT:0> ) <EOL> class UtcOffset ( tzinfo ) : <EOL> '''<STR_LIT>''' <EOL> def utcoffset ( self , dt ) : <EOL> return ZERO_OFFSET <EOL> def tzname ( self , dt ) : <EOL> return "<STR_LIT>" <EOL> def dst ( self , dt ) : <EOL> return ZERO_OFFSET <EOL> def parseArguments ( ) : <EOL> parser = argparse . ArgumentParser ( ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" <EOL> + HOST , default = HOST ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" <EOL> + str ( PORT ) , default = PORT ) <EOL> parser . add_argument ( "<STR_LIT>" , help = "<STR_LIT>" , type = int , default = <NUM_LIT:0> ) <EOL> parser . add_argument ( "<STR_LIT:file>" , help = "<STR_LIT>" ) <EOL> return parser . parse_args ( ) <EOL> def generateRecords ( csv_filename , start_date , interval , end_date ) : <EOL> '''<STR_LIT>''' <EOL> csv_data = [ ] <EOL> csv_file = open ( csv_filename , '<STR_LIT:rb>' ) <EOL> reader = csv . reader ( csv_file ) <EOL> header = reader . next ( ) <EOL> time_field_idx = - <NUM_LIT:1> <EOL> for i in range ( len ( header ) ) : <EOL> if header [ i ] == '<STR_LIT:time>' : <EOL> time_field_idx = i <EOL> break <EOL> if time_field_idx == - <NUM_LIT:1> : <EOL> logging . error ( "<STR_LIT>" ) <EOL> return <EOL> reverse = False <EOL> while start_date < end_date : <EOL> try : <EOL> yield header <EOL> if len ( csv_data ) == <NUM_LIT:0> : <EOL> for row in reader : <EOL> row [ time_field_idx ] = start_date . isoformat ( ) <EOL> start_date += interval <EOL> csv_data . append ( row ) <EOL> yield row <EOL> if start_date > end_date : <EOL> break <EOL> csv_file . close ( ) <EOL> else : <EOL> if reverse : <EOL> for row in reversed ( csv_data ) : <EOL> row [ time_field_idx ] = start_date . isoformat ( ) <EOL> start_date += interval <EOL> yield row <EOL> if start_date > end_date : <EOL> break <EOL> else : <EOL> for row in csv_data : <EOL> row [ time_field_idx ] = start_date . isoformat ( ) <EOL> start_date += interval <EOL> yield row <EOL> if start_date > end_date : <EOL> break <EOL> reverse = not reverse <EOL> except KeyboardInterrupt : <EOL> raise StopIteration <EOL> def main ( ) : <EOL> args = parseArguments ( ) <EOL> start_date = datetime ( <NUM_LIT> , <NUM_LIT:0> <NUM_LIT:5> , <NUM_LIT> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , <NUM_LIT:0> , UtcOffset ( ) ) <EOL> interval = timedelta ( seconds = <NUM_LIT> ) <EOL> if args . duration <= <NUM_LIT:0> : <EOL> end_date = datetime . now ( UtcOffset ( ) ) <EOL> else : <EOL> duration = timedelta ( hours = args . duration ) <EOL> end_date = start_date + duration <EOL> job_config = '<STR_LIT>' <EOL> engine_client = EngineApiClient ( args . host , BASE_URL , args . port ) <EOL> ( http_status_code , response ) = engine_client . createJob ( job_config ) <EOL> if http_status_code != <NUM_LIT> : <EOL> print ( http_status_code , json . dumps ( response ) ) <EOL> return <EOL> job_id = response [ '<STR_LIT:id>' ] <EOL> print '<STR_LIT>' + job_id <EOL> record_generator = generateRecords ( args . file , start_date , interval , end_date ) <EOL> header = '<STR_LIT:U+002C>' . join ( next ( record_generator ) ) <EOL> header += '<STR_LIT:\n>' <EOL> count = <NUM_LIT:0> <EOL> try : <EOL> next_bucket_id = <NUM_LIT:1> <EOL> print <EOL> print "<STR_LIT>" <EOL> data = header <EOL> for record in record_generator : <EOL> csv = '<STR_LIT:U+002C>' . join ( record ) + '<STR_LIT:\n>' <EOL> data += csv <EOL> count += <NUM_LIT:1> <EOL> if count == <NUM_LIT:100> : <EOL> ( http_status_code , response ) = engine_client . upload ( job_id , data ) <EOL> if http_status_code != <NUM_LIT> : <EOL> print ( http_status_code , json . dumps ( response ) ) <EOL> break <EOL> ( http_status_code , response ) = engine_client . getBucketsByDate ( job_id = job_id , <EOL> start_date = str ( next_bucket_id ) , end_date = None ) <EOL> if http_status_code != <NUM_LIT:200> : <EOL> print ( http_status_code , json . dumps ( response ) ) <EOL> break <EOL> for bucket in response : <EOL> print "<STR_LIT>" . format ( bucket [ '<STR_LIT>' ] , bucket [ '<STR_LIT:id>' ] , <EOL> bucket [ '<STR_LIT>' ] , bucket [ '<STR_LIT>' ] ) <EOL> if len ( response ) > <NUM_LIT:0> : <EOL> next_bucket_id = int ( response [ - <NUM_LIT:1> ] [ '<STR_LIT:id>' ] ) + <NUM_LIT:1> <EOL> data = header <EOL> count = <NUM_LIT:0> <EOL> except KeyboardInterrupt : <EOL> print "<STR_LIT>" <EOL> ( http_status_code , response ) = engine_client . close ( job_id ) <EOL> if http_status_code != <NUM_LIT> : <EOL> print ( http_status_code , json . dumps ( response ) ) <EOL> if __name__ == "<STR_LIT:__main__>" : <EOL> main ( ) </s>
