Normalizing OSM Speed Limits and Units Jump to heading#

#!/usr/bin/env python3
"""Normalise OSM maxspeed values to km/h, keeping what cannot be a number."""
from __future__ import annotations

import logging
import re
from dataclasses import dataclass
from enum import Enum

logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s")
logger = logging.getLogger(__name__)

MPH_TO_KMH = 1.609344
KNOTS_TO_KMH = 1.852


class SpeedKind(str, Enum):
    EXPLICIT = "explicit"        # a number was stated
    IMPLICIT = "implicit"        # resolved from a country default
    CATEGORY = "category"        # none / walk / signals — no honest number
    UNKNOWN = "unknown"          # absent
    UNPARSEABLE = "unparseable"  # present, not understood


@dataclass(frozen=True)
class Speed:
    kmh: float | None
    kind: SpeedKind
    category: str | None = None
    raw: str | None = None


#: Values that deliberately are not numbers. Mapping them to one loses information.
CATEGORIES = {
    "none": "no_legal_limit",       # German autobahn — advisory 130, not a limit
    "walk": "walking_pace",
    "signals": "variable_signed",
    "variable": "variable_signed",
}

#: A small extract of the implicit-value table. The real one is maintained upstream
#: in the OSM wiki and should be loaded as data, not embedded in code.
IMPLICIT: dict[str, float] = {
    "DE:urban": 50, "DE:rural": 100, "DE:living_street": 7, "DE:motorway": 130,
    "GB:nsl_single": 60 * MPH_TO_KMH, "GB:nsl_dual": 70 * MPH_TO_KMH,
    "GB:motorway": 70 * MPH_TO_KMH,
    "FR:urban": 50, "FR:rural": 80, "FR:motorway": 130,
    "RO:urban": 50, "RO:rural": 90, "RO:motorway": 130,
    "AT:urban": 50, "AT:rural": 100, "AT:motorway": 130,
}

_NUMERIC = re.compile(r"""
    ^\s*
    (?P<value>\d+(?:\.\d+)?)        # the number
    \s*
    (?P<unit>km/h|kmh|kph|mph|knots|kn)?   # optional unit; absent means km/h
    \s*$
""", re.IGNORECASE | re.VERBOSE)

_UNIT_FACTOR = {
    None: 1.0, "": 1.0, "km/h": 1.0, "kmh": 1.0, "kph": 1.0,
    "mph": MPH_TO_KMH, "knots": KNOTS_TO_KMH, "kn": KNOTS_TO_KMH,
}

#: Above this, the value is a data error rather than a fast road.
MAX_PLAUSIBLE_KMH = 300.0


def normalise(raw: str | None, country: str | None = None) -> Speed:
    """Parse one maxspeed value. Never guesses; never returns a number it invented."""
    if raw is None or not raw.strip():
        return Speed(None, SpeedKind.UNKNOWN, raw=raw)

    value = raw.strip()
    lowered = value.lower()

    if lowered in CATEGORIES:
        return Speed(None, SpeedKind.CATEGORY, category=CATEGORIES[lowered], raw=raw)

    match = _NUMERIC.match(value)
    if match:
        unit = (match.group("unit") or "").lower()
        kmh = float(match.group("value")) * _UNIT_FACTOR[unit or None]
        if not 0 < kmh <= MAX_PLAUSIBLE_KMH:
            return Speed(None, SpeedKind.UNPARSEABLE, raw=raw)
        return Speed(round(kmh, 1), SpeedKind.EXPLICIT, raw=raw)

    # Implicit form: either a full "CC:type" token, or a bare type with a known country.
    key = value if ":" in value else (f"{country}:{value}" if country else None)
    if key and key in IMPLICIT:
        return Speed(round(IMPLICIT[key], 1), SpeedKind.IMPLICIT, raw=raw)

    return Speed(None, SpeedKind.UNPARSEABLE, raw=raw)


def normalise_many(values: list[tuple[str | None, str | None]]) -> list[Speed]:
    results = [normalise(raw, country) for raw, country in values]
    counts: dict[SpeedKind, int] = {kind: 0 for kind in SpeedKind}
    for speed in results:
        counts[speed.kind] += 1
    total = len(results) or 1
    for kind, n in counts.items():
        logger.info("%-12s %8d  %5.1f%%", kind.value, n, 100 * n / total)
    return results

def test_mph_is_converted_not_stripped():
    assert normalise("30 mph").kmh == 48.3
    assert normalise("70 mph").kmh == 112.7

def test_bare_number_is_kmh():
    assert normalise("50").kmh == 50.0

def test_none_is_a_category_not_a_number():
    speed = normalise("none")
    assert speed.kmh is None and speed.category == "no_legal_limit"

def test_implicit_needs_a_country():
    assert normalise("urban").kind is SpeedKind.UNPARSEABLE
    assert normalise("urban", country="DE").kmh == 50.0

def test_absurd_values_are_rejected_not_clamped():
    assert normalise("999").kind is SpeedKind.UNPARSEABLE

speeds = [s.kmh for s in results if s.kmh is not None]
import statistics
logger.info("median %.0f km/h, p95 %.0f km/h",
            statistics.median(speeds), sorted(speeds)[int(0.95 * len(speeds))])