Research GPS lane connection weighting
This commit is contained in:
Executable
+203
@@ -0,0 +1,203 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Summarize GPS-relevant structure in all.traffic_persistent JSON."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import collections
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
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 flag_value(flags: Any) -> int:
|
||||
if isinstance(flags, int):
|
||||
return flags
|
||||
if isinstance(flags, str):
|
||||
return int(flags)
|
||||
if isinstance(flags, dict):
|
||||
if "Value" in flags:
|
||||
return int(flags["Value"])
|
||||
if "$value" in flags:
|
||||
return int(flags["$value"])
|
||||
return int(flags or 0)
|
||||
|
||||
|
||||
def has(flags: int, name: str) -> bool:
|
||||
return bool(flags & FLAGS[name])
|
||||
|
||||
|
||||
def category(flags: int) -> str:
|
||||
if has(flags, "Highway"):
|
||||
return "highway"
|
||||
if has(flags, "GPSOnly"):
|
||||
return "gpsonly"
|
||||
if has(flags, "Road"):
|
||||
return "road"
|
||||
if has(flags, "Pavement"):
|
||||
return "pavement"
|
||||
return "other"
|
||||
|
||||
|
||||
def ref_index(ref: Any) -> int | None:
|
||||
if isinstance(ref, int):
|
||||
return ref
|
||||
if isinstance(ref, str):
|
||||
try:
|
||||
return int(ref)
|
||||
except ValueError:
|
||||
return None
|
||||
if isinstance(ref, dict):
|
||||
for key in ("Index", "index", "laneIndex", "LaneIndex", "value", "$value"):
|
||||
if key in ref:
|
||||
return ref_index(ref[key])
|
||||
return None
|
||||
|
||||
|
||||
def lane_refs(value: Any) -> list[int]:
|
||||
if value is None:
|
||||
return []
|
||||
if isinstance(value, list):
|
||||
out: list[int] = []
|
||||
for item in value:
|
||||
idx = ref_index(item)
|
||||
if idx is not None:
|
||||
out.append(idx)
|
||||
return out
|
||||
if isinstance(value, dict) and "Elements" in value:
|
||||
return lane_refs(value["Elements"])
|
||||
return []
|
||||
|
||||
|
||||
def gps_info(lane: dict[str, Any]) -> tuple[int | None, int | None]:
|
||||
info = lane.get("playerGPSInfo") or {}
|
||||
return info.get("subGraphId"), info.get("stronglyConnectedComponentId")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("json_file", type=Path)
|
||||
parser.add_argument("--lane-connections", type=Path)
|
||||
parser.add_argument("--samples", type=int, default=12)
|
||||
args = parser.parse_args()
|
||||
|
||||
data = json.loads(args.json_file.read_text(encoding="utf-8"))
|
||||
root = data["Data"]["RootChunk"]["data"]
|
||||
lanes = root["lanes"]
|
||||
neighbor_groups = root.get("neighborGroups", [])
|
||||
|
||||
categories = []
|
||||
flags_by_lane = []
|
||||
for lane in lanes:
|
||||
flags = flag_value(lane.get("flags", 0))
|
||||
flags_by_lane.append(flags)
|
||||
categories.append(category(flags))
|
||||
|
||||
print(f"lanes: {len(lanes)}")
|
||||
print(f"neighborGroups: {len(neighbor_groups)}")
|
||||
print("categories:", dict(collections.Counter(categories).most_common()))
|
||||
|
||||
gps_counter = collections.Counter(gps_info(lane) for lane in lanes)
|
||||
print(f"gps components: {len(gps_counter)}")
|
||||
print("top gps components:")
|
||||
for (subgraph, scc), count in gps_counter.most_common(20):
|
||||
cats = collections.Counter(
|
||||
categories[i] for i, lane in enumerate(lanes) if gps_info(lane) == (subgraph, scc)
|
||||
)
|
||||
print(f" subGraph={subgraph} scc={scc}: {count} {dict(cats.most_common())}")
|
||||
|
||||
connection_rows: list[dict[str, Any]] | None = None
|
||||
if args.lane_connections:
|
||||
connection_data = json.loads(args.lane_connections.read_text(encoding="utf-8"))
|
||||
connection_rows = connection_data["Data"]["RootChunk"]["data"]
|
||||
|
||||
transitions = collections.Counter()
|
||||
probabilities: dict[tuple[str, str], collections.Counter[int]] = collections.defaultdict(
|
||||
collections.Counter
|
||||
)
|
||||
sharp_angles = collections.Counter()
|
||||
out_degree = collections.Counter()
|
||||
missing_refs = 0
|
||||
rows = enumerate(lanes) if connection_rows is None else (
|
||||
(row["index"], row["value"]) for row in connection_rows
|
||||
)
|
||||
for i, lane_or_connections in rows:
|
||||
refs = (
|
||||
lane_refs(lane_or_connections.get("outLanes"))
|
||||
if connection_rows is None
|
||||
else lane_refs(lane_or_connections.get("outlanes"))
|
||||
)
|
||||
out_degree[len(refs)] += 1
|
||||
raw_outs = (
|
||||
lane_or_connections.get("outLanes", [])
|
||||
if connection_rows is None
|
||||
else lane_or_connections.get("outlanes", [])
|
||||
)
|
||||
for out, ref in zip(raw_outs, refs):
|
||||
if ref < 0 or ref >= len(lanes):
|
||||
missing_refs += 1
|
||||
continue
|
||||
pair = (categories[i], categories[ref])
|
||||
transitions[pair] += 1
|
||||
if isinstance(out, dict) and "exitProbabilityCompressed" in out:
|
||||
probabilities[pair][out["exitProbabilityCompressed"]] += 1
|
||||
if isinstance(out, dict) and "isSharpAngle" in out:
|
||||
sharp_angles[(pair, out["isSharpAngle"])] += 1
|
||||
print("out degree:", dict(sorted(out_degree.items())))
|
||||
print(f"missing out refs: {missing_refs}")
|
||||
print("category transitions:")
|
||||
for (src, dst), count in transitions.most_common(30):
|
||||
extra = ""
|
||||
if probabilities[(src, dst)]:
|
||||
extra = f" prob={probabilities[(src, dst)].most_common(6)}"
|
||||
sharp = [(key[1], value) for key, value in sharp_angles.items() if key[0] == (src, dst)]
|
||||
if sharp:
|
||||
extra += f" sharp={sharp}"
|
||||
print(f" {src:8s} -> {dst:8s}: {count}{extra}")
|
||||
|
||||
print("highway connector samples:")
|
||||
shown = 0
|
||||
for i, lane in enumerate(lanes):
|
||||
if categories[i] != "highway":
|
||||
continue
|
||||
if connection_rows is None:
|
||||
refs = [ref for ref in lane_refs(lane.get("outLanes")) if 0 <= ref < len(lanes)]
|
||||
else:
|
||||
row = connection_rows[i]["value"]
|
||||
refs = [ref for ref in lane_refs(row.get("outlanes")) if 0 <= ref < len(lanes)]
|
||||
ref_cats = collections.Counter(categories[ref] for ref in refs)
|
||||
if ref_cats and any(cat != "highway" for cat in ref_cats):
|
||||
subgraph, scc = gps_info(lane)
|
||||
print(
|
||||
f" lane={i} flags={flags_by_lane[i]} len={lane.get('length')} "
|
||||
f"speed={lane.get('maxSpeed')} gps=({subgraph},{scc}) out={dict(ref_cats)}"
|
||||
)
|
||||
shown += 1
|
||||
if shown >= args.samples:
|
||||
break
|
||||
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Executable
+132
@@ -0,0 +1,132 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Patch traffic lane connection probabilities toward highway routing."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import argparse
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
|
||||
FLAGS = {
|
||||
"Road": 16,
|
||||
"Pavement": 8,
|
||||
"TrafficDisabled": 128,
|
||||
"CrossWalk": 256,
|
||||
"GPSOnly": 512,
|
||||
"Blockade": 2048,
|
||||
"Highway": 16384,
|
||||
}
|
||||
|
||||
|
||||
def load_json(path: Path) -> Any:
|
||||
return json.loads(path.read_text(encoding="utf-8"))
|
||||
|
||||
|
||||
def flag_value(flags: Any) -> int:
|
||||
return int(flags or 0)
|
||||
|
||||
|
||||
def has(flags: int, name: str) -> bool:
|
||||
return bool(flags & FLAGS[name])
|
||||
|
||||
|
||||
def category(flags: int) -> str:
|
||||
if has(flags, "Highway"):
|
||||
return "highway"
|
||||
if has(flags, "GPSOnly"):
|
||||
return "gpsonly"
|
||||
if has(flags, "Road"):
|
||||
return "road"
|
||||
if has(flags, "Pavement"):
|
||||
return "pavement"
|
||||
return "other"
|
||||
|
||||
|
||||
def write_json(path: Path, data: Any) -> None:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(json.dumps(data, indent=2) + "\n", encoding="utf-8")
|
||||
|
||||
|
||||
def main() -> int:
|
||||
parser = argparse.ArgumentParser()
|
||||
parser.add_argument("traffic_persistent_json", type=Path)
|
||||
parser.add_argument("lane_connections_json", type=Path)
|
||||
parser.add_argument("out_json", type=Path)
|
||||
parser.add_argument("--high", type=int, default=255)
|
||||
parser.add_argument("--low", type=int, default=1)
|
||||
args = parser.parse_args()
|
||||
|
||||
traffic = load_json(args.traffic_persistent_json)
|
||||
lanes = traffic["Data"]["RootChunk"]["data"]["lanes"]
|
||||
lane_categories = [category(flag_value(lane.get("flags", 0))) for lane in lanes]
|
||||
|
||||
connections = load_json(args.lane_connections_json)
|
||||
rows = connections["Data"]["RootChunk"]["data"]
|
||||
|
||||
changed = 0
|
||||
high_set = 0
|
||||
low_set = 0
|
||||
touched_rows = 0
|
||||
|
||||
for row in rows:
|
||||
source_index = row["index"]
|
||||
source_category = lane_categories[source_index]
|
||||
outlanes = row["value"]["outlanes"]
|
||||
if not outlanes:
|
||||
continue
|
||||
|
||||
target_categories = [
|
||||
lane_categories[out["laneIndex"]]
|
||||
for out in outlanes
|
||||
if 0 <= out["laneIndex"] < len(lane_categories)
|
||||
]
|
||||
has_highway_target = "highway" in target_categories
|
||||
has_non_highway_target = any(cat != "highway" for cat in target_categories)
|
||||
row_changed = False
|
||||
|
||||
for out in outlanes:
|
||||
target_category = lane_categories[out["laneIndex"]]
|
||||
old = out["exitProbabilityCompressed"]
|
||||
new = old
|
||||
|
||||
if target_category == "highway" and source_category in {"road", "gpsonly", "highway"}:
|
||||
new = args.high
|
||||
elif (
|
||||
source_category == "highway"
|
||||
and has_highway_target
|
||||
and target_category != "highway"
|
||||
):
|
||||
new = args.low
|
||||
elif (
|
||||
source_category in {"road", "gpsonly"}
|
||||
and has_highway_target
|
||||
and has_non_highway_target
|
||||
and target_category != "highway"
|
||||
):
|
||||
new = args.low
|
||||
|
||||
if new != old:
|
||||
out["exitProbabilityCompressed"] = new
|
||||
changed += 1
|
||||
row_changed = True
|
||||
if new == args.high:
|
||||
high_set += 1
|
||||
elif new == args.low:
|
||||
low_set += 1
|
||||
|
||||
if row_changed:
|
||||
touched_rows += 1
|
||||
|
||||
write_json(args.out_json, connections)
|
||||
print(f"rows seen: {len(rows)}")
|
||||
print(f"rows touched: {touched_rows}")
|
||||
print(f"edges changed: {changed}")
|
||||
print(f"edges set high: {high_set}")
|
||||
print(f"edges set low: {low_set}")
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
Reference in New Issue
Block a user