Validating Oneway and Access Tags for Routing Jump to heading#

from __future__ import annotations

import logging
from collections.abc import Iterable, Mapping
from dataclasses import dataclass, field

import networkx as nx

logging.basicConfig(level=logging.INFO, format="%(levelname)s: %(message)s")
logger = logging.getLogger("osm.routing.access")

# Values the specification recognises. Anything else is a finding, not a guess.
ONEWAY_FORWARD = {"yes", "true", "1"}
ONEWAY_REVERSE = {"-1", "reverse"}
ONEWAY_NONE = {"no", "false", "0"}
ACCESS_DENY = {"no", "private", "agricultural", "forestry", "delivery",
               "customers", "permit"}
ACCESS_ALLOW = {"yes", "designated", "permissive", "destination", "official"}


@dataclass(frozen=True)
class Profile:
    """Which keys the router reads, in precedence order. Must match the router."""
    name: str
    access_keys: tuple[str, ...]          # most specific LAST
    honours_oneway: bool = True
    oneway_exception_key: str | None = None   # e.g. "oneway:bicycle"


CAR = Profile("car", ("access", "vehicle", "motor_vehicle", "motorcar"))
BIKE = Profile("bike", ("access", "vehicle", "bicycle"),
               oneway_exception_key="oneway:bicycle")
FOOT = Profile("foot", ("access", "foot"), honours_oneway=False)


@dataclass
class Findings:
    unparseable: list[tuple[int, str, str]] = field(default_factory=list)
    contradictory: list[tuple[int, str]] = field(default_factory=list)
    unreachable: list[int] = field(default_factory=list)
    inescapable: list[int] = field(default_factory=list)


def direction(tags: Mapping[str, str], profile: Profile
              ) -> tuple[bool, bool, str | None]:
    """Return (forward_allowed, backward_allowed, problem)."""
    if not profile.honours_oneway:
        return True, True, None

    value = tags.get("oneway", "no").strip().lower()
    if profile.oneway_exception_key:
        value = tags.get(profile.oneway_exception_key, value).strip().lower()

    if value in ONEWAY_NONE:
        return True, True, None
    if value in ONEWAY_FORWARD:
        return True, False, None
    if value in ONEWAY_REVERSE:
        return False, True, None
    # Never guess: an unrecognised value is a finding, and treating it as
    # bidirectional invents a route while treating it as oneway removes one.
    return True, True, f"unparseable oneway value {value!r}"


def accessible(tags: Mapping[str, str], profile: Profile
               ) -> tuple[bool, str | None]:
    """Apply the profile's keys in order; the most specific one present wins."""
    allowed: bool | None = None
    seen: list[tuple[str, str]] = []
    for key in profile.access_keys:
        value = tags.get(key)
        if value is None:
            continue
        value = value.strip().lower()
        seen.append((key, value))
        if value in ACCESS_DENY:
            allowed = False
        elif value in ACCESS_ALLOW:
            allowed = True
        else:
            return True, f"unparseable {key} value {value!r}"

    if len(seen) > 1 and len({v for _, v in seen}) > 1:
        # Not an error by itself — specific overrides general is the point —
        # but worth surfacing when the general key denies and no specific
        # key re-permits for THIS profile.
        general = dict(seen).get("access")
        if general in ACCESS_DENY and allowed is False:
            return False, None
    return (True if allowed is None else allowed), None


def build(ways: Iterable[tuple[int, list[int], Mapping[str, str]]],
          profile: Profile, findings: Findings) -> nx.DiGraph:
    graph = nx.DiGraph()
    for way_id, nodes, tags in ways:
        ok, problem = accessible(tags, profile)
        if problem:
            findings.unparseable.append((way_id, "access", problem))
        if not ok:
            continue
        forward, backward, problem = direction(tags, profile)
        if problem:
            findings.unparseable.append((way_id, "oneway", problem))
        for a, b in zip(nodes, nodes[1:]):
            if forward:
                graph.add_edge(a, b, way=way_id)
            if backward:
                graph.add_edge(b, a, way=way_id)
    return graph


def connectivity(graph: nx.DiGraph, findings: Findings,
                 min_component: int = 20) -> None:
    """A healthy car network is nearly one giant strongly connected component.

    Everything outside it is a genuine dead end or a directional bug, and the
    two are distinguished by whether the pocket has an undirected connection
    to the giant component that the directions have closed off.
    """
    components = sorted(nx.strongly_connected_components(graph),
                        key=len, reverse=True)
    if not components:
        return
    giant = components[0]
    logger.info("giant component holds %.2f%% of nodes",
                100 * len(giant) / graph.number_of_nodes())

    undirected = graph.to_undirected(as_view=True)
    for component in components[1:]:
        if len(component) < min_component:
            continue
        touches_giant = any(neighbour in giant
                            for node in component
                            for neighbour in undirected.neighbors(node))
        if not touches_giant:
            continue
        # Physically connected, directionally isolated: a oneway bug.
        into = any(graph.has_edge(n, m) for n in giant for m in component
                   if graph.has_edge(n, m))
        (findings.inescapable if into else findings.unreachable).extend(
            sorted(component)[:5])


if __name__ == "__main__":
    logger.info("validate against the profile the router actually uses")