Correlate GPS route records with lane resources

tools/summarize_gps_route_records.py

@@ -0,0 +1,178 @@
+#!/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())