ShitboxInfinity/tools/osm_junction.py

#!/usr/bin/env python3
"""Generate crash-junction events from real OSM intersections.

Usage: osm_junction.py <area.json> <out junctions.json> <lat0> <lon0> "Name|Street A|Street B" ...

For each spec: junction = shared node of the two streets. Every same-named way
passing within RADIUS of it becomes a convoy stream (clipped, plus a reversed
copy lane-offset the other way). Spawn point sits SPAWN_DIST up Street A aimed
at the junction. Coordinates are level-local Godot metres (x east, z south).

RADIUS is the whole event. Clip too wide and the convoy spreads thin over
hundreds of metres, so you arrive at the intersection through a 20 m gap and
score nothing -- "Margaret x Edward" was unwinnable that way, 27 convoy cars
alive and zero collisions. Keep the streams hugging the junction so the cars
are actually crossing when you get there.
"""
import json
import math
import sys
from itertools import zip_longest

RADIUS = 95.0        # convoy clip window around the junction
SPAWN_DIST = 120.0   # launch distance up street A
MIN_STREAM = 28.0    # shorter than this and a "stream" is a couple of parked cars
LANE = 2.7
TARGETS = [15000, 35000, 60000]

src, out = sys.argv[1], sys.argv[2]
LAT0, LON0 = float(sys.argv[3]), float(sys.argv[4])
M_LAT = 110574.0
M_LON = 111320.0 * math.cos(math.radians(LAT0))

def xz(lat, lon):
    return ((lon - LON0) * M_LON, -(lat - LAT0) * M_LAT)

data = json.load(open(src))
ways = {}
for e in data["elements"]:
    t = e.get("tags", {})
    if e["type"] == "way" and t.get("highway") and t.get("name") and e.get("geometry"):
        ways.setdefault(t["name"], []).append(e)

def offset(pts, d):
    outp = []
    for i, p in enumerate(pts):
        a = pts[max(i - 1, 0)]
        b = pts[min(i + 1, len(pts) - 1)]
        dx, dz = b[0] - a[0], b[1] - a[1]
        L = math.hypot(dx, dz) or 1.0
        outp.append([p[0] + dz / L * d, p[1] - dx / L * d])
    return outp

def launch_point(pts, jpt):
    """Walk street A outward from the junction to SPAWN_DIST, aimed back at it.

    Taking the farthest clipped point instead (the old way) tied the launch
    distance to RADIUS, so tightening the convoy window would have dragged the
    start line in on top of the intersection.
    """
    i = min(range(len(pts)), key=lambda k: math.hypot(pts[k][0] - jpt[0], pts[k][1] - jpt[1]))
    best = None
    for step in (1, -1):
        run, prev, k, end, hit = 0.0, pts[i], i + step, pts[i], False
        while 0 <= k < len(pts):
            seg = math.hypot(pts[k][0] - prev[0], pts[k][1] - prev[1])
            if run + seg >= SPAWN_DIST and seg > 0.01:
                # land exactly on SPAWN_DIST -- OSM segments can be hundreds of
                # metres, so stopping at the far vertex overshot by 100m+
                f = (SPAWN_DIST - run) / seg
                end = [prev[0] + (pts[k][0] - prev[0]) * f,
                       prev[1] + (pts[k][1] - prev[1]) * f]
                run, hit = SPAWN_DIST, True
                break
            run += seg
            prev, end, k = pts[k], pts[k], k + step
        if best is None or (hit and not best[3]) or (hit == best[3] and run > best[0]):
            best = (run, end, step, hit)
    run, end, step, hit = best
    if not hit and run > 1.0:
        # street ran out -- extend along its last heading so every event
        # launches from the same distance regardless of how OSM split the way
        j = max(min(i + step, len(pts) - 1), 0)
        dx, dz = end[0] - pts[j][0], end[1] - pts[j][1]
        L = math.hypot(dx, dz)
        if L > 0.01:
            grow = SPAWN_DIST - run
            end = [end[0] + dx / L * grow, end[1] + dz / L * grow]
    dx, dz = jpt[0] - end[0], jpt[1] - end[1]
    L = math.hypot(dx, dz) or 1.0
    return {"x": end[0], "z": end[1], "dx": dx / L, "dz": dz / L}

