<s> """<STR_LIT>""" <EOL> from pyspark . mllib . regression import LabeledPoint <EOL> from constants import raster_extent <EOL> from raster_utils import coords_to_utm <EOL> def parse_taxi_record ( line ) : <EOL> """<STR_LIT>""" <EOL> ( medallion , hack_license , vendor_id , rate_code , store_and_fwd_flag , <EOL> pickup_datetime , dropoff_datetime , passenger_count , trip_time_in_secs , <EOL> trip_distance , pickup_longitude , pickup_latitude , dropoff_longitude , <EOL> dropoff_latitude ) = line . split ( '<STR_LIT:U+002C>' ) <EOL> pickup_date , pickup_time = pickup_datetime . split ( ) <EOL> pickup_year , pickup_month , pickup_day = pickup_date . split ( '<STR_LIT:->' ) <EOL> pickup_hour , pickup_minute , _ = pickup_time . split ( '<STR_LIT::>' ) <EOL> try : <EOL> pickup_lat , pickup_lng = coords_to_utm ( pickup_latitude , pickup_longitude ) <EOL> except : <EOL> pickup_lat , pickup_lng = <NUM_LIT:0.0> , <NUM_LIT:0.0> <EOL> def get_row_col ( x , y ) : <EOL> """<STR_LIT>""" <EOL> col = int ( ( x - raster_extent . x_min ) / raster_extent . cell_width ) <EOL> row = - <NUM_LIT:1> * int ( ( y - raster_extent . y_max ) / raster_extent . cell_height ) <EOL> return col , row <EOL> pickup_col , pickup_row = get_row_col ( pickup_lat , pickup_lng ) <EOL> period = "<STR_LIT>" . format ( <EOL> pickup_year = int ( pickup_year ) , pickup_month = int ( pickup_month ) , pickup_day = int ( pickup_day ) , <EOL> pickup_hour = int ( pickup_hour ) ) <EOL> return ( ( pickup_row , pickup_col , period ) , <NUM_LIT:1> ) <EOL> def combine_values ( d , v ) : <EOL> """<STR_LIT>""" <EOL> key , value = v <EOL> d [ key ] = value <EOL> return d <EOL> def merge_combiner ( d1 , d2 ) : <EOL> """<STR_LIT>""" <EOL> d1 . update ( d2 ) <EOL> return d1 <EOL> def labeled_point_to_row_col_period ( labeled_point ) : <EOL> """<STR_LIT>""" <EOL> features = labeled_point . features <EOL> row , col = features [ <NUM_LIT:0> ] , features [ <NUM_LIT:1> ] <EOL> month , day , hour = features [ <NUM_LIT:2> ] , features [ <NUM_LIT:3> ] , features [ <NUM_LIT:4> ] <EOL> period = '<STR_LIT>' . format ( int ( month ) , int ( day ) , int ( hour ) ) <EOL> return row , col , period <EOL> def get_labeled_points ( period , data_dict , data_ordering , mask , weekday , geovars , weather_data ) : <EOL> """<STR_LIT>""" <EOL> observed_arr = data_dict . pop ( '<STR_LIT>' , None ) <EOL> if observed_arr is None : <EOL> return [ ] <EOL> if set ( data_dict . keys ( ) ) != set ( data_ordering ) : <EOL> return [ ] <EOL> month = int ( period [ <NUM_LIT:4> : <NUM_LIT:6> ] ) <EOL> day = int ( period [ <NUM_LIT:6> : <NUM_LIT:8> ] ) <EOL> hour = int ( period [ - <NUM_LIT:2> : ] ) <EOL> weekday_list = weekday . value [ period [ : - <NUM_LIT:2> ] ] <EOL> weather = weather_data . value [ period ] <EOL> def get_labeled_point ( row , col ) : <EOL> observed = observed_arr [ ( row , col ) ] <EOL> measurements = [ row , col , month , day , hour ] + weekday_list + weather <EOL> for k in data_ordering : <EOL> value = data_dict [ k ] <EOL> measurements . append ( value [ ( row , col ) ] ) <EOL> for geovar in geovars : <EOL> measurements . append ( geovar . value [ ( row , col ) ] ) <EOL> return LabeledPoint ( observed , measurements ) <EOL> return [ get_labeled_point ( row , col ) for row , col in mask . value ] <EOL> def get_aggregates_rdd ( trips_rdd , hours , start , end , step_size ) : <EOL> """<STR_LIT>""" <EOL> def increment_period ( period , increment ) : <EOL> hours_since = hours . value [ '<STR_LIT>' ] [ period ] <EOL> incremented_hours = hours_since + increment <EOL> return hours . value [ '<STR_LIT>' ] [ incremented_hours ] <EOL> def get_aggregates ( args ) : <EOL> ( row , col , period ) , num = args <EOL> return [ ( ( row , col , increment_period ( period , x ) ) , num ) for x in xrange ( start , end , step_size ) ] <EOL> return trips_rdd . flatMap ( get_aggregates ) . reduceByKey ( lambda a , b : a + b ) </s>
