def synthesize()

in next_steps/kinesis_stream_connector/l4m_detector/src/synth_live_data_csv.py [0:0]

• 32 lines of code
• 14 McCabe index (conditional complexity)

def synthesize():
    # create item list - for the given set of dimensions and values
    item_list = []
    for dimension_values in itertools.product( *dimensions.values() ):
        item = Item( dict( zip( dimensions.keys(), dimension_values ) ) )
        item_list.append(item)
    
    # itereate and prepare data    
    dimension_values_list = []
    for i in range( len(dimensions) ):
        dimension_values_list.append([])

    timestamp_list = []

    metric_values_list = []
    for i, metric in enumerate(metrics):
        metric_values_list.append([])

    # for each combination of dimesion values, populate metric & timestamp data points
    t = datetime.datetime.utcnow()
    tstr = datetime.datetime.strftime(t,"%Y-%m-%d %H:%M:%S")
    logger.info('Generating data for timestamp '+tstr)

    for item in item_list:
        
        for i, d in enumerate(item.dimension.values()):
            dimension_values_list[i].append(d)
        
        timestamp_list.append(tstr)
        
        metric_values, is_anomaly = item.get(t)
        for i, metric_value in enumerate(metric_values):
            metric_values_list[i].append(metric_value)

    data = {}
    for dimension_name, dimension_values in zip( dimensions.keys(), dimension_values_list ):
        data[dimension_name] = dimension_values
    data["event_time"] = timestamp_list

    for metric_name, metric_values in zip( metrics, metric_values_list ):
        data[metric_name] = metric_values

    # convert dictionary of lists (containing dimension & metric values) to comma separated string records
    synthdata = []
    for k in zip(*data.values()):
        synthdata.append(",".join(map(str, k)))    
    return synthdata