signature:
def signal_start ( self , config , test_id = None , ** kwargs ) :
    """Initialize an analysis object and set ts _ start for the analysis represented by test _ id
 : param test _ id : integer that represents the analysis
 : param config : config can be a ConfigParser . ConfigParser object or a string specifying local or http ( s ) location
 for config
 : return : test _ id"""

implementation:
if not test_id :
    self . _default_test_id += 1
    test_id = self . _default_test_id
self . _analyses [ test_id ] = _Analysis ( naarad . utils . get_standardized_timestamp ( 'now' , None ) , config , test_id = test_id )
if kwargs :
    if 'description' in kwargs . keys ( ) :
        self . _analyses [ test_id ] . description = kwargs [ 'description' ]
    if 'input_directory' in kwargs . keys ( ) :
        self . _analyses [ test_id ] . input_directory = kwargs [ 'input_directory' ]
    if 'output_directory' in kwargs . keys ( ) :
        self . _analyses [ test_id ] . output_directory = kwargs [ 'output_directory' ]
return test_id


signature:
def signal_stop ( self , test_id = None ) :
    """Set ts _ end for the analysis represented by test _ id
 : param test _ id : integer that represents the analysis
 : return : test _ id"""

implementation:
if test_id is None :
    test_id = self . _default_test_id
if self . _analyses [ test_id ] . ts_end :
    return CONSTANTS . OK
self . _analyses [ test_id ] . ts_end = naarad . utils . get_standardized_timestamp ( 'now' , None )
return CONSTANTS . OK


signature:
def get_failed_analyses ( self ) :
    """Returns a list of test _ id for which naarad analysis failed
 : return : list of test _ ids"""

implementation:
failed_analyses = [ ]
for test_id in self . _analyses . keys ( ) :
    if self . _analyses [ test_id ] . status != CONSTANTS . OK :
        failed_analyses . append ( test_id )
return failed_analyses


signature:
def _set_sla_data ( self , test_id , metrics ) :
    """Get sla data from each metric and set it in the _ Analysis object specified by test _ id to make it available
 for retrieval
 : return : currently always returns CONSTANTS . OK . Maybe enhanced in future to return additional status"""

implementation:
for metric in metrics :
    self . _analyses [ test_id ] . sla_data [ metric . label ] = metric . sla_map
return CONSTANTS . OK


signature:
def _set_stats_data ( self , test_id , metrics ) :
    """Get summary stats data from each metric and set it in the _ Analysis object specified by test _ id to make it available
 for retrieval
 : return : currently always returns CONSTANTS . OK . Maybe enhanced in future to return additional status"""

implementation:
for metric in metrics :
    self . _analyses [ test_id ] . stats_data [ metric . label ] = metric . summary_stats
return CONSTANTS . OK


signature:
def _create_output_directories ( self , analysis ) :
    """Create the necessary output and resource directories for the specified analysis
 : param : analysis : analysis associated with a given test _ id"""

implementation:
try :
    os . makedirs ( analysis . output_directory )
except OSError as exception :
    if exception . errno != errno . EEXIST :
        raise
try :
    resource_directory = os . path . join ( analysis . output_directory , analysis . resource_path )
    os . makedirs ( resource_directory )
except OSError as exception :
    if exception . errno != errno . EEXIST :
        raise


signature:
def _run_pre ( self , analysis , run_steps ) :
    """If Naarad is run in CLI mode , execute any pre run steps specified in the config . ts _ start / ts _ end are set based on
 workload run steps if any .
 : param : analysis : The analysis object being processed
 : param : run _ steps : list of post run steps"""

implementation:
workload_run_steps = [ ]
for run_step in sorted ( run_steps , key = lambda step : step . run_rank ) :
    run_step . run ( )
    if run_step . run_type == CONSTANTS . RUN_TYPE_WORKLOAD :
        workload_run_steps . append ( run_step )
# Get analysis time period from workload run steps
if len ( workload_run_steps ) > 0 :
    analysis . ts_start , analysis . ts_end = naarad . utils . get_run_time_period ( workload_run_steps )
return CONSTANTS . OK


signature:
def _run_post ( self , run_steps ) :
    """If Naarad is run in CLI mode , execute any post run steps specified in the config
 : param : run _ steps : list of post run steps"""

implementation:
for run_step in sorted ( run_steps , key = lambda step : step . run_rank ) :
    run_step . run ( )
return CONSTANTS . OK


signature:
def _process_args ( self , analysis , args ) :
    """When Naarad is run in CLI mode , get the CL arguments and update the analysis
 : param : analysis : The analysis being processed
 : param : args : Command Line Arguments received by naarad"""

implementation:
if args . exit_code :
    self . return_exit_code = args . exit_code
if args . no_plots :
    self . skip_plots = args . no_plots
if args . start :
    analysis . ts_start = naarad . utils . get_standardized_timestamp ( args . start , None )
if args . end :
    analysis . ts_end = naarad . utils . get_standardized_timestamp ( args . end , None )
if args . variables :
    analysis . variables = naarad . utils . get_variables ( args )
return CONSTANTS . OK


signature:
def analyze ( self , input_directory , output_directory , ** kwargs ) :
    """Run all the analysis saved in self . _ analyses , sorted by test _ id .
 This is useful when Naarad ( ) is used by other programs and multiple analyses are run
 In naarad CLI mode , len ( _ analyses ) = = 1
 : param : input _ directory : location of log files
 : param : output _ directory : root directory for analysis output
 : param : * * kwargs : Optional keyword args
 : return : int : status code ."""

implementation:
is_api_call = True
if len ( self . _analyses ) == 0 :
    if 'config' not in kwargs . keys ( ) :
        return CONSTANTS . ERROR
    self . create_analysis ( kwargs [ 'config' ] )
if 'args' in kwargs :
    self . _process_args ( self . _analyses [ 0 ] , kwargs [ 'args' ] )
    is_api_call = False
error_count = 0
self . _input_directory = input_directory
self . _output_directory = output_directory
for test_id in sorted ( self . _analyses . keys ( ) ) : # Setup
    if not self . _analyses [ test_id ] . input_directory :
        self . _analyses [ test_id ] . input_directory = input_directory
    if not self . _analyses [ test_id ] . output_directory :
        if len ( self . _analyses ) > 1 :
            self . _analyses [ test_id ] . output_directory = os . path . join ( output_directory , str ( test_id ) )
        else :
            self . _analyses [ test_id ] . output_directory = output_directory
    if ( 'config' in kwargs . keys ( ) ) and ( not self . _analyses [ test_id ] . config ) :
        self . _analyses [ test_id ] . config = kwargs [ 'config' ]
    self . _create_output_directories ( self . _analyses [ test_id ] )
    # Actually run analysis
    self . _analyses [ test_id ] . status = self . run ( self . _analyses [ test_id ] , is_api_call , ** kwargs )
    if self . _analyses [ test_id ] . status != CONSTANTS . OK :
        error_count += 1
if len ( self . _analyses ) == 1 :
    return self . _analyses [ 0 ] . status
elif error_count > 0 :
    return CONSTANTS . ERROR
else :
    return CONSTANTS . OK


signature:
def run ( self , analysis , is_api_call , ** kwargs ) :
    """: param analysis : Run naarad analysis for the specified analysis object
 : param * * kwargs : Additional keyword args can be passed in here for future enhancements
 : return :"""

implementation:
threads = [ ]
crossplots = [ ]
report_args = { }
metrics = defaultdict ( )
run_steps = defaultdict ( list )
discovery_mode = False
graph_timezone = None
graphing_library = None
if isinstance ( analysis . config , str ) :
    if not naarad . utils . is_valid_file ( analysis . config ) :
        return CONSTANTS . INVALID_CONFIG
    config_object = ConfigParser . ConfigParser ( analysis . variables )
    config_object . optionxform = str
    config_object . read ( analysis . config )
elif isinstance ( analysis . config , ConfigParser . ConfigParser ) :
    config_object = analysis . config
else :
    if is_api_call :
        return CONSTANTS . INVALID_CONFIG
    else :
        metrics [ 'metrics' ] = naarad . utils . discover_by_name ( analysis . input_directory , analysis . output_directory )
        if len ( metrics [ 'metrics' ] ) == 0 :
            logger . warning ( 'Unable to auto detect metrics in the specified input directory: %s' , analysis . input_directory )
            return CONSTANTS . ERROR
        else :
            discovery_mode = True
            metrics [ 'aggregate_metrics' ] = [ ]
if not discovery_mode :
    metrics , run_steps , crossplots , report_args , graph_timezone , graphing_library = self . _process_naarad_config ( config_object , analysis )
if graphing_library is None :
    graphing_library = CONSTANTS . DEFAULT_GRAPHING_LIBRARY
# If graphing libraries are not installed , skip static images
if graphing_library not in self . available_graphing_modules . keys ( ) :
    logger . error ( "Naarad cannot import graphing library %s on your system. Will not generate static charts" , graphing_library )
    self . skip_plots = True
if not is_api_call :
    self . _run_pre ( analysis , run_steps [ 'pre' ] )
for metric in metrics [ 'metrics' ] :
    if analysis . ts_start :
        metric . ts_start = analysis . ts_start
    if analysis . ts_end :
        metric . ts_end = analysis . ts_end
    thread = threading . Thread ( target = naarad . utils . parse_and_plot_single_metrics , args = ( metric , graph_timezone , analysis . output_directory , analysis . input_directory , graphing_library , self . skip_plots ) )
    thread . start ( )
    threads . append ( thread )
for t in threads :
    t . join ( )
for metric in metrics [ 'aggregate_metrics' ] :
    thread = threading . Thread ( target = naarad . utils . parse_and_plot_single_metrics , args = ( metric , graph_timezone , analysis . output_directory , analysis . input_directory , graphing_library , self . skip_plots ) )
    thread . start ( )
    threads . append ( thread )
for t in threads :
    t . join ( )
self . _set_sla_data ( analysis . test_id , metrics [ 'metrics' ] + metrics [ 'aggregate_metrics' ] )
self . _set_stats_data ( analysis . test_id , metrics [ 'metrics' ] + metrics [ 'aggregate_metrics' ] )
if len ( crossplots ) > 0 and not self . skip_plots :
    correlated_plots = naarad . utils . nway_plotting ( crossplots , metrics [ 'metrics' ] + metrics [ 'aggregate_metrics' ] , os . path . join ( analysis . output_directory , analysis . resource_path ) , analysis . resource_path , graphing_library )
else :
    correlated_plots = [ ]
rpt = reporting_modules [ 'report' ] ( None , analysis . output_directory , os . path . join ( analysis . output_directory , analysis . resource_path ) , analysis . resource_path , metrics [ 'metrics' ] + metrics [ 'aggregate_metrics' ] , correlated_plots = correlated_plots , ** report_args )
rpt . generate ( )
if not is_api_call :
    self . _run_post ( run_steps [ 'post' ] )
if self . return_exit_code :
    for metric in metrics [ 'metrics' ] + metrics [ 'aggregate_metrics' ] :
        if metric . status == CONSTANTS . SLA_FAILED :
            return CONSTANTS . SLA_FAILURE
return CONSTANTS . OK


signature:
def diff ( self , test_id_1 , test_id_2 , config = None , ** kwargs ) :
    """Create a diff report using test _ id _ 1 as a baseline
 : param : test _ id _ 1 : test id to be used as baseline
 : param : test _ id _ 2 : test id to compare against baseline
 : param : config file for diff ( optional )
 : param : * * kwargs : keyword arguments"""

implementation:
output_directory = os . path . join ( self . _output_directory , 'diff_' + str ( test_id_1 ) + '_' + str ( test_id_2 ) )
if kwargs :
    if 'output_directory' in kwargs . keys ( ) :
        output_directory = kwargs [ 'output_directory' ]
diff_report = Diff ( [ NaaradReport ( self . _analyses [ test_id_1 ] . output_directory , None ) , NaaradReport ( self . _analyses [ test_id_2 ] . output_directory , None ) ] , 'diff' , output_directory , os . path . join ( output_directory , self . _resource_path ) , self . _resource_path )
if config :
    naarad . utils . extract_diff_sla_from_config_file ( diff_report , config )
diff_report . generate ( )
if diff_report . sla_failures > 0 :
    return CONSTANTS . SLA_FAILURE
if diff_report . status != 'OK' :
    return CONSTANTS . ERROR
return CONSTANTS . OK


signature:
def diff_reports_by_location ( self , report1_location , report2_location , output_directory , config = None , ** kwargs ) :
    """Create a diff report using report1 as a baseline
 : param : report1 _ location : report to be used as baseline
 : param : report2 _ location : report to compare against baseline
 : param : config file for diff ( optional )
 : param : * * kwargs : keyword arguments"""

implementation:
if kwargs :
    if 'output_directory' in kwargs . keys ( ) :
        output_directory = kwargs [ 'output_directory' ]
diff_report = Diff ( [ NaaradReport ( report1_location , None ) , NaaradReport ( report2_location , None ) ] , 'diff' , output_directory , os . path . join ( output_directory , self . _resource_path ) , self . _resource_path )
if config :
    naarad . utils . extract_diff_sla_from_config_file ( diff_report , config )
diff_report . generate ( )
if diff_report . sla_failures > 0 :
    return CONSTANTS . SLA_FAILURE
if diff_report . status != 'OK' :
    return CONSTANTS . ERROR
return CONSTANTS . OK


