2077-gps-mod/tools/summarize_gps_route_records.py

#!/usr/bin/env python3
"""Summarize EdgeWeightGPS route-result records from RED4ext logs."""

from __future__ import annotations

import argparse
import collections
import json
import re
from pathlib import Path
from typing import Any


RE_RECORD_BLOCK = re.compile(r"gpsResult_ptr28_rec40=\[(.*?)\]")
RE_RECORD = re.compile(r"(\d+):([0-9a-f]+),0x([0-9a-f]{8}),([^;\]]+)")
RE_JOB_COUNT = re.compile(r"GPSRouteJobBuild result call=(\d+).*?job_count50=(\d+)/0x([0-9a-f]+)")

FLAGS = {
    "FromRoadSpline": 1,
    "Bidirectional": 2,
    "PatrolRoute": 4,
    "Pavement": 8,
    "Road": 16,
    "Intersection": 32,
    "NeverDeadEnd": 64,
    "TrafficDisabled": 128,
    "CrossWalk": 256,
    "GPSOnly": 512,
    "ShowDebug": 1024,
    "Blockade": 2048,
    "Yield": 4096,
    "NoAIDriving": 8192,
    "Highway": 16384,
    "NoAutodrive": 32768,
}


def bytes_le(value: int) -> tuple[int, int, int, int]:
    return tuple((value >> (8 * index)) & 0xFF for index in range(4))


def flag_value(flags: Any) -> int:
    if isinstance(flags, dict):
        return int(flags.get("Value", flags.get("$value", 0)) or 0)
    return int(flags or 0)


def lane_category(flags: int) -> str:
    if flags & FLAGS["Highway"]:
        return "highway"
    if flags & FLAGS["GPSOnly"]:
        return "gpsonly"
    if flags & FLAGS["Road"]:
        return "road"
    if flags & FLAGS["Pavement"]:
        return "pavement"
    return "other"


def load_lane_lookup(path: Path | None) -> dict[int, dict[str, Any]]:
    if not path:
        return {}

    data = json.loads(path.read_text(encoding="utf-8"))
    lanes = data["Data"]["RootChunk"]["data"]["lanes"]
    lookup: dict[int, dict[str, Any]] = {}
    for index, lane in enumerate(lanes):
        node_hash = lane.get("nodeRefHash")
        if node_hash is None:
            continue
        flags = flag_value(lane.get("flags"))
        lookup[int(node_hash)] = {
            "index": index,
            "flags": flags,
            "category": lane_category(flags),
            "speed": flag_value(lane.get("maxSpeed")),
            "length": lane.get("length"),
        }
    return lookup


def summarize(path: Path, lane_lookup: dict[int, dict[str, Any]]) -> None:
    routes = 0
    records = 0
    first_byte = collections.Counter()
    byte_positions: list[collections.Counter[int]] = [collections.Counter() for _ in range(4)]
    patterns = collections.Counter()
    segments = collections.Counter()
    class_categories: dict[int, collections.Counter[str]] = collections.defaultdict(collections.Counter)
    class_flags: dict[int, collections.Counter[int]] = collections.defaultdict(collections.Counter)
    class_speed: dict[int, collections.Counter[int]] = collections.defaultdict(collections.Counter)
    matched_handles = 0
    unmatched_handles = 0
    job_counts: list[int] = []

    for line in path.read_text(encoding="utf-8", errors="replace").splitlines():
        job_match = RE_JOB_COUNT.search(line)
        if job_match:
            job_counts.append(int(job_match.group(2)))

        block_match = RE_RECORD_BLOCK.search(line)
        if not block_match:
            continue

        route_records = 0
        for record_match in RE_RECORD.finditer(block_match.group(1)):
            handle = record_match.group(2)
            packed = int(record_match.group(3), 16)
            pieces = record_match.group(4).split(",")
            if len(pieces) < 4:
                continue

            route_records += 1
            records += 1
            bytes_tuple = bytes_le(packed)
            class_id = bytes_tuple[0]
            patterns[bytes_tuple] += 1
            first_byte[class_id] += 1
            for index, byte in enumerate(bytes_tuple):
                byte_positions[index][byte] += 1
            segments[(handle, class_id)] += 1

            if lane_lookup:
                lane = lane_lookup.get(int(handle, 16))
                if lane:
                    matched_handles += 1
                    class_categories[class_id][lane["category"]] += 1
                    class_flags[class_id][lane["flags"]] += 1
                    class_speed[class_id][lane["speed"]] += 1
                else:
                    unmatched_handles += 1

        if route_records:
            routes += 1

    print(f"{path}:")
    print(f"  routes={routes} records={records}")
    if job_counts:
        sorted_counts = sorted(job_counts)
        p50 = sorted_counts[len(sorted_counts) // 2]
        p90 = sorted_counts[min(len(sorted_counts) - 1, int(len(sorted_counts) * 0.9))]
        avg = sum(sorted_counts) / len(sorted_counts)
        print(
            "  job_count50 "
            f"n={len(sorted_counts)} min={sorted_counts[0]} p50={p50} "
            f"p90={p90} max={sorted_counts[-1]} avg={avg:.1f}"
        )

    print("  low-byte class counts:", dict(first_byte.most_common()))
    if lane_lookup:
        print(f"  matched route records to lane hashes: {matched_handles}/{matched_handles + unmatched_handles}")
        print("  class -> resource categories:")
        for class_id, count in first_byte.most_common():
            categories = dict(class_categories[class_id].most_common())
            speeds = dict(class_speed[class_id].most_common(8))
            flags = dict(class_flags[class_id].most_common(8))
            print(f"    {class_id}: categories={categories} speeds={speeds} flags={flags}")
    for index, counter in enumerate(byte_positions):
        print(f"  byte{index} counts:", dict(counter.most_common(12)))
    print("  packed metadata patterns:")
    for pattern, count in patterns.most_common(24):
        print(f"    {pattern}: {count}")


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--traffic-json", type=Path)
    parser.add_argument("logs", nargs="+", type=Path)
    args = parser.parse_args()

    lane_lookup = load_lane_lookup(args.traffic_json)
    for path in args.logs:
        summarize(path, lane_lookup)
    return 0


if __name__ == "__main__":
    raise SystemExit(main())