tests/test_traffic.py

"""Tests for the NEMOCITY Python traffic mirror (static assignment).

Covers: mulberry32 cross-language equivalence (values pinned from
web/city/rng.js via node), trip-gen determinism, A* sanity over the bridge,
genesis bridge demand, fix candidates, the <2s perf budget, and the
ACCEPTANCE test: genesis + 3 extra west houses jams the Old Bridge
(demand ratio >= 1.0 at rush).
"""

import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parents[1]))

from engine import constants as C
from engine import traffic
from engine.city import CityState
from engine.genesis import genesis_features
from engine.world import World

NOW = C.CITY_EPOCH_S + 10_000
BRIDGE = ((8, 0), (9, 0))


def genesis_city() -> CityState:
    return CityState.from_events(World.load(genesis_features()).features)


def west_growth_world() -> World:
    """Genesis + 3 extra west houses on 1st Ave (the acceptance scenario)."""
    world = World.load(genesis_features())
    for i, (cx, cz) in enumerate(((4, -1), (5, -1), (6, -1))):
        feature, obs = world.apply(
            "w_000001", i,
            {"tool": "place_building", "args": {
                "kind": "house", "name": f"West House {i + 1}", "cx": cx, "cz": cz,
                "w": 1, "d": 1, "floors": 1, "hue": 30, "variant": i}},
            t=C.CITY_EPOCH_S + 60,
        )
        assert feature is not None, obs
    return world


# ------------------------------------------------------------------ mulberry32


def test_mulberry32_matches_js_reference():
    # Pinned from web/city/rng.js (node, June 12) — byte-stable contract.
    expected = {
        0: (0.26642920868471265, 0.0003297457005828619, 0.22327202744781971),
        1: (0.62707394058816135, 0.0027357211802154779, 0.52744703995995224),
        12345: (0.97972826776094735, 0.30675226449966431, 0.484205421525985),
        0x7FFFFFFF: (0.4290980885270983, 0.12713524978607893, 0.38527749828062952),
    }
    for seed, values in expected.items():
        rng = traffic.mulberry32(seed)
        for v in values:
            assert rng() == v


# -------------------------------------------------------------------- trip gen


def test_trip_table_deterministic():
    a = traffic.trip_table(genesis_city(), NOW)
    b = traffic.trip_table(genesis_city(), NOW)
    assert [(h.id, d.id) for h, d in a] == [(h.id, d.id) for h, d in b]
    assert len(a) == 20  # one trip per genesis resident


def test_trips_only_from_completed_buildings():
    city = genesis_city()
    # 1 s after epoch nothing is finished -> no homes emit commuters
    assert traffic.trip_table(city, C.CITY_EPOCH_S + 1) == []


# ------------------------------------------------------------------------- A*


def test_astar_crosses_the_only_bridge():
    city = genesis_city()
    net = traffic.RoadNet.from_city(city)
    west_home = next(b for b in city.buildings if b.name == "Elm Cottage")
    factory = next(b for b in city.buildings if b.kind == "factory")
    path = traffic._astar(net, {}, net.door(west_home), net.door(factory))
    assert path is not None
    assert any(cell in path for cell in BRIDGE)
    # 4-connected, all on roads
    for prev_cell, cell in zip(path, path[1:]):
        assert abs(prev_cell[0] - cell[0]) + abs(prev_cell[1] - cell[1]) == 1
        assert cell in net.cells


def test_assignment_deterministic():
    a1 = traffic.assign(genesis_city(), NOW)
    a2 = traffic.assign(genesis_city(), NOW)
    assert a1.demand == a2.demand
    assert a1.traffic_index == a2.traffic_index


def test_genesis_bridge_carries_demand_but_under_gate():
    a = traffic.assign(genesis_city(), NOW)
    ratios = [a.ratio(cell) for cell in BRIDGE]
    assert all(r > 0 for r in ratios)
    # the bottleneck-by-design sits NEAR the jam line but below the fix gate
    assert all(r < C.FIX_GATE_RATIO for r in ratios)


# ------------------------------------------------------------------ ACCEPTANCE


def test_acceptance_west_growth_jams_old_bridge():
    city = CityState.from_events(west_growth_world().features)
    a = traffic.assign(city, NOW)
    assert max(a.ratio(cell) for cell in BRIDGE) >= C.JAM_RATIO


# ------------------------------------------------------------------ candidates


def test_snapshot_stats_and_candidates():
    city = CityState.from_events(west_growth_world().features)
    stats, candidates = traffic.snapshot(city, NOW)
    assert stats["max_ratio"] >= C.FIX_GATE_RATIO
    assert stats["worst"] and stats["top"]
    assert all(t["street"] for t in stats["top"])
    assert 2 <= len(candidates) <= 4
    current = stats["traffic_index"]
    for i, cand in enumerate(candidates):
        assert cand["id"] == f"F{i + 1}"
        assert cand["action"] in ("new_road", "upgrade_avenue")
        assert cand["cells"] and cand["klass"] in C.ROAD_CLASSES
        assert cand["desc"]
        assert cand["predicted_index"] < current  # every fix actually helps
    # best predicted first (F1 is the engine fallback choice)
    predicted = [cand["predicted_index"] for cand in candidates]
    assert predicted == sorted(predicted)


def test_candidate_apply_improves_assignment():
    city = CityState.from_events(west_growth_world().features)
    base = traffic.assign(city, NOW)
    _, candidates = traffic.snapshot(city, NOW)
    fixed = traffic.assign(city, NOW, candidates[0])
    assert fixed.traffic_index < base.traffic_index
    assert fixed.traffic_index == candidates[0]["predicted_index"]


def test_upgrade_candidate_repaints_in_place():
    city = CityState.from_events(west_growth_world().features)
    _, candidates = traffic.snapshot(city, NOW)
    upgrades = [c for c in candidates if c["action"] == "upgrade_avenue"]
    for cand in upgrades:
        for cx, cz in cand["cells"]:
            assert (cx, cz) in city.roads
            assert city.roads[(cx, cz)].klass == "street"


# ------------------------------------------------------------------------ perf


def test_snapshot_under_two_seconds():
    city = CityState.from_events(west_growth_world().features)
    started = time.perf_counter()
    traffic.snapshot(city, NOW)
    assert time.perf_counter() - started < 2.0