signature:
def _process_naarad_config ( self , config , analysis ) :
    """Process the config file associated with a particular analysis and return metrics , run _ steps and crossplots .
 Also sets output directory and resource _ path for an anlaysis"""

implementation:
graph_timezone = None
output_directory = analysis . output_directory
resource_path = analysis . resource_path
run_steps = defaultdict ( list )
metrics = defaultdict ( list )
indir_default = ''
crossplots = [ ]
report_args = { }
graphing_library = None
ts_start , ts_end = None , None
if config . has_section ( 'GLOBAL' ) :
    ts_start , ts_end = naarad . utils . parse_global_section ( config , 'GLOBAL' )
    if config . has_option ( 'GLOBAL' , 'user_defined_metrics' ) :
        naarad . utils . parse_user_defined_metric_classes ( config , metric_classes )
    config . remove_section ( 'GLOBAL' )
if config . has_section ( 'REPORT' ) :
    report_args = naarad . utils . parse_report_section ( config , 'REPORT' )
    config . remove_section ( 'REPORT' )
for section in config . sections ( ) : # GRAPH section is optional
    if section == 'GRAPH' :
        graphing_library , crossplots , outdir_default , indir_default , graph_timezone = naarad . utils . parse_graph_section ( config , section , output_directory , indir_default )
    elif section . startswith ( 'RUN-STEP' ) :
        run_step = naarad . utils . parse_run_step_section ( config , section )
        if not run_step :
            logger . error ( 'Ignoring section %s, could not parse it correctly' , section )
            continue
        if run_step . run_order == CONSTANTS . PRE_ANALYSIS_RUN :
            run_steps [ 'pre' ] . append ( run_step )
        # DURING _ ANALYSIS _ RUN not supported yet
        elif run_step . run_order == CONSTANTS . DURING_ANALYSIS_RUN :
            run_steps [ 'in' ] . append ( run_step )
        elif run_step . run_order == CONSTANTS . POST_ANALYSIS_RUN :
            run_steps [ 'post' ] . append ( run_step )
        else :
            logger . error ( 'Unknown RUN-STEP run_order specified' )
    else : # section name is used to create sub - directories , so enforce it .
        if not naarad . utils . is_valid_metric_name ( section ) :
            logger . critical ( 'Section name %s is invalid! Only letters, digits, dot(.), dash(-), underscore(_) are allowed' % section )
            return CONSTANTS . CRITICAL_FAILURE
        if section == 'SAR-*' :
            hostname , infile , label , ts_start , ts_end , precision , kwargs , rule_strings = naarad . utils . parse_basic_metric_options ( config , section )
            sar_metrics = naarad . utils . get_all_sar_objects ( metrics , infile , hostname , output_directory , label , ts_start , ts_end , None )
            for sar_metric in sar_metrics :
                if sar_metric . ts_start is None and ( sar_metric . ts_end is None or sar_metric . ts_end > ts_start ) :
                    sar_metric . ts_start = ts_start
                if sar_metric . ts_end is None and ( sar_metric . ts_start is None or ts_end > sar_metric . ts_start ) :
                    sar_metric . ts_end = ts_end
            metrics [ 'metrics' ] . extend ( sar_metrics )
        else :
            new_metric = naarad . utils . parse_metric_section ( config , section , metric_classes , metrics [ 'metrics' ] , aggregate_metric_classes , output_directory , resource_path )
            if new_metric . ts_start is None and ( new_metric . ts_end is None or new_metric . ts_end > ts_start ) :
                new_metric . ts_start = ts_start
            if new_metric . ts_end is None and ( new_metric . ts_start is None or ts_end > new_metric . ts_start ) :
                new_metric . ts_end = ts_end
            metric_type = section . split ( '-' ) [ 0 ]
            if metric_type in aggregate_metric_classes :
                metrics [ 'aggregate_metrics' ] . append ( new_metric )
            else :
                metrics [ 'metrics' ] . append ( new_metric )
return metrics , run_steps , crossplots , report_args , graph_timezone , graphing_library


signature:
def parse ( self ) :
    """Parse the vmstat file
 : return : status of the metric parse"""

implementation:
file_status = True
for input_file in self . infile_list :
    file_status = file_status and naarad . utils . is_valid_file ( input_file )
    if not file_status :
        return False
status = True
cur_zone = None
cur_submetric = None
cur_value = None
data = { }
# stores the data of each column
for input_file in self . infile_list :
    logger . info ( 'Processing : %s' , input_file )
    timestamp_format = None
    with open ( input_file ) as fh :
        for line in fh :
            words = line . replace ( ',' , ' ' ) . split ( )
            # [0 ] is day ; [ 1 ] is seconds ; [ 2 . . . ] is field names : ;
            if len ( words ) < 3 :
                continue
            ts = words [ 0 ] + " " + words [ 1 ]
            if not timestamp_format or timestamp_format == 'unknown' :
                timestamp_format = naarad . utils . detect_timestamp_format ( ts )
            if timestamp_format == 'unknown' :
                continue
            ts = naarad . utils . get_standardized_timestamp ( ts , timestamp_format )
            if self . ts_out_of_range ( ts ) :
                continue
            if words [ 2 ] == 'Node' : # Node 0 zone DMA
                cols = words [ 2 : ]
                cur_zone = '.' . join ( cols )
                continue
            elif words [ 2 ] == 'pages' : # pages free 3936
                cur_submetric = words [ 2 ] + '.' + words [ 3 ]
                # pages . free
                cur_value = words [ 4 ]
            elif words [ 2 ] in self . processed_sub_metrics :
                cur_submetric = 'pages' + '.' + words [ 2 ]
                # pages . min
                cur_value = words [ 3 ]
            elif words [ 2 ] in self . skipped_sub_metrics :
                continue
            else : # other useful submetrics
                cur_submetric = words [ 2 ]
                cur_value = words [ 3 ]
            col = cur_zone + '.' + cur_submetric
            # prefix with ' Node . 0 . zone . DMA .
            # only process zones specified in config
            if cur_zone and self . zones and cur_zone not in self . zones :
                continue
            self . sub_metric_unit [ col ] = 'pages'
            # The unit of the sub metric . For / proc / zoneinfo , they are all in pages
            # only process sub _ metrics specified in config .
            if self . sub_metrics and cur_submetric and cur_submetric not in self . sub_metrics :
                continue
            if col in self . column_csv_map :
                out_csv = self . column_csv_map [ col ]
            else :
                out_csv = self . get_csv ( col )
                # column _ csv _ map [ ] is assigned in get _ csv ( )
                data [ out_csv ] = [ ]
            data [ out_csv ] . append ( ts + "," + cur_value )
# post processing , putting data in csv files ;
for csv in data . keys ( ) :
    self . csv_files . append ( csv )
    with open ( csv , 'w' ) as fh :
        fh . write ( '\n' . join ( sorted ( data [ csv ] ) ) )
return status


signature:
def collect ( self ) :
    """Take a list of metrics , filter all metrics based on hostname , and metric _ type
 For each metric , merge the corresponding csv files into one , update corresponding properties such as csv _ column _ map .
 Users can specify functions : raw , count ( qps ) , sum ( aggregated value ) , avg ( averaged value )
 The timestamp granularity of aggregated submetrics is in seconds ( sub - second is not supported )"""

implementation:
for aggr_metric in self . aggr_metrics : # e . g . , SAR - device . sda . await : count , sum , avg
    functions_aggr = [ ]
    fields = aggr_metric . split ( ":" )
    cur_metric_type = fields [ 0 ] . split ( "." ) [ 0 ]
    # e . g . SAR - device
    if len ( fields ) > 1 : # The user has to specify the aggregate functions ( i . e . , : raw , count , sum , avg )
        func_user = '' . join ( fields [ 1 ] . split ( ) )
        functions_aggr . extend ( func_user . split ( "," ) )
    else : # no user input of aggregate functions
        return True
    cur_column = '.' . join ( fields [ 0 ] . split ( '.' ) [ 1 : ] )
    # e . g . sda . await or all . percent - sys
    # Store data points of various aggregation functions
    aggr_data = { }
    aggr_data [ 'raw' ] = [ ]
    # Store all the raw values
    aggr_data [ 'sum' ] = defaultdict ( float )
    # Store the sum values for each timestamp
    aggr_data [ 'count' ] = defaultdict ( int )
    # Store the count of each timestamp ( i . e . qps )
    for metric in self . metrics : # Loop the list to find from all metrics to merge
        if metric . hostname in self . aggr_hosts and cur_column in metric . csv_column_map . values ( ) :
            file_csv = metric . get_csv ( cur_column )
            timestamp_format = None
            with open ( file_csv ) as fh :
                for line in fh :
                    aggr_data [ 'raw' ] . append ( line . rstrip ( ) )
                    words = line . split ( "," )
                    ts = words [ 0 ] . split ( '.' ) [ 0 ]
                    # In case of sub - seconds ; we only want the value of seconds ;
                    if not timestamp_format or timestamp_format == 'unknown' :
                        timestamp_format = naarad . utils . detect_timestamp_format ( ts )
                    if timestamp_format == 'unknown' :
                        continue
                    ts = naarad . utils . get_standardized_timestamp ( ts , timestamp_format )
                    aggr_data [ 'sum' ] [ ts ] += float ( words [ 1 ] )
                    aggr_data [ 'count' ] [ ts ] += 1
    # " raw " csv file
    if 'raw' in functions_aggr :
        out_csv = self . get_csv ( cur_column , 'raw' )
        self . csv_files . append ( out_csv )
        with open ( out_csv , 'w' ) as fh :
            fh . write ( "\n" . join ( sorted ( aggr_data [ 'raw' ] ) ) )
    # " sum " csv file
    if 'sum' in functions_aggr :
        out_csv = self . get_csv ( cur_column , 'sum' )
        self . csv_files . append ( out_csv )
        with open ( out_csv , 'w' ) as fh :
            for ( k , v ) in sorted ( aggr_data [ 'sum' ] . items ( ) ) :
                fh . write ( k + "," + str ( v ) + '\n' )
    # " avg " csv file
    if 'avg' in functions_aggr :
        out_csv = self . get_csv ( cur_column , 'avg' )
        self . csv_files . append ( out_csv )
        with open ( out_csv , 'w' ) as fh :
            for ( k , v ) in sorted ( aggr_data [ 'sum' ] . items ( ) ) :
                fh . write ( k + "," + str ( v / aggr_data [ 'count' ] [ k ] ) + '\n' )
    # " count " csv file ( qps )
    if 'count' in functions_aggr :
        out_csv = self . get_csv ( cur_column , 'count' )
        self . csv_files . append ( out_csv )
        with open ( out_csv , 'w' ) as fh :
            for ( k , v ) in sorted ( aggr_data [ 'count' ] . items ( ) ) :
                fh . write ( k + "," + str ( v ) + '\n' )
    gc . collect ( )
return True


signature:
def get_times ( self , native ) :
    """get start time stamp , launch time duration , and nus update time duration from JSON object native
 : param JSON OBJECT native
 : return : LONG event time stamp , LONG launch time , and LONG nus update time"""

implementation:
start_time = 0
end_time = 0
launch_time = 0
nus_update_time = 0
for item in native :
    if item [ CONSTANTS . LIA_TIMING_NAME ] == CONSTANTS . LIA_APP_ON_CREATE and item [ CONSTANTS . LIA_START ] is not None :
        start_time = item [ CONSTANTS . LIA_START ] [ CONSTANTS . LIA_LONG ]
    if item [ CONSTANTS . LIA_TIMING_NAME ] == CONSTANTS . LIA_NUS_UPDATE :
        if item [ CONSTANTS . LIA_TIMING_VALUE ] is not None :
            nus_update_time = item [ CONSTANTS . LIA_TIMING_VALUE ] [ CONSTANTS . LIA_LONG ]
        if item [ CONSTANTS . LIA_START ] is not None :
            end_time = item [ CONSTANTS . LIA_START ] [ CONSTANTS . LIA_LONG ]
if start_time == 0 or end_time == 0 :
    time_stamp = 0
    launch_time = 0
else :
    time_stamp = start_time
    launch_time = end_time - start_time
return ( time_stamp , launch_time , nus_update_time )


signature:
def run ( self ) :
    """Perform the Oct2Py speed analysis .
 Uses timeit to test the raw execution of an Octave command ,
 Then tests progressively larger array passing ."""

implementation:
print ( 'Oct2Py speed test' )
print ( '*' * 20 )
time . sleep ( 1 )
print ( 'Raw speed: ' )
avg = timeit . timeit ( self . raw_speed , number = 10 ) / 10
print ( '    {0:0.01f} usec per loop' . format ( avg * 1e6 ) )
sides = [ 1 , 10 , 100 , 1000 ]
runs = [ 10 , 10 , 10 , 5 ]
for ( side , nruns ) in zip ( sides , runs ) :
    self . array = np . reshape ( np . arange ( side ** 2 ) , ( - 1 ) )
    print ( 'Put {0}x{1}: ' . format ( side , side ) )
    avg = timeit . timeit ( self . large_array_put , number = nruns ) / nruns
    print ( '    {0:0.01f} msec' . format ( avg * 1e3 ) )
    print ( 'Get {0}x{1}: ' . format ( side , side ) )
    avg = timeit . timeit ( self . large_array_get , number = nruns ) / nruns
    print ( '    {0:0.01f} msec' . format ( avg * 1e3 ) )
self . octave . exit ( )
print ( '*' * 20 )
print ( 'Test complete!' )


