def plot_graph()

in analytics/circleci_analyze.py [0:0]

• 40 lines of code
• 11 McCabe index (conditional complexity)

def plot_graph(name_filter: Optional[str] = None,
               output_file: Optional[str] = None,
               branch: Optional[str] = None) -> None:
    import matplotlib.pyplot as plt
    import matplotlib.dates as mdates

    ci_cache = CircleCICache(token=get_circleci_token())
    summary = ci_cache.get_jobs_summary()
    test_jobs = [name for name in summary.keys() if name.startswith('pytorch') and 'test' in name]
    filtered_jobs = filter_names(test_jobs, name_filter)
    prefix = common_prefix(filtered_jobs)
    if len(filtered_jobs) == 0:
        print(f'Filter "{name_filter}" does not match to any of {test_jobs}')
        return
    series = []
    labels = []
    styles = [f'{color}{style}' for (style, color) in itertools.product(['-', '--', '-.', ':'], ['b', 'g', 'r', 'c', 'm', 'y', 'k'])]
    fig, ax = plt.subplots()
    for name in test_jobs:
        label = f"{name}(p95 = {int(summary[name]['metrics']['duration_metrics']['p95']/60)} min)"
        if name not in filtered_jobs:
            print(label)
            continue
        ts = ci_cache.get_job_timeseries(name, branch=branch)
        if len(ts) == 0:
            print(f'{label} time series is empty!')
            continue
        print(f'{label} time series has {len(ts)} elements')
        labels.append(label[len(prefix):])
        series.append(ts)
        x, y = zip(*aggregate_by_day(ts))
        plt.plot(x, [i / 60.0 for i in y], styles[len(labels) % len(styles)])
    plt.legend(labels, loc='upper left')
    plt.title(f'{prefix} timeseries')
    ax.set_ylabel("Duration (m)")
    # Format date
    ax.xaxis.set_major_formatter(mdates.DateFormatter('%b %d'))
    # Rotate tick labels
    plt.setp(ax.get_xticklabels(), rotation=45, ha='right', rotation_mode='anchor')
    if output_file is not None:
        plt.savefig(output_file)
    else:
        plt.show()