events = []
for spec in sys.argv[5:]:
    name, a, b = spec.split("|")[:3]
    na = {n for w in ways.get(a, []) for n in w["nodes"]}
    nb = {n for w in ways.get(b, []) for n in w["nodes"]}
    shared = na & nb
    if not shared:
        print("SKIP %s: %s never meets %s" % (name, a, b))
        continue
    # Two named streets often share several nodes (OSM splits ways at every
    # change of tag), and the lowest node id is arbitrary -- for Albert x
    # Adelaide it landed inside a CBD building block with the convoy 60 m away
    # and no drivable run-up. Pick the shared node with the most street geometry
    # around it: the busiest crossing is the best crash junction anyway.
    def node_pt(n):
        for street in (a, b):
            for w in ways[street]:
                if n in w["nodes"]:
                    g = w["geometry"][w["nodes"].index(n)]
                    return xz(g["lat"], g["lon"])
        return None

    def busyness(n):
        p = node_pt(n)
        if p is None:
            return -1
        return sum(1 for w in ways.get(a, []) + ways.get(b, [])
                   for g in w["geometry"]
                   if math.hypot(*[c - d for c, d in zip(xz(g["lat"], g["lon"]), p)]) < RADIUS)

    jn = max(sorted(shared), key=busyness)
    jpt = node_pt(jn)
    by_street = {a: [], b: []}
    spawn = None
    spawn_d = 1e9
    for street in (a, b):
        for w in ways[street]:
            pts = [xz(g["lat"], g["lon"]) for g in w["geometry"]]
            near = [p for p in pts if math.hypot(p[0] - jpt[0], p[1] - jpt[1]) < RADIUS]
            if len(near) < 2:
                continue
            length = sum(math.hypot(near[i + 1][0] - near[i][0], near[i + 1][1] - near[i][1])
                         for i in range(len(near) - 1))
            if length < MIN_STREAM:
                continue
            by_street[street].append({"pts": offset(near, LANE), "speed": 12.0, "interval": 1.5})
            by_street[street].append({"pts": offset(list(reversed(near)), LANE),
                                      "speed": 11.0, "interval": 1.8})
            if street == a:
                # OSM splits a street into many ways; only the one that actually
                # reaches the junction can define the launch, or we start 40 m
                # off the approach and never aim at the intersection
                d_near = min(math.hypot(p[0] - jpt[0], p[1] - jpt[1]) for p in pts)
                if d_near < spawn_d:
                    spawn, spawn_d = launch_point(pts, jpt), d_near
    # Interleave the two streets before capping. Appending A's streams then B's
    # and slicing [:6] could drop street B entirely -- leaving an "intersection"
    # with traffic only on the road you launch down, and nothing to hit.
    streams = []
    for pair in zip_longest(by_street[a], by_street[b]):
        for s in pair:
            if s is not None:
                streams.append(s)
    streams = streams[:6]
    if not streams or spawn is None:
        print("SKIP %s: no usable streams" % name)
        continue
    cross = sum(1 for s in streams if s in by_street[b])
    events.append({"name": name, "spawn": spawn, "streams": streams, "targets": TARGETS})
    print("OK %s: %d streams (%d crossing), spawn %.0fm out"
          % (name, len(streams), cross, math.hypot(spawn["x"] - jpt[0], spawn["z"] - jpt[1])))

json.dump(events, open(out, "w"), indent=1)
print("wrote %s (%d events)" % (out, len(events)))
if len(events) < len(sys.argv[5:]):
    sys.exit("FAILED: %d of %d specs produced nothing -- refusing to ship a level "
             "with missing junctions" % (len(sys.argv[5:]) - len(events), len(sys.argv[5:])))