signature:
def exit ( self ) :
    """Quits this octave session and cleans up ."""

implementation:
if self . _engine :
    self . _engine . repl . terminate ( )
self . _engine = None


signature:
def push ( self , name , var , timeout = None , verbose = True ) :
    """Put a variable or variables into the Octave session .
 Parameters
 name : str or list
 Name of the variable ( s ) .
 var : object or list
 The value ( s ) to pass .
 timeout : float
 Time to wait for response from Octave ( per line ) .
 * * kwargs : Deprecated kwargs , ignored .
 Examples
 > > > from oct2py import octave
 > > > y = [ 1 , 2]
 > > > octave . push ( ' y ' , y )
 > > > octave . pull ( ' y ' )
 array ( [ [ 1 . , 2 . ] ] )
 > > > octave . push ( [ ' x ' , ' y ' ] , [ ' spam ' , [ 1 , 2 , 3 , 4 ] ] )
 > > > octave . pull ( [ ' x ' , ' y ' ] ) # doctest : + SKIP
 [ u ' spam ' , array ( [ [ 1 , 2 , 3 , 4 ] ] ) ]
 Notes
 Integer type arguments will be converted to floating point
 unless ` convert _ to _ float = False ` ."""

implementation:
if isinstance ( name , ( str , unicode ) ) :
    name = [ name ]
    var = [ var ]
for ( n , v ) in zip ( name , var ) :
    self . feval ( 'assignin' , 'base' , n , v , nout = 0 , timeout = timeout , verbose = verbose )


signature:
def pull ( self , var , timeout = None , verbose = True ) :
    """Retrieve a value or values from the Octave session .
 Parameters
 var : str or list
 Name of the variable ( s ) to retrieve .
 timeout : float , optional .
 Time to wait for response from Octave ( per line ) .
 * * kwargs : Deprecated kwargs , ignored .
 Returns
 out : object
 Object returned by Octave .
 Raises
 Oct2PyError
 If the variable does not exist in the Octave session .
 Examples
 > > > from oct2py import octave
 > > > y = [ 1 , 2]
 > > > octave . push ( ' y ' , y )
 > > > octave . pull ( ' y ' )
 array ( [ [ 1 . , 2 . ] ] )
 > > > octave . push ( [ ' x ' , ' y ' ] , [ ' spam ' , [ 1 , 2 , 3 , 4 ] ] )
 > > > octave . pull ( [ ' x ' , ' y ' ] ) # doctest : + SKIP
 [ u ' spam ' , array ( [ [ 1 , 2 , 3 , 4 ] ] ) ]"""

implementation:
if isinstance ( var , ( str , unicode ) ) :
    var = [ var ]
outputs = [ ]
for name in var :
    exist = self . _exist ( name )
    if exist == 1 :
        outputs . append ( self . feval ( 'evalin' , 'base' , name , timeout = timeout , verbose = verbose ) )
    else :
        outputs . append ( self . get_pointer ( name , timeout = timeout ) )
if len ( outputs ) == 1 :
    return outputs [ 0 ]
return outputs


signature:
def get_pointer ( self , name , timeout = None ) :
    """Get a pointer to a named object in the Octave workspace .
 Parameters
 name : str
 The name of the object in the Octave workspace .
 timemout : float , optional .
 Time to wait for response from Octave ( per line ) .
 Examples
 > > > from oct2py import octave
 > > > octave . eval ( ' foo = [ 1 , 2 ] ; ' )
 > > > ptr = octave . get _ pointer ( ' foo ' )
 > > > ptr . value
 array ( [ [ 1 . , 2 . ] ] )
 > > > ptr . address
 ' foo '
 > > > # Can be passed as an argument
 > > > octave . disp ( ptr ) # doctest : + SKIP
 1 2
 > > > from oct2py import octave
 > > > sin = octave . get _ pointer ( ' sin ' ) # equivalent to ` octave . sin `
 > > > sin . address
 ' @ sin '
 > > > x = octave . quad ( sin , 0 , octave . pi ( ) )
 2.0
 Notes
 Pointers can be passed to ` feval ` or dynamic functions as function arguments . A pointer passed as a nested value will be passed by value instead .
 Raises
 Oct2PyError
 If the variable does not exist in the Octave session or is of
 unknown type .
 Returns
 A variable , object , user class , or function pointer as appropriate ."""

implementation:
exist = self . _exist ( name )
isobject = self . _isobject ( name , exist )
if exist == 0 :
    raise Oct2PyError ( '"%s" is undefined' % name )
elif exist == 1 :
    return _make_variable_ptr_instance ( self , name )
elif isobject :
    return self . _get_user_class ( name )
elif exist in [ 2 , 3 , 5 ] :
    return self . _get_function_ptr ( name )
raise Oct2PyError ( 'Unknown type for object "%s"' % name )


signature:
def extract_figures ( self , plot_dir , remove = False ) :
    """Extract the figures in the directory to IPython display objects .
 Parameters
 plot _ dir : str
 The plot dir where the figures were created .
 remove : bool , optional .
 Whether to remove the plot directory after saving ."""

implementation:
figures = self . _engine . extract_figures ( plot_dir , remove )
return figures


signature:
def feval ( self , func_path , * func_args , ** kwargs ) :
    """Run a function in Octave and return the result .
 Parameters
 func _ path : str
 Name of function to run or a path to an m - file .
 func _ args : object , optional
 Args to send to the function .
 nout : int , optional
 Desired number of return arguments , defaults to 1.
 store _ as : str , optional
 If given , saves the result to the given Octave variable name
 instead of returning it .
 verbose : bool , optional
 Log Octave output at INFO level . If False , log at DEBUG level .
 stream _ handler : callable , optional
 A function that is called for each line of output from the
 evaluation .
 timeout : float , optional
 The timeout in seconds for the call .
 plot _ dir : str , optional
 If specificed , save the session ' s plot figures to the plot
 directory instead of displaying the plot window .
 plot _ name : str , optional
 Saved plots will start with ` plot _ name ` and
 end with " _ % % . xxx ' where % % is the plot number and
 xxx is the ` plot _ format ` .
 plot _ format : str , optional
 The format in which to save the plot .
 plot _ width : int , optional
 The plot with in pixels .
 plot _ height : int , optional
 The plot height in pixels .
 Notes
 The function arguments passed follow Octave calling convention , not
 Python . That is , all values must be passed as a comma separated list ,
 not using ` x = foo ` assignment .
 Examples
 > > > from oct2py import octave
 > > > cell = octave . feval ( ' cell ' , 10 , 10 , 10)
 > > > cell . shape
 (10 , 10 , 10)
 > > > from oct2py import octave
 > > > x = octave . feval ( ' linspace ' , 0 , octave . pi ( ) / 2)
 > > > x . shape
 (1 , 100)
 > > > from oct2py import octave
 > > > x = octave . feval ( ' svd ' , octave . hilb ( 3 ) )
 array ( [ [ 1.40831893 ] ,
 [ 0.12232707 ] ,
 [ 0.00268734 ] ] )
 > > > # specify three return values
 > > > ( u , v , d ) = octave . feval ( ' svd ' , octave . hilb ( 3 ) , nout = 3)
 > > > u . shape
 (3 , 3)
 Returns
 The Python value ( s ) returned by the Octave function call ."""

implementation:
if not self . _engine :
    raise Oct2PyError ( 'Session is not open' )
nout = kwargs . get ( 'nout' , None )
if nout is None :
    nout = 1
plot_dir = kwargs . get ( 'plot_dir' )
settings = dict ( backend = 'inline' if plot_dir else self . backend , format = kwargs . get ( 'plot_format' ) , name = kwargs . get ( 'plot_name' ) , width = kwargs . get ( 'plot_width' ) , height = kwargs . get ( 'plot_height' ) , resolution = kwargs . get ( 'plot_res' ) )
self . _engine . plot_settings = settings
dname = osp . dirname ( func_path )
fname = osp . basename ( func_path )
func_name , ext = osp . splitext ( fname )
if ext and not ext == '.m' :
    raise TypeError ( 'Need to give path to .m file' )
if func_name == 'clear' :
    raise Oct2PyError ( 'Cannot use `clear` command directly, use' + ' eval("clear(var1, var2)")' )
stream_handler = kwargs . get ( 'stream_handler' )
verbose = kwargs . get ( 'verbose' , True )
store_as = kwargs . get ( 'store_as' , '' )
timeout = kwargs . get ( 'timeout' , self . timeout )
if not stream_handler :
    stream_handler = self . logger . info if verbose else self . logger . debug
return self . _feval ( func_name , func_args , dname = dname , nout = nout , timeout = timeout , stream_handler = stream_handler , store_as = store_as , plot_dir = plot_dir )


signature:
def eval ( self , cmds , verbose = True , timeout = None , stream_handler = None , temp_dir = None , plot_dir = None , plot_name = 'plot' , plot_format = 'svg' , plot_width = None , plot_height = None , plot_res = None , nout = 0 , ** kwargs ) :
    """Evaluate an Octave command or commands .
 Parameters
 cmds : str or list
 Commands ( s ) to pass to Octave .
 verbose : bool , optional
 Log Octave output at INFO level . If False , log at DEBUG level .
 stream _ handler : callable , optional
 A function that is called for each line of output from the
 evaluation .
 timeout : float , optional
 Time to wait for response from Octave ( per line ) . If not given ,
 the instance ` timeout ` is used .
 nout : int , optional .
 The desired number of returned values , defaults to 0 . If nout
 is 0 , the ` ans ` will be returned as the return value .
 temp _ dir : str , optional
 If specified , the session ' s MAT files will be created in the
 directory , otherwise a the instance ` temp _ dir ` is used .
 a shared memory ( tmpfs ) path .
 plot _ dir : str , optional
 If specificed , save the session ' s plot figures to the plot
 directory instead of displaying the plot window .
 plot _ name : str , optional
 Saved plots will start with ` plot _ name ` and
 end with " _ % % . xxx ' where % % is the plot number and
 xxx is the ` plot _ format ` .
 plot _ format : str , optional
 The format in which to save the plot ( PNG by default ) .
 plot _ width : int , optional
 The plot with in pixels .
 plot _ height : int , optional
 The plot height in pixels .
 plot _ res : int , optional
 The plot resolution in pixels per inch .
 * * kwargs Deprectated kwargs .
 Examples
 > > > from oct2py import octave
 > > > octave . eval ( ' disp ( " hello " ) ' ) # doctest : + SKIP
 hello
 > > > x = octave . eval ( ' round ( quad ( @ sin , 0 , pi / 2 ) ) ; ' )
 1.0
 > > > a = octave . eval ( ' disp ( " hello " ) ; 1 ; ' ) # doctest : + SKIP
 hello
 > > > a = octave . eval ( ' disp ( " hello " ) ; 1 ; ' , verbose = False )
 1.0
 > > > from oct2py import octave
 > > > lines = [ ]
 > > > octave . eval ( ' for i = 1:3 ; disp ( i ) ; end ' , \
 stream _ handler = lines . append )
 > > > lines # doctest : + SKIP
 [ ' 1 ' , ' 2 ' , ' 3 ' ]
 Returns
 out : object
 Octave " ans " variable , or None .
 Notes
 The deprecated ` log ` kwarg will temporarily set the ` logger ` level to
 ` WARN ` . Using the ` logger ` settings directly is preferred .
 The deprecated ` return _ both ` kwarg will still work , but the preferred
 method is to use the ` stream _ handler ` . If ` stream _ handler ` is given ,
 the ` return _ both ` kwarg will be honored but will give an empty string
 as the reponse .
 Raises
 Oct2PyError
 If the command ( s ) fail ."""

implementation:
if isinstance ( cmds , ( str , unicode ) ) :
    cmds = [ cmds ]
prev_temp_dir = self . temp_dir
self . temp_dir = temp_dir or self . temp_dir
prev_log_level = self . logger . level
if kwargs . get ( 'log' ) is False :
    self . logger . setLevel ( logging . WARN )
for name in [ 'log' , 'return_both' ] :
    if name not in kwargs :
        continue
    msg = 'Using deprecated `%s` kwarg, see docs on `Oct2Py.eval()`'
    warnings . warn ( msg % name , stacklevel = 2 )
return_both = kwargs . pop ( 'return_both' , False )
lines = [ ]
if return_both and not stream_handler :
    stream_handler = lines . append
ans = None
for cmd in cmds :
    resp = self . feval ( 'evalin' , 'base' , cmd , nout = nout , timeout = timeout , stream_handler = stream_handler , verbose = verbose , plot_dir = plot_dir , plot_name = plot_name , plot_format = plot_format , plot_width = plot_width , plot_height = plot_height , plot_res = plot_res )
    if resp is not None :
        ans = resp
self . temp_dir = prev_temp_dir
self . logger . setLevel ( prev_log_level )
if return_both :
    return '\n' . join ( lines ) , ans
return ans


signature:
def restart ( self ) :
    """Restart an Octave session in a clean state"""

implementation:
if self . _engine :
    self . _engine . repl . terminate ( )
executable = self . _executable
if executable :
    os . environ [ 'OCTAVE_EXECUTABLE' ] = executable
if 'OCTAVE_EXECUTABLE' not in os . environ and 'OCTAVE' in os . environ :
    os . environ [ 'OCTAVE_EXECUTABLE' ] = os . environ [ 'OCTAVE' ]
