foodie-server/specs/032-self-hosted-map-integration/evaluation/run_model_probe.py

"""Reproduce the approved model assessment without changing production files.

Uses the existing SSH alias for read-only baseline inspection and disposable,
resource-limited OSRM containers. Synthetic XML and candidate Lua are generated
only under an owned /tmp directory. No business database or credentials are read.
"""
import hashlib
import json
from pathlib import Path
import subprocess
import sys
import urllib.request

VALHALLA_SHA = "34639a33510eb0764da77650e91f6efbb70e8daa"
MAP_REPO = Path("D:/QtwCode/foodie/Foodie_HCM_Map")
OUTPUT = Path(__file__).with_name("model-probe-results.json")


def remote_probe():
    """This function executes on the SSH host, with the standard library only."""
    import datetime
    import math
    import tempfile
    import time
    import urllib.error
    import urllib.parse
    import xml.etree.ElementTree as ET

    image = "ghcr.io/project-osrm/osrm-backend:v5.27.1"
    release = Path("/opt/foodie-map-tw-20260909")
    production = ["foodie-map-tw-20260909-api-1", "foodie-map-tw-20260909-osrm-1",
                  "foodie-map-tw-20260909-osrm-motorcycle-1"]

    def command(args):
        completed = subprocess.run(args, stdout=subprocess.PIPE, stderr=subprocess.STDOUT, timeout=90)
        if completed.returncode:
            raise RuntimeError("Command failed: " + " ".join(args) + "\n" + completed.stdout.decode(errors="replace")[-4000:])
        return completed.stdout

    def fingerprints():
        files = ["map_api/settings.py", "map_api/routing.py", "docker/osrm/profiles/motorcycle.lua"]
        hashes = {p: hashlib.sha256((release / p).read_bytes()).hexdigest() for p in files}
        for p in ("settings.py", "routing.py"):
            data = command(["docker", "exec", production[0], "cat", "/app/" + p])
            hashes["container:/app/" + p] = hashlib.sha256(data).hexdigest()
        hashes["container:car.lua"] = hashlib.sha256(command(
            ["docker", "exec", production[2], "cat", "/opt/car.lua"])).hexdigest()
        hashes["built_profile"] = command(["docker", "exec", production[2], "cat",
                                          "/data/profile.sha256"]).decode().strip()
        hashes["pbf"] = command(["docker", "exec", production[2], "cat",
                                "/data/build.sha256"]).decode().strip()
        return hashes

    def read_url(url):
        with urllib.request.urlopen(url, timeout=8) as response:
            return json.load(response)

    result = {"timestamp": datetime.datetime.now(datetime.timezone.utc).isoformat(),
              "image": image, "release": str(release), "before": fingerprints(),
              "cases": [], "production_probes": {}, "errors": []}
    baseline = (release / "docker/osrm/profiles/motorcycle.lua").read_text()
    car_lua = command(["docker", "exec", production[2], "cat", "/opt/car.lua"])
    result["combined_profile_matches_graph"] = hashlib.sha256(
        car_lua + baseline.encode()).hexdigest() == result["before"]["built_profile"]
    for name in ("config", "health/ready"):
        result["production_probes"][name] = read_url("http://127.0.0.1:8081/api/" + name)
    points = "pickup_lat=25.0330&pickup_lng=121.5654&drop_lat=25.0375&drop_lng=121.5637"
    for profile in ("car", "motorcycle"):
        for endpoint, params in (("route", points),
                                 ("route/multi", "coords=25.0330,121.5654;25.0375,121.5637")):
            route = read_url("http://127.0.0.1:8081/api/" + endpoint + "?" + params + "&profile=" + profile)
            result["production_probes"][endpoint + ":" + profile] = {
                k: route.get(k) for k in ("profile", "distance_meters", "duration_seconds", "traffic_aware")}

    # Probe chosen by this assessment: keep access restrictions; penalize gravel
    # through rate (not speed), and add a signal weight (not duration). Neither
    # number is a copied Valhalla constant or an approved production calibration.
    candidate = baseline.replace("    car.process_way(profile, proxy, result, relations)", """
    car.process_way(profile, proxy, result, relations)
    if way:get_value_by_key('surface') == 'gravel' then
        if result.forward_rate and result.forward_rate > 0 then
            result.forward_rate = result.forward_rate * 0.2
        end
        if result.backward_rate and result.backward_rate > 0 then
            result.backward_rate = result.backward_rate * 0.2
        end
    end""").replace("return {setup=setup, process_way=process_way, process_node=car.process_node, process_turn=car.process_turn}", """
local function process_turn(profile, turn)
    car.process_turn(profile, turn)
    if turn.has_traffic_light then turn.weight = turn.weight + 8 end
    if math.abs(turn.angle) > 60 then turn.weight = turn.weight + 3 end
end
return {setup=setup, process_way=process_way, process_node=car.process_node, process_turn=process_turn}
""")
    if candidate == baseline or "turn.weight = turn.weight + 8" not in candidate:
        raise RuntimeError("Unexpected baseline shape; no automatic production adaptation.")
    result["candidate_sha256"] = hashlib.sha256(candidate.encode()).hexdigest()

    root = ET.Element("osm", version="0.6", generator="foodie-isolated-model-assessment")
    counter = 0
    samples = {}

    def node(lat, lon, tags=None):
        nonlocal counter
        counter += 1
        element = ET.SubElement(root, "node", id=str(counter), lat=str(lat), lon=str(lon), version="1")
        for key, value in (tags or {}).items():
            ET.SubElement(element, "tag", k=key, v=value)
        return counter

    def way(ids, tags):
        nonlocal counter
        counter += 1
        element = ET.SubElement(root, "way", id=str(counter), version="1")
        for identifier in ids:
            ET.SubElement(element, "nd", ref=str(identifier))
        for key, value in tags.items():
            ET.SubElement(element, "tag", k=key, v=value)
        return counter

    # Independent diamonds ensure a prohibited short road has a real legal
    # detour. Inputs use exact shared junctions, avoiding arbitrary snap choices.
    def diamond(name, tags, width=0.004, signal=False, detour=True):
        lat, lon = 25.1 + len(samples) * 0.01, 121.5
        a, b = node(lat, lon), node(lat + 0.002, lon)
        mid = node(lat + 0.001, lon, {"highway": "traffic_signals"} if signal else {})
        way([a, mid, b], {"highway": "residential", "name": name, **tags})
        if detour:
            c, d = node(lat, lon + width), node(lat + 0.002, lon + width)
            way([a, c, d, b], {"highway": "residential", "name": name + "_detour"})
        if signal:
            branch = node(lat + 0.001, lon - 0.0003)
            way([mid, branch], {"highway": "residential", "name": "signal_branch"})
        samples[name] = [(lon, lat), (lon, lat + 0.002)]

    diamond("motorcycle_no", {"motorcycle": "no"})
    diamond("motorcar_no", {"motorcar": "no", "motorcycle": "yes"})
    for road in ("motorway", "motorway_link", "trunk"):
        diamond(road, {"highway": road})
    diamond("trunk_permitted", {"highway": "trunk", "motorcycle": "yes"})
    diamond("motorroad", {"motorroad": "yes"})
    diamond("motorroad_permitted", {"motorroad": "yes", "motorcycle": "yes"})
    diamond("oneway", {"oneway": "yes"})
    diamond("oneway_override", {"oneway": "yes", "oneway:motorcycle": "no"})
    diamond("speed_override", {"maxspeed:forward": "90", "maxspeed:motorcycle": "20"}, detour=False)
    diamond("gravel_choice", {"highway": "primary", "surface": "gravel"}, width=0.001)
    diamond("gravel_forced", {"highway": "primary", "surface": "gravel"}, detour=False)
    diamond("signals", {}, signal=True, detour=False)
    diamond("conditional_access", {"motorcycle:conditional": "no @ (00:00-24:00)"})

    # Turn relation distinguishes real path exclusion from simply assigning a
    # higher turn penalty. No two-stage-turn claim is made by this fixture.
    def turn_case(name, restriction_key):
        nonlocal counter
        lat, lon = 25.1 + len(samples) * 0.01, 121.5
        a, via, b = node(lat, lon), node(lat + 0.001, lon), node(lat + 0.001, lon + 0.001)
        c, d = node(lat + 0.003, lon), node(lat + 0.003, lon + 0.001)
        from_way = way([a, via], {"highway": "residential"})
        to_way = way([via, b], {"highway": "residential"})
        way([via, c, d, b], {"highway": "residential"})
        if restriction_key:
            counter += 1
            relation = ET.SubElement(root, "relation", id=str(counter), version="1")
            for kind, ref, role in (("way", from_way, "from"), ("node", via, "via"), ("way", to_way, "to")):
                ET.SubElement(relation, "member", type=kind, ref=str(ref), role=role)
            ET.SubElement(relation, "tag", k="type", v="restriction")
            ET.SubElement(relation, "tag", k=restriction_key, v="no_right_turn")
        samples[name] = [(lon, lat + 0.0001), (lon + 0.0009, lat + 0.001)]

    turn_case("motorcycle_turn", "restriction:motorcycle")
    turn_case("generic_turn", "restriction")
    turn_case("unrestricted_turn", None)
    # libosmium expects ordered entity types even though generated scenarios
    # were appended independently. Keep nodes before ways before relations.
    entities = sorted(list(root), key=lambda e: ({"node": 0, "way": 1, "relation": 2}[e.tag], int(e.attrib["id"])))
    root[:] = entities
    fixture = ET.tostring(root, encoding="utf-8", xml_declaration=True)
    samples["offroad"] = [samples["motorcycle_no"][0], (121.7, 25.7)]
    samples["disconnected"] = [samples["motorcycle_no"][0], samples["trunk_permitted"][1]]
    result["fixture_sha256"] = hashlib.sha256(fixture).hexdigest()
    containers = []
    urls = {}

    def progress(message):
        print(message, file=sys.stderr, flush=True)

    def route(profile, case, reverse=False):
        coords = samples[case][::-1] if reverse else samples[case]
        coordinates = ";".join(f"{lon},{lat}" for lon, lat in coords)
        url = urls[profile] + "/route/v1/driving/" + coordinates + "?overview=full&geometries=geojson&radiuses=10;10"
        try:
            data = read_url(url)
        except urllib.error.HTTPError as error:
            data = json.load(error)
        data["test_profile"] = profile
        return data

    def check(name, condition, observations, kind="assertion"):
        row = {"name": name, "kind": kind, "passed": bool(condition), "observations": observations}
        result["cases"].append(row)
        progress(f"{name}: {'PASS' if condition else 'FAIL'}")

    def metric(data):
        if data.get("code") != "Ok":
            return {"code": data.get("code")}
        return {k: data["routes"][0][k] for k in ("distance", "duration", "weight")}

    def distance(data):
        return data["routes"][0]["distance"] if data.get("code") == "Ok" else math.inf

    try:
        with tempfile.TemporaryDirectory(prefix="foodie-model-eval-", dir="/tmp") as temp:
            workspace = Path(temp)
            result["temporary_workspace"] = str(workspace)
            for profile, lua in (("car", None), ("baseline", baseline), ("candidate", candidate)):
                folder = workspace / profile
                folder.mkdir()
                (folder / "fixture.osm").write_bytes(fixture)
                if lua:
                    (folder / "model.lua").write_text(lua)
                profile_path = "/data/model.lua" if lua else "/opt/car.lua"
                base_args = ["docker", "run", "--rm", "--cpus=0.5", "--memory=256m",
                             "-v", f"{folder}:/data", "--entrypoint", "sh", image]
                progress("Building isolated " + profile)
                build = "osrm-extract --threads 1 -p " + profile_path + " /data/fixture.osm && osrm-partition --threads 1 /data/fixture.osrm && osrm-customize --threads 1 /data/fixture.osrm"
                command(base_args[:3] + ["--network", "none"] + base_args[3:] + ["-c", build])
                identifier = command(["docker", "run", "-d", "--rm", "--cpus=0.5", "--memory=256m",
                                      "-p", "127.0.0.1::5000", "-v", f"{folder}:/data:ro", image,
                                      "osrm-routed", "--algorithm", "mld", "--threads", "1", "/data/fixture.osrm"]).decode().strip()
                containers.append(identifier)
                info = json.loads(command(["docker", "inspect", identifier]))[0]
                port = info["NetworkSettings"]["Ports"]["5000/tcp"][0]["HostPort"]
                urls[profile] = "http://127.0.0.1:" + port
                for _ in range(30):
                    try:
                        read_url(urls[profile] + "/nearest/v1/driving/121.5,25.1")
                        break
                    except (urllib.error.URLError, TimeoutError):
                        time.sleep(0.2)
                else:
                    raise RuntimeError("Isolated engine did not start")

            for case, blocked in (("motorcycle_no", True), ("motorcar_no", False),
                                  ("motorway", True), ("motorway_link", True), ("trunk", True),
                                  ("trunk_permitted", False), ("motorroad", True),
                                  ("motorroad_permitted", False)):
                for profile in ("baseline", "candidate"):
                    data = route(profile, case)
                    ok = data.get("code") == "Ok" and (distance(data) > 700 if blocked else distance(data) < 300)
                    check(profile + ":" + case, ok, metric(data))
            car, moto = route("car", "motorcycle_no"), route("baseline", "motorcycle_no")
            check("car_vs_motorcycle_access", distance(car) < 300 < distance(moto),
                  {"car": metric(car), "motorcycle": metric(moto)})
            for profile in ("baseline", "candidate"):
                forward, backward = route(profile, "oneway"), route(profile, "oneway", True)
                check(profile + ":oneway", distance(forward) < 300 and distance(backward) > 700,
                      {"forward": metric(forward), "reverse": metric(backward)})
                override = route(profile, "oneway_override", True)
                check(profile + ":oneway_motorcycle_override", distance(override) < 300, metric(override))
                for turn in ("motorcycle_turn", "generic_turn"):
                    data = route(profile, turn)
                    check(profile + ":" + turn, distance(data) > 500 and data.get("code") == "Ok", metric(data))
            speed_car, speed_moto = route("car", "speed_override"), route("baseline", "speed_override")
            check("motorcycle_speed_limit", metric(speed_moto).get("duration", 0) > metric(speed_car).get("duration", 0),
                  {"car": metric(speed_car), "motorcycle": metric(speed_moto)})
            for case in ("gravel_forced", "signals", "unrestricted_turn"):
                base, proposed = route("baseline", case), route("candidate", case)
                base_metric, proposed_metric = metric(base), metric(proposed)
                ok = (base.get("code") == proposed.get("code") == "Ok" and
                      abs(base_metric["duration"] - proposed_metric["duration"]) < 0.2 and
                      proposed_metric["weight"] > base_metric["weight"])
                check(case + ":weight_not_eta", ok, {"baseline": base_metric, "candidate": proposed_metric})
            base, proposed = route("baseline", "gravel_choice"), route("candidate", "gravel_choice")
            check("gravel_preference_changes_path", distance(base) < 300 and distance(proposed) > 350,
                  {"baseline": metric(base), "candidate": metric(proposed)})
            for profile in ("baseline", "candidate"):
                for case, expected in (("offroad", "NoSegment"), ("disconnected", "NoRoute")):
                    data = route(profile, case)
                    check(profile + ":" + case, data.get("code") == expected, metric(data))
            conditional = route("baseline", "conditional_access")
            # A diagnostic, not a hidden assertion that unsupported time rules work.
            check("conditional_access_not_enforced", distance(conditional) < 300, metric(conditional), "limitation_probe")
            repeated = [*samples["motorcycle_no"], samples["motorcycle_no"][0]]
            coords = ";".join(f"{lon},{lat}" for lon, lat in repeated)
            multi = read_url(urls["baseline"] + "/route/v1/driving/" + coords + "?overview=false")
            check("ordered_three_stop_route", multi.get("code") == "Ok" and len(multi["routes"][0]["legs"]) == 2,
                  {"legs": len(multi.get("routes", [{}])[0].get("legs", [])), **metric(multi)})
    except Exception as error:
        result["errors"].append(str(error))
    finally:
        for identifier in containers:
            # Only identifiers created by this run are stopped; no Compose
            # lifecycle command can accidentally target a production service.
            subprocess.run(["docker", "stop", "--time", "1", identifier], capture_output=True, timeout=15)
        result["after"] = fingerprints()
        result["production_files_unchanged"] = result["before"] == result["after"]
        result["production_containers_running"] = all(
            command(["docker", "inspect", "--format", "{{.State.Running}}", name]).decode().strip() == "true"
            for name in production)
        result["temporary_containers_removed"] = all(
            subprocess.run(["docker", "inspect", identifier], capture_output=True).returncode != 0
            for identifier in containers)
        result["temporary_workspace_removed"] = not Path(result.get("temporary_workspace", "/nonexistent")).exists()
    print(json.dumps(result, ensure_ascii=False))


