def find_issues()

in tools/analysis/ping-patterns/ping-patterns.py [0:0]

• 45 lines of code
• 18 McCabe index (conditional complexity)

def find_issues(client_data, stats):
    """
    Find and notate issues for a client's data.
    """
    client_data = sorted(client_data, key=lambda x: (x["end_time_local"], x["seq"]))
    last_ping = None
    last_by_type = {}
    client_stats = {}
    for ping in client_data:
        # Find zero-length pings
        if (
            ping["ping_type"] == "metrics"
            and ping["start_time_local"] == ping["end_time_local"]
        ):
            ping["notes"].add(ZERO_LENGTH)

        # Find multiple pings with the same end_time
        if last_ping is not None:
            if ping["ping_type"] == "metrics" and last_ping["ping_type"] == "metrics":
                if ping["end_time_local"] == last_ping["end_time_local"]:
                    ping["notes"].add(DUPLICATE_TIME)
                    last_ping["notes"].add(DUPLICATE_TIME)
                else:
                    ping["notes"].add(NO_BASELINE)

        # Find missing or duplicate seq numbers
        last_of_same_type = last_by_type.get(ping["ping_type"])
        if last_of_same_type is not None:
            if int(last_of_same_type["seq"]) + 1 != int(ping["seq"]):
                ping["notes"].add(MISSING_SEQ)
            elif int(last_of_same_type["seq"]) == int(ping["seq"]):
                ping["notes"].add(DUPLICATE_SEQ)
                last_of_same_type["notes"].add(DUPLICATE_SEQ)

        if ping["ping_type"] == "metrics":
            # Find metrics pings that are more than 24 hours after the last baseline ping
            last_baseline = last_by_type.get("baseline")
            if last_baseline is not None and ping["end_time_local"] > last_baseline[
                "end_time_local"
            ] + datetime.timedelta(days=1):
                ping["notes"].add(TOO_LATE)

            # Find metrics pings that are more than +/-1 hour from 4am
            if abs(
                ping["end_time_local"]
                - ping["end_time_local"].replace(hour=4, minute=0, second=0)
            ) > datetime.timedelta(hours=1):
                ping["notes"].add(FAR_FROM_4AM)

        # Add notes to the overall client stats
        for note in ping["notes"]:
            client_stats.setdefault(note, 0)
            client_stats[note] += 1

        last_ping = ping
        last_by_type[ping["ping_type"]] = ping

    # Add client stats to the overall stats
    for note in client_stats.keys():
        stats.setdefault(note, 0)
        stats[note] += 1

    return client_stats