self . _engine = OctaveEngine ( stdin_handler = self . _handle_stdin , logger = self . logger )
# Add local Octave scripts .
self . _engine . eval ( 'addpath("%s");' % HERE . replace ( osp . sep , '/' ) )


signature:
def _feval ( self , func_name , func_args = ( ) , dname = '' , nout = 0 , timeout = None , stream_handler = None , store_as = '' , plot_dir = None ) :
    """Run the given function with the given args ."""

implementation:
engine = self . _engine
if engine is None :
    raise Oct2PyError ( 'Session is closed' )
# Set up our mat file paths .
out_file = osp . join ( self . temp_dir , 'writer.mat' )
out_file = out_file . replace ( osp . sep , '/' )
in_file = osp . join ( self . temp_dir , 'reader.mat' )
in_file = in_file . replace ( osp . sep , '/' )
func_args = list ( func_args )
ref_indices = [ ]
for ( i , value ) in enumerate ( func_args ) :
    if isinstance ( value , OctavePtr ) :
        ref_indices . append ( i + 1 )
        func_args [ i ] = value . address
ref_indices = np . array ( ref_indices )
# Save the request data to the output file .
req = dict ( func_name = func_name , func_args = tuple ( func_args ) , dname = dname or '' , nout = nout , store_as = store_as or '' , ref_indices = ref_indices )
write_file ( req , out_file , oned_as = self . _oned_as , convert_to_float = self . convert_to_float )
# Set up the engine and evaluate the ` _ pyeval ( ) ` function .
engine . stream_handler = stream_handler or self . logger . info
if timeout is None :
    timeout = self . timeout
try :
    engine . eval ( '_pyeval("%s", "%s");' % ( out_file , in_file ) , timeout = timeout )
except KeyboardInterrupt as e :
    stream_handler ( engine . repl . interrupt ( ) )
    raise
except TIMEOUT :
    stream_handler ( engine . repl . interrupt ( ) )
    raise Oct2PyError ( 'Timed out, interrupting' )
except EOF :
    stream_handler ( engine . repl . child . before )
    self . restart ( )
    raise Oct2PyError ( 'Session died, restarting' )
# Read in the output .
resp = read_file ( in_file , self )
if resp [ 'err' ] :
    msg = self . _parse_error ( resp [ 'err' ] )
    raise Oct2PyError ( msg )
result = resp [ 'result' ] . ravel ( ) . tolist ( )
if isinstance ( result , list ) and len ( result ) == 1 :
    result = result [ 0 ]
# Check for sentinel value .
if ( isinstance ( result , Cell ) and result . size == 1 and isinstance ( result [ 0 ] , string_types ) and result [ 0 ] == '__no_value__' ) :
    result = None
if plot_dir :
    self . _engine . make_figures ( plot_dir )
return result


signature:
def _parse_error ( self , err ) :
    """Create a traceback for an Octave evaluation error ."""

implementation:
self . logger . debug ( err )
stack = err . get ( 'stack' , [ ] )
if not err [ 'message' ] . startswith ( 'parse error:' ) :
    err [ 'message' ] = 'error: ' + err [ 'message' ]
errmsg = 'Octave evaluation error:\n%s' % err [ 'message' ]
if not isinstance ( stack , StructArray ) :
    return errmsg
errmsg += '\nerror: called from:'
for item in stack [ : - 1 ] :
    errmsg += '\n    %(name)s at line %(line)d' % item
    try :
        errmsg += ', column %(column)d' % item
    except Exception :
        pass
return errmsg


signature:
def _get_doc ( self , name ) :
    """Get the documentation of an Octave procedure or object .
 Parameters
 name : str
 Function name to search for .
 Returns
 out : str
 Documentation string .
 Raises
 Oct2PyError
 If the procedure or object function has a syntax error ."""

implementation:
doc = 'No documentation for %s' % name
engine = self . _engine
doc = engine . eval ( 'help("%s")' % name , silent = True )
if 'syntax error:' in doc . lower ( ) :
    raise Oct2PyError ( doc )
if 'error:' in doc . lower ( ) :
    doc = engine . eval ( 'type("%s")' % name , silent = True )
    doc = '\n' . join ( doc . splitlines ( ) [ : 3 ] )
default = self . feval . __doc__
default = '        ' + default [ default . find ( 'func_args:' ) : ]
default = '\n' . join ( [ line [ 8 : ] for line in default . splitlines ( ) ] )
doc = '\n' . join ( doc . splitlines ( ) )
doc = '\n' + doc + '\n\nParameters\n----------\n' + default
doc += '\n**kwargs - Deprecated keyword arguments\n\n'
doc += 'Notes\n-----\n'
doc += 'Keyword arguments to dynamic functions are deprecated.\n'
doc += 'The `plot_*` kwargs will be ignored, but the rest will\n'
doc += 'used as key - value pairs as in version 3.x.\n'
doc += 'Use `set_plot_settings()` for plot settings, and use\n'
doc += '`func_args` directly for key - value pairs.'
return doc


signature:
def _exist ( self , name ) :
    """Test whether a name exists and return the name code .
 Raises an error when the name does not exist ."""

implementation:
cmd = 'exist("%s")' % name
resp = self . _engine . eval ( cmd , silent = True ) . strip ( )
exist = int ( resp . split ( ) [ - 1 ] )
if exist == 0 :
    msg = 'Value "%s" does not exist in Octave workspace'
    raise Oct2PyError ( msg % name )
return exist


signature:
def _isobject ( self , name , exist ) :
    """Test whether the name is an object ."""

implementation:
if exist in [ 2 , 5 ] :
    return False
cmd = 'isobject(%s)' % name
resp = self . _engine . eval ( cmd , silent = True ) . strip ( )
return resp == 'ans =  1'


signature:
def _get_function_ptr ( self , name ) :
    """Get or create a function pointer of the given name ."""

implementation:
func = _make_function_ptr_instance
self . _function_ptrs . setdefault ( name , func ( self , name ) )
return self . _function_ptrs [ name ]


signature:
def _get_user_class ( self , name ) :
    """Get or create a user class of the given type ."""

implementation:
self . _user_classes . setdefault ( name , _make_user_class ( self , name ) )
return self . _user_classes [ name ]


signature:
def _cleanup ( self ) :
    """Clean up resources used by the session ."""

implementation:
self . exit ( )
workspace = osp . join ( os . getcwd ( ) , 'octave-workspace' )
if osp . exists ( workspace ) :
    os . remove ( workspace )


signature:
def demo ( delay = 1 , interactive = True ) :
    """Play a demo script showing most of the oct2py api features .
 Parameters
 delay : float
 Time between each command in seconds ."""

implementation:
script = """
    #########################
    # Oct2Py demo
    #########################
    import numpy as np
    from oct2py import Oct2Py
    oc = Oct2Py()
    # basic commands
    print(oc.abs(-1))
    print(oc.upper('xyz'))
    # plotting
    oc.plot([1,2,3],'-o', 'linewidth', 2)
    raw_input('Press Enter to continue...')
    oc.close()
    xx = np.arange(-2*np.pi, 2*np.pi, 0.2)
    oc.surf(np.subtract.outer(np.sin(xx), np.cos(xx)))
    raw_input('Press Enter to continue...')
    oc.close()
    # getting help
    help(oc.svd)
    # single vs. multiple return values
    print(oc.svd(np.array([[1,2], [1,3]])))
    U, S, V = oc.svd([[1,2], [1,3]], nout=3)
    print(U, S, V)
    # low level constructs
    oc.eval("y=ones(3,3)")
    print(oc.pull("y"))
    oc.eval("x=zeros(3,3)", verbose=True)
    t = oc.eval('rand(1, 2)', verbose=True)
    y = np.zeros((3,3))
    oc.push('y', y)
    print(oc.pull('y'))
    from oct2py import Struct
    y = Struct()
    y.b = 'spam'
    y.c.d = 'eggs'
    print(y.c['d'])
    print(y)
    #########################
    # Demo Complete!
    #########################
    """
if not PY2 :
    script = script . replace ( 'raw_input' , 'input' )
for line in script . strip ( ) . split ( '\n' ) :
    line = line . strip ( )
    if not 'input(' in line :
        time . sleep ( delay )
        print ( ">>> {0}" . format ( line ) )
        time . sleep ( delay )
    if not interactive :
        if 'plot' in line or 'surf' in line or 'input(' in line :
            line = 'print()'
    exec ( line )


signature:
def kill_octave ( ) :
    """Kill all octave instances ( cross - platform ) .
 This will restart the " octave " instance . If you have instantiated
 Any other Oct2Py objects , you must restart them ."""

implementation:
import os
if os . name == 'nt' :
    os . system ( 'taskkill /im octave /f' )
else :
    os . system ( 'killall -9 octave' )
    os . system ( 'killall -9 octave-cli' )
octave . restart ( )


signature:
def thread_check ( nthreads = 3 ) :
    """Start a number of threads and verify each has a unique Octave session .
 Parameters
 nthreads : int
 Number of threads to use .
 Raises
 Oct2PyError
 If the thread does not sucessfully demonstrate independence ."""

implementation:
print ( "Starting {0} threads at {1}" . format ( nthreads , datetime . datetime . now ( ) ) )
threads = [ ]
for i in range ( nthreads ) :
    thread = ThreadClass ( )
    thread . setDaemon ( True )
    thread . start ( )
    threads . append ( thread )
for thread in threads :
    thread . join ( )
print ( 'All threads closed at {0}' . format ( datetime . datetime . now ( ) ) )


signature:
def run ( self ) :
    """Create a unique instance of Octave and verify namespace uniqueness .
 Raises
 Oct2PyError
 If the thread does not sucessfully demonstrate independence"""

implementation:
octave = Oct2Py ( )
# write the same variable name in each thread and read it back
octave . push ( 'name' , self . getName ( ) )
name = octave . pull ( 'name' )
now = datetime . datetime . now ( )
print ( "{0} got '{1}' at {2}" . format ( self . getName ( ) , name , now ) )
octave . exit ( )
try :
    assert self . getName ( ) == name
except AssertionError : # pragma : no cover
    raise Oct2PyError ( 'Thread collision detected' )
return


signature:
def read_file ( path , session = None ) :
    """Read the data from the given file path ."""

implementation:
try :
    data = loadmat ( path , struct_as_record = True )
except UnicodeDecodeError as e :
    raise Oct2PyError ( str ( e ) )
out = dict ( )
for ( key , value ) in data . items ( ) :
    out [ key ] = _extract ( value , session )
return out


signature:
def write_file ( obj , path , oned_as = 'row' , convert_to_float = True ) :
    """Save a Python object to an Octave file on the given path ."""

implementation:
data = _encode ( obj , convert_to_float )
try : # scipy . io . savemat is not thread - save .
# See https : / / github . com / scipy / scipy / issues / 7260
    with _WRITE_LOCK :
        savemat ( path , data , appendmat = False , oned_as = oned_as , long_field_names = True )
except KeyError : # pragma : no cover
    raise Exception ( 'could not save mat file' )


signature:
def _extract ( data , session = None ) :
    """Convert the Octave values to values suitable for Python ."""

implementation:
# Extract each item of a list .
if isinstance ( data , list ) :
    return [ _extract ( d , session ) for d in data ]
# Ignore leaf objects .
if not isinstance ( data , np . ndarray ) :
    return data
# Extract user defined classes .
if isinstance ( data , MatlabObject ) :
    cls = session . _get_user_class ( data . classname )
    return cls . from_value ( data )
# Extract struct data .
if data . dtype . names : # Singular struct
    if data . size == 1 :
        return _create_struct ( data , session )
    # Struct array
    return StructArray ( data , session )
# Extract cells .
if data . dtype . kind == 'O' :
    return Cell ( data , session )
# Compress singleton values .
if data . size == 1 :
    return data . item ( )
# Compress empty values .
if data . size == 0 :
    if data . dtype . kind in 'US' :
        return ''
    return [ ]
# Return standard array .
return data


signature:
def _create_struct ( data , session ) :
    """Create a struct from session data ."""

implementation:
out = Struct ( )
for name in data . dtype . names :
    item = data [ name ]
    # Extract values that are cells ( they are doubly wrapped ) .
    if isinstance ( item , np . ndarray ) and item . dtype . kind == 'O' :
        item = item . squeeze ( ) . tolist ( )
    out [ name ] = _extract ( item , session )
return out


signature:
def _encode ( data , convert_to_float ) :
    """Convert the Python values to values suitable to send to Octave ."""

implementation:
ctf = convert_to_float
# Handle variable pointer .
if isinstance ( data , ( OctaveVariablePtr ) ) :
    return _encode ( data . value , ctf )
# Handle a user defined object .
if isinstance ( data , OctaveUserClass ) :
    return _encode ( OctaveUserClass . to_value ( data ) , ctf )
# Handle a function pointer .
if isinstance ( data , ( OctaveFunctionPtr , MatlabFunction ) ) :
    raise Oct2PyError ( 'Cannot write Octave functions' )
# Handle matlab objects .
if isinstance ( data , MatlabObject ) :
    view = data . view ( np . ndarray )
    out = MatlabObject ( data , data . classname )
    for name in out . dtype . names :
        out [ name ] = _encode ( view [ name ] , ctf )
    return out
# Handle pandas series and dataframes
if isinstance ( data , ( DataFrame , Series ) ) :
    return _encode ( data . values , ctf )
# Extract and encode values from dict - like objects .
if isinstance ( data , dict ) :
    out = dict ( )
    for ( key , value ) in data . items ( ) :
        out [ key ] = _encode ( value , ctf )
    return out
