def build_execution_summary()

in apps/cloudwatch-dashboard/lambdas/scheduler-last-execution-details/handler.py [0:0]

• 46 lines of code
• 8 McCabe index (conditional complexity)

def build_execution_summary(scheduler_name, width, height):
    list_executions = execution_summaries = list_inference_executions(
        scheduler_name,
        start_time=None,
        end_time=None,
        execution_status=None
    )
    
    # Loops through the executions summaries:
    results_json = []
    for execution_summary in list_executions:
        # We only request an output if the inference execution is a sucess:
        status = execution_summary['Status']
        if status == 'SUCCESS':
            # Download the JSON-line file locally:
            bucket = execution_summary['CustomerResultObject']['Bucket']
            key = execution_summary['CustomerResultObject']['Key']
            current_timestamp = key.split('/')[-2]
            local_fname = os.path.join('/tmp', f'{scheduler_name}_{current_timestamp}.jsonl')

            if not os.path.exists(local_fname):
                content_object = s3.Object(bucket, key)
                file_content = content_object.get()['Body'].read().decode('utf-8')
                with open(local_fname, 'w') as f:
                    f.write(file_content)

            # Opens the file and concatenate the results into a dataframe:
            with open(local_fname, 'r') as f:
                content = [eval(line) for line in f.readlines()]
                results_json = results_json + content

    # Build the final dataframes with all the results:
    results_df = pd.DataFrame(results_json)
    results_df['timestamp'] = pd.to_datetime(results_df['timestamp'])
    results_df = results_df.set_index('timestamp')
    results_df = results_df.sort_index()
    
    expanded_results = expand_signal_diagnostics(results_df)
    expanded_results = expanded_results[expanded_results['prediction'] == 1]
    
    if expanded_results.shape[0] > 0:
        event_details = pd.DataFrame(expanded_results.iloc[-1, 1:]).reset_index()

        event_details.columns = ['name', 'value']
        event_details = event_details.sort_values(by='value', ascending=False)
        event_details = event_details.iloc[:15, :].reset_index(drop=True)
        event_details = event_details.sort_values(by='value', ascending=True)
        
        title = f'Last event detected at {expanded_results.index[-1]}'
        html = plot_single_diagnostic(
            event_details, 
            len(expanded_results.columns) - 1, 
            title,
            width, 
            height
        )
        
    else:
        html = '<div>No anomaly detected by this scheduler yet</div>'
    
    return html