<s> import bandicoot as bc <EOL> from bandicoot . helper . group import group_records <EOL> from bandicoot . core import User , Record <EOL> from bisect import bisect_right <EOL> from functools import partial <EOL> import datetime as dt <EOL> import itertools <EOL> import csv <EOL> import math <EOL> def create_punchcards ( user , split_interval = <NUM_LIT> ) : <EOL> """<STR_LIT>""" <EOL> if not float ( <NUM_LIT> * <NUM_LIT> / split_interval ) . is_integer ( ) : <EOL> raise ValueError ( <EOL> "<STR_LIT>" ) <EOL> contacts_in = partial ( bc . individual . number_of_contacts , <EOL> direction = '<STR_LIT>' , interaction = '<STR_LIT>' , summary = None ) <EOL> contacts_out = partial ( bc . individual . number_of_contacts , <EOL> direction = '<STR_LIT>' , interaction = '<STR_LIT>' , summary = None ) <EOL> calls_in = partial ( bc . individual . number_of_interactions , <EOL> direction = '<STR_LIT>' , interaction = '<STR_LIT>' , summary = None ) <EOL> calls_out = partial ( bc . individual . number_of_interactions , <EOL> direction = '<STR_LIT>' , interaction = '<STR_LIT>' , summary = None ) <EOL> texts_in = partial ( bc . individual . number_of_interactions , <EOL> direction = '<STR_LIT>' , interaction = '<STR_LIT:text>' , summary = None ) <EOL> texts_out = partial ( bc . individual . number_of_interactions , <EOL> direction = '<STR_LIT>' , interaction = '<STR_LIT:text>' , summary = None ) <EOL> time_spent_in = partial ( bc . individual . call_duration , <EOL> direction = '<STR_LIT>' , interaction = '<STR_LIT>' , summary = None ) <EOL> time_spent_out = partial ( bc . individual . call_duration , <EOL> direction = '<STR_LIT>' , interaction = '<STR_LIT>' , summary = None ) <EOL> core_func = [ <EOL> ( contacts_in , "<STR_LIT>" ) , <EOL> ( contacts_out , "<STR_LIT>" ) , <EOL> ( calls_in , "<STR_LIT>" ) , <EOL> ( calls_out , "<STR_LIT>" ) , <EOL> ( texts_in , "<STR_LIT>" ) , <EOL> ( texts_out , "<STR_LIT>" ) <EOL> ] <EOL> time_func = [ <EOL> ( time_spent_in , "<STR_LIT>" ) , <EOL> ( time_spent_out , "<STR_LIT>" ) <EOL> ] <EOL> pc = [ ] <EOL> sections = [ <EOL> ( i + <NUM_LIT:1> ) * split_interval for i in range ( <NUM_LIT:7> * <NUM_LIT> * <NUM_LIT> / split_interval ) ] <EOL> temp_user = _extract_user_info ( user ) <EOL> for grouped_records in group_records ( user , groupby = '<STR_LIT>' ) : <EOL> week_records = list ( grouped_records ) <EOL> time_spent_rec = _transform_to_time_spent ( <EOL> week_records , split_interval , sections ) <EOL> pc . extend ( _calculate_channels ( <EOL> week_records , sections , split_interval , core_func , temp_user ) ) <EOL> pc . extend ( _calculate_channels ( <EOL> time_spent_rec , sections , split_interval , time_func , temp_user , len ( core_func ) ) ) <EOL> return pc <EOL> def to_csv ( punchcards , filename , digits = <NUM_LIT:5> ) : <EOL> """<STR_LIT>""" <EOL> with open ( filename , '<STR_LIT:wb>' ) as f : <EOL> w = csv . writer ( f ) <EOL> w . writerow ( [ '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT:value>' ] ) <EOL> def make_repr ( item ) : <EOL> if item is None : <EOL> return None <EOL> elif isinstance ( item , float ) : <EOL> return repr ( round ( item , digits ) ) <EOL> else : <EOL> return str ( item ) <EOL> for row in punchcards : <EOL> w . writerow ( [ make_repr ( item ) for item in row ] ) <EOL> def read_csv ( filename ) : <EOL> """<STR_LIT>""" <EOL> with open ( filename , '<STR_LIT:rb>' ) as f : <EOL> r = csv . reader ( f ) <EOL> next ( r ) <EOL> pc = list ( r ) <EOL> for i , row in enumerate ( pc ) : <EOL> row [ <NUM_LIT:1> : <NUM_LIT:4> ] = map ( int , row [ <NUM_LIT:1> : <NUM_LIT:4> ] ) <EOL> row [ <NUM_LIT:4> ] = float ( row [ <NUM_LIT:4> ] ) <EOL> return pc <EOL> def _calculate_channels ( records , sections , split_interval , channel_funcs , user , c_start = <NUM_LIT:0> ) : <EOL> """<STR_LIT>""" <EOL> week_matrix = [ ] <EOL> if len ( records ) == <NUM_LIT:0> : <EOL> return week_matrix <EOL> if not isinstance ( records , list ) : <EOL> records = [ records ] <EOL> year_week = str ( records [ <NUM_LIT:0> ] . datetime . isocalendar ( ) [ <EOL> <NUM_LIT:0> ] ) + '<STR_LIT:->' + str ( records [ <NUM_LIT:0> ] . datetime . isocalendar ( ) [ <NUM_LIT:1> ] ) <EOL> section_lists , section_id = _punchcard_grouping ( records , sections , split_interval ) <EOL> for c , fun in enumerate ( channel_funcs ) : <EOL> for b , section_records in enumerate ( section_lists ) : <EOL> indicator_fun , return_type = fun <EOL> user . _records = section_records <EOL> output = indicator_fun ( user ) [ '<STR_LIT>' ] [ '<STR_LIT>' ] . values ( ) [ <NUM_LIT:0> ] <EOL> if return_type == '<STR_LIT>' : <EOL> indicator = sum ( d for d in output if d is not None ) <EOL> elif return_type == '<STR_LIT>' : <EOL> indicator = sum ( d for group in output for d in group if d is not None ) <EOL> if indicator != <NUM_LIT:0> : <EOL> week_matrix . append ( <EOL> [ year_week , c + c_start , section_id [ b ] [ <NUM_LIT:0> ] , section_id [ b ] [ <NUM_LIT:1> ] , float ( indicator ) ] ) <EOL> return week_matrix <EOL> def _punchcard_grouping ( records , sections , split_interval ) : <EOL> """<STR_LIT>""" <EOL> def _group_by_weektime ( records , sections ) : <EOL> for _ , group in itertools . groupby ( records , key = lambda r : bisect_right ( sections , _find_weektime ( r . datetime ) ) ) : <EOL> yield group <EOL> section_records = _group_by_weektime ( records , sections ) <EOL> section_lists = _extract_list_from_generator ( section_records ) <EOL> section_indices = [ bisect_right ( <EOL> sections , _find_weektime ( r_list [ <NUM_LIT:0> ] . datetime ) ) for r_list in section_lists ] <EOL> section_id = _find_day_section_from_indices ( <EOL> section_indices , split_interval ) <EOL> assert ( len ( section_lists ) == len ( section_id ) and len ( <EOL> section_indices ) == len ( set ( section_indices ) ) ) <EOL> return section_lists , section_id <EOL> def _transform_to_time_spent ( records , split_interval , sections ) : <EOL> """<STR_LIT>""" <EOL> t_records = [ ] <EOL> week_nr = records [ <NUM_LIT:0> ] . datetime . isocalendar ( ) [ <NUM_LIT:1> ] <EOL> for r in filter ( lambda rec : rec . interaction == '<STR_LIT>' and rec . call_duration > <NUM_LIT:0> , records ) : <EOL> t_left = r . call_duration <EOL> t_to_next_section = _seconds_to_section_split ( r , sections ) <EOL> t_spent_total = <NUM_LIT:0> <EOL> while ( t_left > <NUM_LIT:0> ) : <EOL> t_spent = min ( t_to_next_section , t_left ) <EOL> dt_new = r . datetime + dt . timedelta ( seconds = t_spent_total ) <EOL> if dt_new . isocalendar ( ) [ <NUM_LIT:1> ] > week_nr : <EOL> dt_new -= dt . timedelta ( days = <NUM_LIT:7> ) <EOL> t_records . append ( <EOL> Record ( '<STR_LIT>' , r . direction , None , dt_new , t_spent , None ) ) <EOL> t_left -= t_spent <EOL> t_spent_total += t_spent <EOL> t_to_next_section = split_interval * <NUM_LIT> <EOL> return sorted ( t_records , key = lambda r : _find_weektime ( r . datetime ) ) <EOL> def _extract_user_info ( user ) : <EOL> """<STR_LIT>""" <EOL> temp_user = User ( ) <EOL> copy_attributes = [ <EOL> '<STR_LIT>' , '<STR_LIT:name>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' , '<STR_LIT>' ] <EOL> for attr in copy_attributes : <EOL> setattr ( temp_user , attr , getattr ( user , attr ) ) <EOL> return temp_user <EOL> def _find_weektime ( datetime , time_type = '<STR_LIT>' ) : <EOL> """<STR_LIT>""" <EOL> if time_type == '<STR_LIT>' : <EOL> return datetime . weekday ( ) * <NUM_LIT> * <NUM_LIT> * <NUM_LIT> + datetime . hour * <NUM_LIT> * <NUM_LIT> + datetime . minute * <NUM_LIT> + datetime . second <EOL> elif time_type == '<STR_LIT>' : <EOL> return datetime . weekday ( ) * <NUM_LIT> * <NUM_LIT> + datetime . hour * <NUM_LIT> + datetime . minute <EOL> else : <EOL> raise ValueError ( "<STR_LIT>" ) <EOL> def _extract_list_from_generator ( generator ) : <EOL> """<STR_LIT>""" <EOL> extracted = [ ] <EOL> for i in generator : <EOL> extracted . append ( list ( i ) ) <EOL> return extracted <EOL> def _seconds_to_section_split ( record , sections ) : <EOL> """<STR_LIT>""" <EOL> next_section = sections [ <EOL> bisect_right ( sections , _find_weektime ( record . datetime ) ) ] * <NUM_LIT> <EOL> return next_section - _find_weektime ( record . datetime , time_type = '<STR_LIT>' ) <EOL> def _find_day_section_from_indices ( indices , split_interval ) : <EOL> """<STR_LIT>""" <EOL> cells_day = <NUM_LIT> * <NUM_LIT> / split_interval <EOL> return [ [ int ( math . floor ( i / cells_day ) ) , i % cells_day ] for i in indices ] </s>