# Send None as nan .
if data is None :
    return np . NaN
# Sets are treated like lists .
if isinstance ( data , set ) :
    return _encode ( list ( data ) , ctf )
# Lists can be interpreted as numeric arrays or cell arrays .
if isinstance ( data , list ) :
    if _is_simple_numeric ( data ) :
        return _encode ( np . array ( data ) , ctf )
    return _encode ( tuple ( data ) , ctf )
# Tuples are handled as cells .
if isinstance ( data , tuple ) :
    obj = np . empty ( len ( data ) , dtype = object )
    for ( i , item ) in enumerate ( data ) :
        obj [ i ] = _encode ( item , ctf )
    return obj
# Sparse data must be floating type .
if isinstance ( data , spmatrix ) :
    return data . astype ( np . float64 )
# Return other data types unchanged .
if not isinstance ( data , np . ndarray ) :
    return data
# Extract and encode data from object - like arrays .
if data . dtype . kind in 'OV' :
    out = np . empty ( data . size , dtype = data . dtype )
    for ( i , item ) in enumerate ( data . ravel ( ) ) :
        if data . dtype . names :
            for name in data . dtype . names :
                out [ i ] [ name ] = _encode ( item [ name ] , ctf )
        else :
            out [ i ] = _encode ( item , ctf )
    return out . reshape ( data . shape )
# Complex 128 is the highest supported by savemat .
if data . dtype . name == 'complex256' :
    return data . astype ( np . complex128 )
# Convert to float if applicable .
if ctf and data . dtype . kind in 'ui' :
    return data . astype ( np . float64 )
# Return standard array .
return data


signature:
def _is_simple_numeric ( data ) :
    """Test if a list contains simple numeric data ."""

implementation:
for item in data :
    if isinstance ( item , set ) :
        item = list ( item )
    if isinstance ( item , list ) :
        if not _is_simple_numeric ( item ) :
            return False
    elif not isinstance ( item , ( int , float , complex ) ) :
        return False
return True


signature:
def get_log ( name = None ) :
    """Return a console logger .
 Output may be sent to the logger using the ` debug ` , ` info ` , ` warning ` ,
 ` error ` and ` critical ` methods .
 Parameters
 name : str
 Name of the log .
 References
 . . [ 1 ] Logging facility for Python ,
 http : / / docs . python . org / library / logging . html"""

implementation:
if name is None :
    name = 'oct2py'
else :
    name = 'oct2py.' + name
log = logging . getLogger ( name )
log . setLevel ( logging . INFO )
return log


signature:
def _setup_log ( ) :
    """Configure root logger ."""

implementation:
try :
    handler = logging . StreamHandler ( stream = sys . stdout )
except TypeError : # pragma : no cover
    handler = logging . StreamHandler ( strm = sys . stdout )
log = get_log ( )
log . addHandler ( handler )
log . setLevel ( logging . INFO )
log . propagate = False


signature:
def _make_user_class ( session , name ) :
    """Make an Octave class for a given class name"""

implementation:
attrs = session . eval ( 'fieldnames(%s);' % name , nout = 1 ) . ravel ( ) . tolist ( )
methods = session . eval ( 'methods(%s);' % name , nout = 1 ) . ravel ( ) . tolist ( )
ref = weakref . ref ( session )
doc = _DocDescriptor ( ref , name )
values = dict ( __doc__ = doc , _name = name , _ref = ref , _attrs = attrs , __module__ = 'oct2py.dynamic' )
for method in methods :
    doc = _MethodDocDescriptor ( ref , name , method )
    cls_name = '%s_%s' % ( name , method )
    method_values = dict ( __doc__ = doc )
    method_cls = type ( str ( cls_name ) , ( OctaveUserClassMethod , ) , method_values )
    values [ method ] = method_cls ( ref , method , name )
for attr in attrs :
    values [ attr ] = OctaveUserClassAttr ( ref , attr , attr )
return type ( str ( name ) , ( OctaveUserClass , ) , values )


signature:
def from_value ( cls , value ) :
    """This is how an instance is created when we read a
 MatlabObject from a MAT file ."""

implementation:
instance = OctaveUserClass . __new__ ( cls )
instance . _address = '%s_%s' % ( instance . _name , id ( instance ) )
instance . _ref ( ) . push ( instance . _address , value )
return instance


signature:
def to_value ( cls , instance ) :
    """Convert to a value to send to Octave ."""

implementation:
if not isinstance ( instance , OctaveUserClass ) or not instance . _attrs :
    return dict ( )
# Bootstrap a MatlabObject from scipy . io
# From https : / / github . com / scipy / scipy / blob / 93a0ea9e5d4aba1f661b6bb0e18f9c2d1fce436a / scipy / io / matlab / mio5 . py # L435 - L443
# and https : / / github . com / scipy / scipy / blob / 93a0ea9e5d4aba1f661b6bb0e18f9c2d1fce436a / scipy / io / matlab / mio5 _ params . py # L224
dtype = [ ]
values = [ ]
for attr in instance . _attrs :
    dtype . append ( ( str ( attr ) , object ) )
    values . append ( getattr ( instance , attr ) )
struct = np . array ( [ tuple ( values ) ] , dtype )
return MatlabObject ( struct , instance . _name )


signature:
def to_pointer ( cls , instance ) :
    """Get a pointer to the private object ."""

implementation: return OctavePtr ( instance . _ref , instance . _name , instance . _address )


signature:
def document_func_view ( serializer_class = None , response_serializer_class = None , filter_backends = None , permission_classes = None , authentication_classes = None , doc_format_args = list ( ) , doc_format_kwargs = dict ( ) ) :
    """Decorator to make functional view documentable via drf - autodocs"""

implementation:
def decorator ( func ) :
    if serializer_class :
        func . cls . serializer_class = func . view_class . serializer_class = serializer_class
    if response_serializer_class :
        func . cls . response_serializer_class = func . view_class . response_serializer_class = response_serializer_class
    if filter_backends :
        func . cls . filter_backends = func . view_class . filter_backends = filter_backends
    if permission_classes :
        func . cls . permission_classes = func . view_class . permission_classes = permission_classes
    if authentication_classes :
        func . cls . authentication_classes = func . view_class . authentication_classes = authentication_classes
    if doc_format_args or doc_format_kwargs :
        func . cls . __doc__ = func . view_class . __doc__ = getdoc ( func ) . format ( * doc_format_args , ** doc_format_kwargs )
    return func
return decorator


signature:
def format_docstring ( * args , ** kwargs ) :
    """Decorator for clean docstring formatting"""

implementation:
def decorator ( func ) :
    func . __doc__ = getdoc ( func ) . format ( * args , ** kwargs )
    return func
return decorator


signature:
def is_rarfile ( filename ) :
    """Return true if file is a valid RAR file ."""

implementation:
mode = constants . RAR_OM_LIST_INCSPLIT
archive = unrarlib . RAROpenArchiveDataEx ( filename , mode = mode )
try :
    handle = unrarlib . RAROpenArchiveEx ( ctypes . byref ( archive ) )
except unrarlib . UnrarException :
    return False
unrarlib . RARCloseArchive ( handle )
return ( archive . OpenResult == constants . SUCCESS )


signature:
def _read_header ( self , handle ) :
    """Read current member header into a RarInfo object ."""

implementation:
header_data = unrarlib . RARHeaderDataEx ( )
try :
    res = unrarlib . RARReadHeaderEx ( handle , ctypes . byref ( header_data ) )
    rarinfo = RarInfo ( header = header_data )
except unrarlib . ArchiveEnd :
    return None
except unrarlib . MissingPassword :
    raise RuntimeError ( "Archive is encrypted, password required" )
except unrarlib . BadPassword :
    raise RuntimeError ( "Bad password for Archive" )
except unrarlib . UnrarException as e :
    raise BadRarFile ( str ( e ) )
return rarinfo


signature:
def _process_current ( self , handle , op , dest_path = None , dest_name = None ) :
    """Process current member with ' op ' operation ."""

implementation: unrarlib . RARProcessFileW ( handle , op , dest_path , dest_name )


signature:
def _load_metadata ( self , handle ) :
    """Load archive members metadata ."""

implementation:
rarinfo = self . _read_header ( handle )
while rarinfo :
    self . filelist . append ( rarinfo )
    self . NameToInfo [ rarinfo . filename ] = rarinfo
    self . _process_current ( handle , constants . RAR_SKIP )
    rarinfo = self . _read_header ( handle )


signature:
def _open ( self , archive ) :
    """Open RAR archive file ."""

implementation:
try :
    handle = unrarlib . RAROpenArchiveEx ( ctypes . byref ( archive ) )
except unrarlib . UnrarException :
    raise BadRarFile ( "Invalid RAR file." )
return handle


signature:
def open ( self , member , pwd = None ) :
    """Return file - like object for ' member ' .
 ' member ' may be a filename or a RarInfo object ."""

implementation:
if isinstance ( member , RarInfo ) :
    member = member . filename
archive = unrarlib . RAROpenArchiveDataEx ( self . filename , mode = constants . RAR_OM_EXTRACT )
handle = self . _open ( archive )
password = pwd or self . pwd
if password is not None :
    unrarlib . RARSetPassword ( handle , b ( password ) )
# based on BrutuZ ( https : / / github . com / matiasb / python - unrar / pull / 4)
# and Cubixmeister work
data = _ReadIntoMemory ( )
c_callback = unrarlib . UNRARCALLBACK ( data . _callback )
unrarlib . RARSetCallback ( handle , c_callback , 0 )
try :
    rarinfo = self . _read_header ( handle )
    while rarinfo is not None :
        if rarinfo . filename == member :
            self . _process_current ( handle , constants . RAR_TEST )
            break
        else :
            self . _process_current ( handle , constants . RAR_SKIP )
        rarinfo = self . _read_header ( handle )
    if rarinfo is None :
        data = None
except unrarlib . MissingPassword :
    raise RuntimeError ( "File is encrypted, password required" )
except unrarlib . BadPassword :
    raise RuntimeError ( "Bad password for File" )
except unrarlib . BadDataError :
    if password is not None :
        raise RuntimeError ( "File CRC error or incorrect password" )
    else :
        raise RuntimeError ( "File CRC error" )
except unrarlib . UnrarException as e :
    raise BadRarFile ( "Bad RAR archive data: %s" % str ( e ) )
finally :
    self . _close ( handle )
if data is None :
    raise KeyError ( 'There is no item named %r in the archive' % member )
# return file - like object
return data . get_bytes ( )


signature:
def namelist ( self ) :
    """Return a list of file names in the archive ."""

implementation:
names = [ ]
for member in self . filelist :
    names . append ( member . filename )
return names


signature:
def getinfo ( self , name ) :
    """Return the instance of RarInfo given ' name ' ."""

implementation:
rarinfo = self . NameToInfo . get ( name )
if rarinfo is None :
    raise KeyError ( 'There is no item named %r in the archive' % name )
return rarinfo


signature:
def printdir ( self ) :
    """Print a table of contents for the RAR file ."""

implementation:
print ( "%-46s %19s %12s" % ( "File Name" , "Modified    " , "Size" ) )
for rarinfo in self . filelist :
    date = "%d-%02d-%02d %02d:%02d:%02d" % rarinfo . date_time [ : 6 ]
    print ( "%-46s %s %12d" % ( rarinfo . filename , date , rarinfo . file_size ) )


signature:
def extract ( self , member , path = None , pwd = None ) :
    """Extract a member from the archive to the current working directory ,
 using its full name . Its file information is extracted as accurately
 as possible . ` member ' may be a filename or a RarInfo object . You can
 specify a different directory using ` path ' ."""

implementation:
if isinstance ( member , RarInfo ) :
    member = member . filename
if path is None :
    path = os . getcwd ( )
self . _extract_members ( [ member ] , path , pwd )
return os . path . join ( path , member )


signature:
def extractall ( self , path = None , members = None , pwd = None ) :
    """Extract all members from the archive to the current working
 directory . ` path ' specifies a different directory to extract to .
 ` members ' is optional and must be a subset of the list returned
 by namelist ( ) ."""

implementation:
if members is None :
    members = self . namelist ( )
self . _extract_members ( members , path , pwd )


signature:
def _extract_members ( self , members , targetpath , pwd ) :
    """Extract the RarInfo objects ' members ' to a physical
 file on the path targetpath ."""

implementation:
archive = unrarlib . RAROpenArchiveDataEx ( self . filename , mode = constants . RAR_OM_EXTRACT )
handle = self . _open ( archive )
password = pwd or self . pwd
if password is not None :
    unrarlib . RARSetPassword ( handle , b ( password ) )
try :
    rarinfo = self . _read_header ( handle )
    while rarinfo is not None :
        if rarinfo . filename in members :
            self . _process_current ( handle , constants . RAR_EXTRACT , targetpath )
        else :
            self . _process_current ( handle , constants . RAR_SKIP )
        rarinfo = self . _read_header ( handle )
except unrarlib . MissingPassword :
    raise RuntimeError ( "File is encrypted, password required" )
except unrarlib . BadPassword :
    raise RuntimeError ( "Bad password for File" )
except unrarlib . BadDataError :
    raise RuntimeError ( "File CRC Error" )
except unrarlib . UnrarException as e :
    raise BadRarFile ( "Bad RAR archive data: %s" % str ( e ) )
finally :
    self . _close ( handle )


