crunch_the_numbers.py

import json
import pathlib
from datetime import datetime


def key(job_id: str, op: str):
    return f"{job_id}-{op}"


if __name__ == "__main__":
    JOB_TO_CONSIDER = 589

    csv_path = pathlib.Path("/Users/laser/Desktop/cnc_program_execution.csv")

    with csv_path.open("r") as f:
        lines = f.readlines()

    to_consider = []

    for line in lines[1:]:
        xs = [splitted.strip("'").replace("\n", "") for splitted in line.split(",")]

        (
            id,
            timestamp,
            flow_job_id,
            machine_op,
            is_fixture,
            flow_operation_id,
            event_type,
            machine,
            created_at,
            updated_at,
            is_recovery,
        ) = xs

        record = {
            "id": id,
            "machine": machine,
            "timestamp": round(datetime.strptime(timestamp, "%Y-%m-%d %H:%M:%S.%f+00").timestamp()),
            "flow_job_id": flow_job_id,
            "machine_op": machine_op,
            "is_fixture": True if is_fixture == "true" else False,
            "event_type": event_type,
            "is_recovery": True if is_recovery == "true" else False,
        }

        if not record["is_recovery"] and not record["is_fixture"]:
            if record["flow_job_id"] == f"{JOB_TO_CONSIDER}":
                to_consider.append(record)

    sorted_records = sorted(to_consider, key=lambda obj: obj["timestamp"])

    output = {}

    for record in sorted_records:
        k = key(record["flow_job_id"], record["machine_op"])
        if k not in output:
            output[k] = {"tmp": None, "cycle_time_seconds": []}
            if record["event_type"] == "program_start":
                output[k]["tmp"] = record["timestamp"]
            else:
                print(f"ignoring {record['id']}: wrong 'event_type' when seeing {k} for first time")
        else:
            if output[k]["tmp"] is None:
                if record["event_type"] == "program_start":
                    output[k]["tmp"] = record["timestamp"]
                else:
                    print(f"ignoring {record['id']}: tmp is not set, but 'event_type' was not 'program_start'")
                    pass  # ignore
            else:
                if record["event_type"] == "program_end":
                    output[k]["cycle_time_seconds"].append(record["timestamp"] - output[k]["tmp"])
                    output[k]["tmp"] = None
                else:
                    output[k]["tmp"] = record["timestamp"]

    print(json.dumps(output))