def main():
    import inspect
    sources = {}
    for path in ("src/sif/motorcyclecost.cc", "src/sif/motorscootercost.cc", "LICENSE.md", "COPYING"):
        url = f"https://raw.githubusercontent.com/valhalla/valhalla/{VALHALLA_SHA}/{path}"
        with urllib.request.urlopen(url, timeout=20) as response:
            data = response.read()
        sources[path] = {"url": url, "sha256": hashlib.sha256(data).hexdigest(), "bytes": len(data)}
    refs = {}
    for ref in ("HEAD", "origin/taiwan-map-profiles"):
        refs[ref] = subprocess.check_output(["git", "rev-parse", ref], cwd=MAP_REPO).decode().strip()
    lua = subprocess.check_output(["git", "show", "origin/taiwan-map-profiles:docker/osrm/profiles/motorcycle.lua"], cwd=MAP_REPO)
    branch_files = {}
    for ref in refs:
        branch_files[ref] = {}
        for path in ("map_api/settings.py", "map_api/routing.py", "docker/osrm/profiles/motorcycle.lua"):
            snapshot = subprocess.run(["git", "show", ref + ":" + path], cwd=MAP_REPO,
                                      stdout=subprocess.PIPE, stderr=subprocess.DEVNULL)
            branch_files[ref][path] = hashlib.sha256(snapshot.stdout).hexdigest() if snapshot.returncode == 0 else None
    references = {"valhalla_commit": VALHALLA_SHA, "sources": sources, "map_refs": refs, "branch_files": branch_files,
                  "branch_lua_sha256": hashlib.sha256(lua).hexdigest()}
    source = "import hashlib,json,subprocess,sys,urllib.request\nfrom pathlib import Path\n" + inspect.getsource(remote_probe) + "\nremote_probe()\n"
    process = subprocess.run(["ssh", "-o", "BatchMode=yes", "-o", "ConnectTimeout=8", "tokyo-map", "python3 -"],
                             input=source, text=True, encoding="utf-8", stdout=subprocess.PIPE, timeout=360)
    if process.returncode:
        raise RuntimeError("Remote assessment failed before producing a complete result")
    result = json.loads(process.stdout)
    result["references"] = references
    result["branch_lua_matches_deployment"] = references["branch_lua_sha256"] == result["before"]["docker/osrm/profiles/motorcycle.lua"]
    # Generated machine evidence, not a manually edited source/configuration.
    OUTPUT.write_text(json.dumps(result, ensure_ascii=False, indent=2) + "\n", encoding="utf-8")
    failures = [row["name"] for row in result["cases"] if not row["passed"]]
    print(json.dumps({"cases": len(result["cases"]), "failures": failures, "errors": result["errors"],
                      "files_unchanged": result["production_files_unchanged"], "evidence": str(OUTPUT)}, ensure_ascii=False))
    if failures or result["errors"] or not all(result[key] for key in
            ("combined_profile_matches_graph", "branch_lua_matches_deployment", "production_files_unchanged",
             "production_containers_running", "temporary_containers_removed", "temporary_workspace_removed")):
        sys.exit(1)


if __name__ == "__main__":
    main()