signature:
def dostime_to_timetuple ( dostime ) :
    """Convert a RAR archive member DOS time to a Python time tuple ."""

implementation:
dostime = dostime >> 16
dostime = dostime & 0xffff
day = dostime & 0x1f
month = ( dostime >> 5 ) & 0xf
year = 1980 + ( dostime >> 9 )
second = 2 * ( dostime & 0x1f )
minute = ( dostime >> 5 ) & 0x3f
hour = dostime >> 11
return ( year , month , day , hour , minute , second )


signature:
def _c_func ( func , restype , argtypes , errcheck = None ) :
    """Wrap c function setting prototype ."""

implementation:
func . restype = restype
func . argtypes = argtypes
if errcheck is not None :
    func . errcheck = errcheck
return func


signature:
def _load_savefile_header ( file_h ) :
    """Load and validate the header of a pcap file ."""

implementation:
try :
    raw_savefile_header = file_h . read ( 24 )
except UnicodeDecodeError :
    print ( "\nMake sure the input file is opened in read binary, 'rb'\n" )
    raise InvalidEncoding ( "Could not read file; it might not be opened in binary mode." )
# in case the capture file is not the same endianness as ours , we have to
# use the correct byte order for the file header
if raw_savefile_header [ : 4 ] in [ struct . pack ( ">I" , _MAGIC_NUMBER ) , struct . pack ( ">I" , _MAGIC_NUMBER_NS ) ] :
    byte_order = b'big'
    unpacked = struct . unpack ( '>IhhIIII' , raw_savefile_header )
elif raw_savefile_header [ : 4 ] in [ struct . pack ( "<I" , _MAGIC_NUMBER ) , struct . pack ( "<I" , _MAGIC_NUMBER_NS ) ] :
    byte_order = b'little'
    unpacked = struct . unpack ( '<IhhIIII' , raw_savefile_header )
else :
    raise UnknownMagicNumber ( "No supported Magic Number found" )
( magic , major , minor , tz_off , ts_acc , snaplen , ll_type ) = unpacked
header = __pcap_header__ ( magic , major , minor , tz_off , ts_acc , snaplen , ll_type , ctypes . c_char_p ( byte_order ) , magic == _MAGIC_NUMBER_NS )
if not __validate_header__ ( header ) :
    raise InvalidHeader ( "Invalid Header" )
else :
    return header


signature:
def load_savefile ( input_file , layers = 0 , verbose = False , lazy = False ) :
    """Parse a savefile as a pcap _ savefile instance . Returns the savefile
 on success and None on failure . Verbose mode prints additional information
 about the file ' s processing . layers defines how many layers to descend and
 decode the packet . input _ file should be a Python file object ."""

implementation:
global VERBOSE
old_verbose = VERBOSE
VERBOSE = verbose
__TRACE__ ( '[+] attempting to load {:s}' , ( input_file . name , ) )
header = _load_savefile_header ( input_file )
if __validate_header__ ( header ) :
    __TRACE__ ( '[+] found valid header' )
    if lazy :
        packets = _generate_packets ( input_file , header , layers )
        __TRACE__ ( '[+] created packet generator' )
    else :
        packets = _load_packets ( input_file , header , layers )
        __TRACE__ ( '[+] loaded {:d} packets' , ( len ( packets ) , ) )
    sfile = pcap_savefile ( header , packets )
    __TRACE__ ( '[+] finished loading savefile.' )
else :
    __TRACE__ ( '[!] invalid savefile' )
    sfile = None
VERBOSE = old_verbose
return sfile


signature:
def _load_packets ( file_h , header , layers = 0 ) :
    """Read packets from the capture file . Expects the file handle to point to
 the location immediately after the header ( 24 bytes ) ."""

implementation:
pkts = [ ]
hdrp = ctypes . pointer ( header )
while True :
    pkt = _read_a_packet ( file_h , hdrp , layers )
    if pkt :
        pkts . append ( pkt )
    else :
        break
return pkts


signature:
def _generate_packets ( file_h , header , layers = 0 ) :
    """Read packets one by one from the capture file . Expects the file
 handle to point to the location immediately after the header ( 24
 bytes ) ."""

implementation:
hdrp = ctypes . pointer ( header )
while True :
    pkt = _read_a_packet ( file_h , hdrp , layers )
    if pkt :
        yield pkt
    else :
        break


signature:
def _read_a_packet ( file_h , hdrp , layers = 0 ) :
    """Reads the next individual packet from the capture file . Expects
 the file handle to be somewhere after the header , on the next
 per - packet header ."""

implementation:
raw_packet_header = file_h . read ( 16 )
if not raw_packet_header or len ( raw_packet_header ) != 16 :
    return None
# in case the capture file is not the same endianness as ours , we have to
# use the correct byte order for the packet header
if hdrp [ 0 ] . byteorder == 'big' :
    packet_header = struct . unpack ( '>IIII' , raw_packet_header )
else :
    packet_header = struct . unpack ( '<IIII' , raw_packet_header )
( timestamp , timestamp_us , capture_len , packet_len ) = packet_header
raw_packet_data = file_h . read ( capture_len )
if not raw_packet_data or len ( raw_packet_data ) != capture_len :
    return None
if layers > 0 :
    layers -= 1
    raw_packet = linklayer . clookup ( hdrp [ 0 ] . ll_type ) ( raw_packet_data , layers = layers )
else :
    raw_packet = raw_packet_data
packet = pcap_packet ( hdrp , timestamp , timestamp_us , capture_len , packet_len , raw_packet )
return packet


signature:
def parse_ipv4 ( address ) :
    """Given a raw IPv4 address ( i . e . as an unsigned integer ) , return it in
 dotted quad notation ."""

implementation:
raw = struct . pack ( 'I' , address )
octets = struct . unpack ( 'BBBB' , raw ) [ : : - 1 ]
ipv4 = b'.' . join ( [ ( '%d' % o ) . encode ( 'ascii' ) for o in bytearray ( octets ) ] )
return ipv4


signature:
def strip_ip ( packet ) :
    """Remove the IP packet layer , yielding the transport layer ."""

implementation:
if not isinstance ( packet , IP ) :
    packet = IP ( packet )
payload = packet . payload
return payload


signature:
def strip_ethernet ( packet ) :
    """Strip the Ethernet frame from a packet ."""

implementation:
if not isinstance ( packet , Ethernet ) :
    packet = Ethernet ( packet )
payload = packet . payload
return payload


signature:
def load_network ( self , layers = 1 ) :
    """Given an Ethernet frame , determine the appropriate sub - protocol ;
 If layers is greater than zerol determine the type of the payload
 and load the appropriate type of network packet . It is expected
 that the payload be a hexified string . The layers argument determines
 how many layers to descend while parsing the packet ."""

implementation:
if layers :
    ctor = payload_type ( self . type ) [ 0 ]
    if ctor :
        ctor = ctor
        payload = self . payload
        self . payload = ctor ( payload , layers - 1 )
    else : # if no type is found , do not touch the packet .
        pass


signature:
def WIFI ( frame , no_rtap = False ) :
    """calls wifi packet discriminator and constructor .
 : frame : ctypes . Structure
 : no _ rtap : Bool
 : return : packet object in success
 : return : int
 -1 on known error
 : return : int
 -2 on unknown error"""

implementation:
pack = None
try :
    pack = WiHelper . get_wifi_packet ( frame , no_rtap )
except Exception as e :
    logging . exception ( e )
return pack


signature:
def get_wifi_packet ( frame , no_rtap = False ) :
    """Discriminates Wi - Fi packet and creates
 packet object .
 : frame : ctypes . Structure
 : no _ rtap : Bool
 : return : obj
 Wi - Fi packet"""

implementation:
_ , packet = WiHelper . _strip_rtap ( frame )
frame_control = struct . unpack ( 'BB' , packet [ : 2 ] )
cat = ( frame_control [ 0 ] >> 2 ) & 0b0011
s_type = frame_control [ 0 ] >> 4
if cat not in _CATEGORIES_ . keys ( ) :
    logging . warning ( "unknown category: %d" % ( cat ) )
    return Unknown ( frame , no_rtap )
if s_type not in _SUBTYPES_ [ cat ] . keys ( ) :
    logging . warning ( "unknown subtype %d in %s category" % ( s_type , _CATEGORIES_ [ cat ] ) )
    return Unknown ( frame , no_rtap )
if cat == 0 :
    if s_type == 4 :
        return ProbeReq ( frame , no_rtap )
    elif s_type == 5 :
        return ProbeResp ( frame , no_rtap )
    elif s_type == 8 :
        return Beacon ( frame , no_rtap )
    else :
        return Management ( frame , no_rtap )
elif cat == 1 :
    if s_type == 11 :
        return RTS ( frame , no_rtap )
    elif s_type == 12 :
        return CTS ( frame , no_rtap )
    elif s_type == 9 :
        return BACK ( frame , no_rtap )
    else :
        return Control ( frame , no_rtap )
elif cat == 2 :
    if s_type == 8 :
        return QosData ( frame , no_rtap , parse_amsdu = True )
    else :
        return Data ( frame , no_rtap )


signature:
def _strip_rtap ( frame ) :
    """strip injected radiotap header .
 : return : ctypes . Structure
 radiotap header
 : return : ctypes . Structure
 actual layer 2 Wi - Fi payload"""

implementation:
rtap_len = WiHelper . __get_rtap_len ( frame )
rtap = frame [ : rtap_len ]
packet = frame [ rtap_len : ]
return rtap , packet


signature:
def strip_present ( payload ) :
    """strip ( 4 byte ) radiotap . present . Those are flags that
 identify existence of incoming radiotap meta - data .
 : idx : int
 : return : str
 : return : namedtuple"""

implementation:
present = collections . namedtuple ( 'present' , [ 'tsft' , 'flags' , 'rate' , 'channel' , 'fhss' , 'dbm_antsignal' , 'dbm_antnoise' , 'lock_quality' , 'tx_attenuation' , 'db_tx_attenuation' , 'dbm_tx_power' , 'antenna' , 'db_antsignal' , 'db_antnoise' , 'rxflags' , 'txflags' , 'rts_retries' , 'data_retries' , 'xchannel' , 'mcs' , 'ampdu' , 'vht' , 'rtap_ns' , 'ven_ns' , 'ext' ] )
val = struct . unpack ( '<L' , payload ) [ 0 ]
bits = format ( val , '032b' ) [ : : - 1 ]
present . tsft = int ( bits [ 0 ] )
# timer synchronization function
present . flags = int ( bits [ 1 ] )
# flags
present . rate = int ( bits [ 2 ] )
# rate
present . channel = int ( bits [ 3 ] )
# channel
present . fhss = int ( bits [ 4 ] )
# frequency hoping spread spectrum
present . dbm_antsignal = int ( bits [ 5 ] )
# dbm antenna signal
present . dbm_antnoise = int ( bits [ 6 ] )
# dbm antenna noinse
present . lock_quality = int ( bits [ 7 ] )
# quality of barker code lock
present . tx_attenuation = int ( bits [ 8 ] )
# transmitter attenuation
present . db_tx_attenuation = int ( bits [ 9 ] )
# decibel transmit attenuation
present . dbm_tx_power = int ( bits [ 10 ] )
# dbm transmit power
present . antenna = int ( bits [ 11 ] )
# antenna
present . db_antsignal = int ( bits [ 12 ] )
# db antenna signal
present . db_antnoise = int ( bits [ 13 ] )
# db antenna noise
present . rxflags = int ( bits [ 14 ] )
# receiver flags
present . txflags = int ( bits [ 15 ] )
# transmitter flags
present . rts_retries = int ( bits [ 16 ] )
# rts ( request to send ) retries
present . data_retries = int ( bits [ 17 ] )
# data retries
present . xchannel = int ( bits [ 18 ] )
# xchannel
present . mcs = int ( bits [ 19 ] )
# modulation and coding scheme
present . ampdu = int ( bits [ 20 ] )
# aggregated mac protocol data unit
present . vht = int ( bits [ 21 ] )
# very high throughput
present . rtap_ns = int ( bits [ 29 ] )
# radiotap namespace
present . ven_ns = int ( bits [ 30 ] )
# vendor namespace
present . ext = int ( bits [ 31 ] )
# extension
return present , bits


signature:
def strip_tsft ( self , idx ) :
    """strip ( 8 byte ) radiotap . mactime
 : idx : int
 : return : int
 idx
 : return : int
 mactime"""

implementation:
idx = Radiotap . align ( idx , 8 )
mactime , = struct . unpack_from ( '<Q' , self . _rtap , idx )
return idx + 8 , mactime


signature:
def strip_flags ( self , idx ) :
    """strip ( 1 byte ) radiotap . flags
 : idx : int
 : return : int
 idx
 : return : collections . namedtuple"""

implementation:
flags = collections . namedtuple ( 'flags' , [ 'cfp' , 'preamble' , 'wep' , 'fragmentation' , 'fcs' , 'datapad' , 'badfcs' , 'shortgi' ] )
val , = struct . unpack_from ( '<B' , self . _rtap , idx )
bits = format ( val , '08b' ) [ : : - 1 ]
flags . cfp = int ( bits [ 0 ] )
flags . preamble = int ( bits [ 1 ] )
flags . wep = int ( bits [ 2 ] )
flags . fragmentation = int ( bits [ 3 ] )
flags . fcs = int ( bits [ 4 ] )
flags . datapad = int ( bits [ 5 ] )
flags . badfcs = int ( bits [ 6 ] )
flags . shortgi = int ( bits [ 7 ] )
return idx + 1 , flags


signature:
def strip_rate ( self , idx ) :
    """strip ( 1 byte ) radiotap . datarate
 note that , unit of this field is originally 0.5 Mbps
 : idx : int
 : return : int
 idx
 : return : double
 rate in terms of Mbps"""

implementation:
val , = struct . unpack_from ( '<B' , self . _rtap , idx )
rate_unit = float ( 1 ) / 2
# Mbps
return idx + 1 , rate_unit * val


signature:
def strip_chan ( self , idx ) :
    """strip ( 2 byte ) radiotap . channel . flags
 : idx : int
 : return : int
 idx
 : return : collections . namedtuple"""

implementation:
chan = collections . namedtuple ( 'chan' , [ 'freq' , 'turbo' , 'cck' , 'ofdm' , 'two_g' , 'five_g' , 'passive' , 'dynamic' , 'gfsk' , 'gsm' , 'static_turbo' , 'half_rate' , 'quarter_rate' ] )
idx = Radiotap . align ( idx , 2 )
freq , flags , = struct . unpack_from ( '<HH' , self . _rtap , idx )
chan . freq = freq
bits = format ( flags , '016b' ) [ : : - 1 ]
chan . turbo = int ( bits [ 4 ] )
chan . cck = int ( bits [ 5 ] )
chan . ofdm = int ( bits [ 6 ] )
chan . two_g = int ( bits [ 7 ] )
chan . five_g = int ( bits [ 8 ] )
chan . passive = int ( bits [ 9 ] )
chan . dynamic = int ( bits [ 10 ] )
chan . gfsk = int ( bits [ 11 ] )
chan . gsm = int ( bits [ 12 ] )
chan . static_turbo = int ( bits [ 13 ] )
chan . half_rate = int ( bits [ 14 ] )
chan . quarter_rate = int ( bits [ 15 ] )
return idx + 4 , chan


signature:
def strip_fhss ( self , idx ) :
    """strip ( 2 byte ) radiotap . fhss . hopset ( 1 byte ) and
 radiotap . fhss . pattern ( 1 byte )
 : idx : int
 : return : int
 idx
 : return : collections . namedtuple"""

implementation:
fhss = collections . namedtuple ( 'fhss' , [ 'hopset' , 'pattern' ] )
fhss . hopset , fhss . pattern , = struct . unpack_from ( '<bb' , self . _rtap , idx )
return idx + 2 , fhss


signature:
def strip_dbm_antsignal ( self , idx ) :
    """strip ( 1 byte ) radiotap . dbm . ant _ signal
 : idx : int
 : return : int
 idx
 : return : int"""

implementation:
dbm_antsignal , = struct . unpack_from ( '<b' , self . _rtap , idx )
return idx + 1 , dbm_antsignal


signature:
def strip_dbm_antnoise ( self , idx ) :
    """strip ( 1 byte ) radiotap . dbm _ antnoise
 : idx : int
 : return : int
 idx
 : return : int"""

implementation:
dbm_antnoise , = struct . unpack_from ( '<b' , self . _rtap , idx )
return idx + 1 , dbm_antnoise


signature:
def strip_lock_quality ( self , idx ) :
    """strip ( 2 byte ) lock quality
 : idx : int
 : return : int
 idx
 : return : int"""

implementation:
idx = Radiotap . align ( idx , 2 )
lock_quality , = struct . unpack_from ( '<H' , self . _rtap , idx )
return idx + 2 , lock_quality


signature:
def strip_tx_attenuation ( self , idx ) :
    """strip ( 1 byte ) tx _ attenuation
 : idx : int
 : return : int
 idx
 : return : int"""

implementation:
idx = Radiotap . align ( idx , 2 )
tx_attenuation , = struct . unpack_from ( '<H' , self . _rtap , idx )
return idx + 2 , tx_attenuation


signature:
def strip_db_tx_attenuation ( self , idx ) :
    """strip ( 1 byte ) db _ tx _ attenuation
 : return : int
 idx
 : return : int"""

implementation:
idx = Radiotap . align ( idx , 2 )
db_tx_attenuation , = struct . unpack_from ( '<H' , self . _rtap , idx )
return idx + 2 , db_tx_attenuation


signature:
def strip_dbm_tx_power ( self , idx ) :
    """strip ( 1 byte ) dbm _ tx _ power
 : return : int
 idx
 : return : int"""

implementation:
idx = Radiotap . align ( idx , 1 )
dbm_tx_power , = struct . unpack_from ( '<b' , self . _rtap , idx )
return idx + 1 , dbm_tx_power


signature:
def strip_antenna ( self , idx ) :
    """strip ( 1 byte ) radiotap . antenna
 : return : int
 idx
 : return : int"""

implementation:
antenna , = struct . unpack_from ( '<B' , self . _rtap , idx )
return idx + 1 , antenna


signature:
def strip_db_antsignal ( self , idx ) :
    """strip ( 1 byte ) radiotap . db _ antsignal
 : return : int
 idx
 : return : int"""

implementation:
db_antsignal , = struct . unpack_from ( '<B' , self . _rtap , idx )
return idx + 1 , db_antsignal


signature:
def strip_db_antnoise ( self , idx ) :
    """strip ( 1 byte ) radiotap . db _ antnoise
 : return : int
 idx
 : return : int"""

implementation:
db_antnoise , = struct . unpack_from ( '<B' , self . _rtap , idx )
return idx + 1 , db_antnoise


signature:
def strip_rx_flags ( self , idx ) :
    """strip ( 2 byte ) radiotap . rxflags
 : idx : int
 : return : int
 idx
 : return : collections . namedtuple"""

implementation:
rx_flags = collections . namedtuple ( 'rx_flags' , [ 'reserved' , 'badplcp' ] )
idx = Radiotap . align ( idx , 2 )
flags , = struct . unpack_from ( '<H' , self . _rtap , idx )
flag_bits = format ( flags , '08b' ) [ : : - 1 ]
rx_flags . reserved = int ( flag_bits [ 0 ] )
rx_flags . badplcp = int ( flag_bits [ 1 ] )
return idx + 2 , rx_flags


signature:
def strip_tx_flags ( self , idx ) :
    """strip ( 1 byte ) tx _ flags
 : idx : int
 : return : int
 idx
 : return : int"""

implementation:
idx = Radiotap . align ( idx , 2 )
tx_flags , = struct . unpack_from ( '<B' , self . _rtap , idx )
return idx + 1 , tx_flags


signature:
def strip_rts_retries ( self , idx ) :
    """strip ( 1 byte ) rts _ retries
 : idx : int
 : return : int
 idx
 : return : int"""

implementation:
rts_retries , = struct . unpack_from ( '<B' , self . _rtap , idx )
return idx + 1 , rts_retries


signature:
def strip_data_retries ( self , idx ) :
    """strip ( 1 byte ) data _ retries
 : idx : int
 : return : int
 idx
 : return : int"""

implementation:
data_retries , = struct . unpack_from ( '<B' , self . _rtap , idx )
return idx + 1 , data_retries


signature:
def strip_xchannel ( self , idx ) :
    """strip ( 7 bytes ) radiotap . xchannel . channel ( 1 byte ) ,
 radiotap . xchannel . freq ( 2 bytes ) and radiotap . xchannel . flags ( 4 bytes )
 : idx : int
 : return : int
 idx
 : return : collections . namedtuple"""

implementation:
xchannel = collections . namedtuple ( 'xchannel' , [ 'flags' , 'freq' , 'channel' , 'max_power' ] )
flags = collections . namedtuple ( 'flags' , [ 'turbo' , 'cck' , 'ofdm' , 'two_g' , 'five_g' , 'passive' , 'dynamic' , 'gfsk' , 'gsm' , 'sturbo' , 'hafl' , 'quarter' , 'ht_20' , 'ht_40u' , 'ht_40d' ] )
idx = Radiotap . align ( idx , 2 )
flag_val , freq , channel , max_power = struct . unpack_from ( '<lHBB' , self . _rtap , idx )
xchannel . freq = freq
xchannel . channel = channel
xchannel . max_power = max_power
bits = format ( flag_val , '032b' ) [ : : - 1 ]
flags . turbo = int ( bits [ 4 ] )
flags . cck = int ( bits [ 5 ] )
flags . ofdm = int ( bits [ 6 ] )
flags . two_g = int ( bits [ 7 ] )
flags . five_g = int ( bits [ 8 ] )
flags . passive = int ( bits [ 9 ] )
flags . dynamic = int ( bits [ 10 ] )
flags . gfsk = int ( bits [ 11 ] )
flags . gsm = int ( bits [ 12 ] )
flags . sturbo = int ( bits [ 13 ] )
flags . half = int ( bits [ 14 ] )
flags . quarter = int ( bits [ 15 ] )
flags . ht_20 = int ( bits [ 16 ] )
flags . ht_40u = int ( bits [ 17 ] )
flags . ht_40d = int ( bits [ 18 ] )
xchannel . flags = flags
return idx + 8 , xchannel


signature:
def strip_mcs ( self , idx ) :
    """strip ( 3 byte ) radiotap . mcs which contains 802.11n bandwidth ,
 mcs ( modulation and coding scheme ) and stbc ( space time block coding )
 information .
 : idx : int
 : return : int
 idx
 : return : collections . namedtuple"""

implementation:
mcs = collections . namedtuple ( 'mcs' , [ 'known' , 'index' , 'have_bw' , 'have_mcs' , 'have_gi' , 'have_format' , 'have_fec' , 'have_stbc' , 'have_ness' , 'ness_bit1' ] )
idx = Radiotap . align ( idx , 1 )
known , flags , index = struct . unpack_from ( '<BBB' , self . _rtap , idx )
bits = format ( flags , '032b' ) [ : : - 1 ]
mcs . known = known
# Known MCS information
mcs . index = index
# MCS index
mcs . have_bw = int ( bits [ 0 ] )
# Bandwidth
mcs . have_mcs = int ( bits [ 1 ] )
# MCS
mcs . have_gi = int ( bits [ 2 ] )
# Guard Interval
mcs . have_format = int ( bits [ 3 ] )
# Format
mcs . have_fec = int ( bits [ 4 ] )
# FEC ( Forward Error Correction ) type
mcs . have_stbc = int ( bits [ 5 ] )
# Space Time Block Coding
mcs . have_ness = int ( bits [ 6 ] )
# Number of Extension Spatial Streams
mcs . ness_bit1 = int ( bits [ 7 ] )
# Number of Extension Spatial Streams bit 1
return idx + 3 , mcs


signature:
def strip_ampdu ( self , idx ) :
    """strip ( 8 byte ) radiotap . ampdu
 : idx : int
 : return : int
 idx
 : return : collections . namedtuple"""

implementation:
ampdu = collections . namedtuple ( 'ampdu' , [ 'reference' , 'crc_val' , 'reservered' , 'flags' ] )
flags = collections . namedtuple ( 'flags' , [ 'report_zerolen' , 'is_zerolen' , 'lastknown' , 'last' , 'delim_crc_error' ] )
idx = Radiotap . align ( idx , 4 )
refnum , flag_vals , crc_val , reserved = struct . unpack_from ( '<LHBB' , self . _rtap , idx )
ampdu . flags = flags
ampdu . reference = refnum
ampdu . crc_val = crc_val
ampdu . reserved = reserved
bits = format ( flag_vals , '032b' ) [ : : - 1 ]
ampdu . flags . report_zerolen = int ( bits [ 0 ] )
ampdu . flags . is_zerolen = int ( bits [ 1 ] )
ampdu . flags . lastknown = int ( bits [ 2 ] )
ampdu . flags . last = int ( bits [ 3 ] )
ampdu . flags . delim_crc_error = int ( bits [ 4 ] )
return idx + 8 , ampdu


signature:
def strip_vht ( self , idx ) :
    """strip ( 12 byte ) radiotap . vht
 : idx : int
 : return : int
 idx
 : return : collections . namedtuple"""

implementation:
vht = collections . namedtuple ( 'vht' , [ 'known_bits' , 'have_stbc' , 'have_txop_ps' , 'have_gi' , 'have_sgi_nsym_da' , 'have_ldpc_extra' , 'have_beamformed' , 'have_bw' , 'have_gid' , 'have_paid' , 'stbc' , 'txop_ps' , 'gi' , 'sgi_nysm_da' , 'ldpc_extra' , 'group_id' , 'partial_id' , 'beamformed' , 'user_0' , 'user_1' , 'user_2' , 'user_3' ] )
user = collections . namedtuple ( 'user' , [ 'nss' , 'mcs' , 'coding' ] )
idx = Radiotap . align ( idx , 2 )
known , flags , bw = struct . unpack_from ( '<HBB' , self . _rtap , idx )
mcs_nss_0 , mcs_nss_1 , mcs_nss_2 , mcs_nss_3 = struct . unpack_from ( '<BBBB' , self . _rtap , idx + 4 )
coding , group_id , partial_id = struct . unpack_from ( '<BBH' , self . _rtap , idx + 8 )
known_bits = format ( known , '032b' ) [ : : - 1 ]
vht . known_bits = known_bits
vht . have_stbc = int ( known_bits [ 0 ] )
# Space Time Block Coding
vht . have_txop_ps = int ( known_bits [ 1 ] )
# TXOP _ PS _ NOT _ ALLOWD
vht . have_gi = int ( known_bits [ 2 ] )
# Short / Long Guard Interval
vht . have_sgi_nsym_da = int ( known_bits [ 3 ] )
# Short Guard Interval Nsym Disambiguation
vht . have_ldpc_extra = int ( known_bits [ 4 ] )
# LDPC ( Low Density Parity Check )
vht . have_beamformed = int ( known_bits [ 5 ] )
# Beamformed
vht . have_bw = int ( known_bits [ 6 ] )
# Bandwidth
vht . have_gid = int ( known_bits [ 7 ] )
# Group ID
vht . have_paid = int ( known_bits [ 8 ] )
# Partial AID
flag_bits = format ( flags , '032b' ) [ : : - 1 ]
vht . flag_bits = flag_bits
vht . stbc = int ( flag_bits [ 0 ] )
vht . txop_ps = int ( flag_bits [ 1 ] )
vht . gi = int ( flag_bits [ 2 ] )
vht . sgi_nysm_da = int ( flag_bits [ 3 ] )
vht . ldpc_extra = int ( flag_bits [ 4 ] )
vht . beamformed = int ( flag_bits [ 5 ] )
vht . group_id = group_id
vht . partial_id = partial_id
vht . bw = bw
vht . user_0 = user ( None , None , None )
vht . user_1 = user ( None , None , None )
vht . user_2 = user ( None , None , None )
vht . user_3 = user ( None , None , None )
for ( i , mcs_nss ) in enumerate ( [ mcs_nss_0 , mcs_nss_1 , mcs_nss_2 , mcs_nss_3 ] ) :
    if mcs_nss :
        nss = mcs_nss & 0xf0 >> 4
        mcs = ( mcs_nss & 0xf0 ) >> 4
        coding = ( coding & 2 ** i ) >> i
        if i == 0 :
            vht . user_0 = user ( nss , mcs , coding )
        elif i == 1 :
            vht . user_1 = user ( nss , mcs , coding )
        elif i == 2 :
            vht . user_2 = user ( nss , mcs , coding )
        elif i == 3 :
            vht . user_3 = user ( nss , mcs , coding )
return idx + 12 , vht


signature:
def extract_protocol ( self ) :
    """extract 802.11 protocol from radiotap . channel . flags
 : return : str
 protocol name
 one of below in success
 [ . 11a , . 11b , . 11g , . 11n , . 11ac ]
 None in fail"""

implementation:
if self . present . mcs :
    return '.11n'
if self . present . vht :
    return '.11ac'
if self . present . channel and hasattr ( self , 'chan' ) :
    if self . chan . five_g :
        if self . chan . ofdm :
            return '.11a'
    elif self . chan . two_g :
        if self . chan . cck :
            return '.11b'
        elif self . chan . ofdm or self . chan . dynamic :
            return '.11g'
return 'None'


signature:
def get_shark_field ( self , fields ) :
    """get parameters via wireshark syntax .
 out = x . get _ shark _ field ( ' wlan . fc . type ' )
 out = x . get _ shark _ field ( [ ' wlan . fc . type ' , ' wlan . seq ' ] )
 : fields : str or str [ ]
 : return : dict
 out [ fields [ 0 ] ] = val [ 0 ] or None
 out [ fields [ 1 ] ] = val [ 1 ] or None . . ."""

implementation:
keys , exist , out = None , { } , None
if isinstance ( fields , str ) :
    fields = [ fields ]
elif not isinstance ( fields , list ) :
    logging . error ( 'invalid input type' )
    return None
out = dict . fromkeys ( fields )
if hasattr ( self , '_shark_' ) :
    exist . update ( self . _shark_ )
if hasattr ( self , '_s_shark_' ) :
    exist . update ( self . _s_shark_ )
if hasattr ( self . radiotap , '_r_shark_' ) :
    exist . update ( self . radiotap . _r_shark_ )
keys = exist . keys ( )
for elem in fields :
    if elem in keys :
        obj_field , tmp = exist [ elem ] , None
        try :
            tmp = operator . attrgetter ( obj_field ) ( self )
        except AttributeError :
            tmp = None
        if not tmp :
            try :
                tmp = operator . attrgetter ( obj_field ) ( self . radiotap )
            except AttributeError :
                tmp = None
        out [ elem ] = tmp
return out


signature:
def get_mac_addr ( mac_addr ) :
    """converts bytes to mac addr format
 : mac _ addr : ctypes . structure
 : return : str
 mac addr in format
 11:22:33 : aa : bb : cc"""

implementation:
mac_addr = bytearray ( mac_addr )
mac = b':' . join ( [ ( '%02x' % o ) . encode ( 'ascii' ) for o in mac_addr ] )
return mac


signature:
def strip_mac_addrs ( self ) :
    """strip mac address ( each 6 byte ) information .
 ( wlan . ta , wlan . ra , wlan . sa , wlan . da )
 ( transmitter , receiver , source , destination )
 : return : int
 index of sequence control
 : return : int
 index after mac addresses
 : return : str
 source address ( sa )
 : return : str
 transmitter address ( ta )
 : return : str
 receiver address ( ra )
 : return : str
 destination address ( da )
 : return : str
 basic service sed identifier ( bssid )"""

implementation:
qos_idx , seq_idx = 0 , 0
sa , ta , ra , da , bssid = None , None , None , None , None
if self . to_ds == 1 and self . from_ds == 1 :
    ( ra , ta , da ) = struct . unpack ( '!6s6s6s' , self . _packet [ 4 : 22 ] )
    sa = struct . unpack ( '!6s' , self . _packet [ 24 : 30 ] ) [ 0 ]
    qos_idx = 30
    seq_idx = 22
elif self . to_ds == 0 and self . from_ds == 1 :
    ( ra , ta , sa ) = struct . unpack ( '!6s6s6s' , self . _packet [ 4 : 22 ] )
    qos_idx = 24
    seq_idx = 22
elif self . to_ds == 1 and self . from_ds == 0 :
    ( ra , ta , da ) = struct . unpack ( '!6s6s6s' , self . _packet [ 4 : 22 ] )
    qos_idx = 24
    seq_idx = 22
elif self . to_ds == 0 and self . from_ds == 0 :
    ( ra , ta , bssid ) = struct . unpack ( '!6s6s6s' , self . _packet [ 4 : 22 ] )
    qos_idx = 24
    seq_idx = 22
if ta is not None :
    ta = Wifi . get_mac_addr ( ta )
if ra is not None :
    ra = Wifi . get_mac_addr ( ra )
if sa is not None :
    sa = Wifi . get_mac_addr ( sa )
if da is not None :
    da = Wifi . get_mac_addr ( da )
if bssid is not None :
    bssid = Wifi . get_mac_addr ( bssid )
return seq_idx , qos_idx , sa , ta , ra , da , bssid


signature:
def strip_seq_cntrl ( self , idx ) :
    """strip ( 2 byte ) wlan . seq ( 12 bit ) and wlan . fram ( 4 bit )
 number information .
 : seq _ cntrl : ctypes . Structure
 : return : int
 sequence number
 : return : int
 fragment number"""

implementation:
seq_cntrl = struct . unpack ( 'H' , self . _packet [ idx : idx + 2 ] ) [ 0 ]
seq_num = seq_cntrl >> 4
frag_num = seq_cntrl & 0x000f
return seq_num , frag_num


signature:
def strip_qos_cntrl ( self , idx , prot_type ) :
    """strip ( 2 byte ) wlan . qos
 : idx : int
 : prot _ type : string
 802.11 protocol type ( . 11ac , . 11a , . 11n , etc )
 : return : int
 number of processed bytes
 : return : int
 qos priority
 : return : int
 qos bit
 : return : int
 qos acknowledgement
 : return : int
 amsdupresent ( aggregated mac service data unit )"""

implementation:
qos_cntrl , = struct . unpack ( 'H' , self . _packet [ idx : idx + 2 ] )
qos_cntrl_bits = format ( qos_cntrl , '016b' ) [ : : - 1 ]
qos_pri = qos_cntrl & 0x000f
qos_bit = int ( qos_cntrl_bits [ 5 ] )
qos_ack = int ( qos_cntrl_bits [ 6 : 8 ] , 2 )
amsdupresent = 0
if prot_type == '.11ac' :
    amsdupresent = int ( qos_cntrl_bits [ 7 ] )
return 2 , qos_pri , qos_bit , qos_ack , amsdupresent


signature:
def strip_ccmp ( self , idx ) :
    """strip ( 8 byte ) wlan . ccmp . extiv
 CCMP Extended Initialization Vector
 : return : int
 number of processed bytes
 : return : ctypes . raw
 ccmp vector"""

implementation:
ccmp_extiv = None
if len ( self . _packet [ idx : ] ) >= 8 :
    raw_bytes = self . _packet [ idx : idx + 8 ]
    ccmp_extiv , = struct . unpack_from ( 'Q' , raw_bytes , 0 )
return 8 , ccmp_extiv


signature:
def strip_msdu ( self , idx ) :
    """strip single mac servis data unit ( msdu )
 see - > https : / / mrncciew . com / 2014/11/01 / cwap - 802-11 - data - frame - aggregation /
 : idx : int
 : return : dict
 msdu
 : return : int
 number of processed bytes"""

implementation:
# length of msdu payload has to be multiple of 4,
# this guaranteed with padding
padding = 0
len_payload = 0
msdu = { 'llc' : { } , 'wlan.da' : None , 'wlan.sa' : None , 'payload' : None , 'length' : 0 }
( da_mac , sa_mac ) = struct . unpack ( '!6s6s' , self . _packet [ idx : idx + 12 ] )
msdu [ 'wlan.da' ] = Wifi . get_mac_addr ( da_mac )
msdu [ 'wlan.sa' ] = Wifi . get_mac_addr ( sa_mac )
idx += 12
msdu [ 'length' ] = struct . unpack ( '!H' , self . _packet [ idx : idx + 2 ] ) [ 0 ]
idx += 2
offset , msdu [ 'llc' ] = self . strip_llc ( idx )
idx += offset
len_payload = msdu [ 'length' ] - offset
msdu [ 'payload' ] = self . _packet [ idx : idx + len_payload ]
padding = 4 - ( len_payload % 4 )
return msdu , msdu [ 'length' ] + padding + 12


signature:
def strip_llc ( self , idx ) :
    """strip ( 4 or 8 byte ) logical link control headers
 : return : int
 number of processed bytes
 : return : dict
 llc information
 see - > http : / / www . wildpackets . com / resources / compendium / ethernet / frame _ snap _ iee8023
 ABBRVS .
 ssap : source service access point
 dsap : destination service access point
 SNAP ( Subnetwork Acess Protocol )"""

implementation:
llc = { }
snap = 170
llc_dsap = struct . unpack ( 'B' , self . _packet [ idx : idx + 1 ] ) [ 0 ]
llc [ 'dsap.dsap' ] = llc_dsap >> 1
llc [ 'dsap.ig' ] = llc_dsap & 0b01
idx += 1
llc_ssap = struct . unpack ( 'B' , self . _packet [ idx : idx + 1 ] ) [ 0 ]
llc [ 'ssap.sap' ] = llc_ssap >> 1
llc [ 'ssap.cr' ] = llc_ssap & 0b01
idx += 1
if llc_dsap == snap and llc_ssap == snap :
    llc_control = struct . unpack ( 'B' , self . _packet [ idx : idx + 1 ] ) [ 0 ]
    llc [ 'control.u_modifier_cmd' ] = llc_control >> 2
    llc [ 'control.ftype' ] = llc_control & 0x03
    idx += 1
    llc [ 'organization_code' ] = self . _packet [ idx : idx + 3 ]
    idx += 3
    llc [ 'type' ] = self . _packet [ idx : idx + 2 ]
    return 8 , llc
else :
    return 4 , llc


signature:
def parse_tagged_params ( raw_tagged_params ) :
    """strip tagged information elements wlan _ mgt . tag
 which has generic type - length - value structure
 [ type , length , value ]
 type ( 1 byte ) , length ( 1 byte ) , value ( varies )
 [ wlan _ mgt . tag . number , wlan _ mgt . tag . length , payload ]
 structured fields .
 : return : dict [ ]
 list of tagged params
 : return : int
 0 in succ , 1 for"""

implementation:
fcs_len = 4
# wlan . fcs ( 4 bytes )
idx = 0
tagged_params = [ ]
while idx < len ( raw_tagged_params ) - fcs_len :
    tag_num , tag_len = struct . unpack ( 'BB' , raw_tagged_params [ idx : idx + 2 ] )
    idx += 2
    if len ( raw_tagged_params ) >= idx + tag_len :
        param = { }
        param [ 'number' ] , param [ 'length' ] = tag_num , tag_len
        payload = raw_tagged_params [ idx : idx + tag_len ]
        if tag_num in MNGMT_TAGS :
            param [ 'name' ] = MNGMT_TAGS [ tag_num ]
            if MNGMT_TAGS [ tag_num ] == 'TAG_VENDOR_SPECIFIC_IE' :
                param [ 'payload' ] = Management . parse_vendor_ie ( payload )
            else :
                param [ 'payload' ] = payload
        else :
            param [ 'name' ] = None
        tagged_params . append ( param )
        idx += tag_len
    else :
        logging . warning ( 'out tag length header points out of boundary' )
        log_msg = 'index: {p_idx}, pack_len: {p_len}'
        log_msg = log_msg . format ( p_idx = idx + tag_len , p_len = len ( raw_tagged_params ) )
        logging . warning ( log_msg )
        return 1 , tagged_params
return 0 , tagged_params