rules.py

"""The rule schema: what a San Francisco parking sign stack can say, as data.

Design notes:
- A pole holds a stack of Restriction objects (top to bottom matters for rendering,
  not for semantics).
- Answering "can I park here at time T?" applies every restriction independently;
  the most severe applicable verdict wins (TOW > NO_PARK > NO_STOP > TIME_LIMIT > FREE).
- v1 scope: sign-stack-only. Curb paint and meters are out of frame; queries that
  depend on them must be answered ABSTAIN by a correct model.
"""

from __future__ import annotations

from dataclasses import dataclass, field
from datetime import datetime, time
from enum import Enum


class Day(Enum):
    MON, TUE, WED, THU, FRI, SAT, SUN = range(7)


WEEKDAYS = frozenset({Day.MON, Day.TUE, Day.WED, Day.THU, Day.FRI})
EVERY_DAY = frozenset(Day)


class Kind(Enum):
    NO_STOPPING = "no_stopping"            # CA R26(S): no stopping at any time in window
    NO_PARKING = "no_parking"              # CA R26: no parking in window
    TOW_AWAY = "tow_away"                  # modifier or standalone: violation = tow
    TIME_LIMIT = "time_limit"              # CA R30: e.g. 2-hour parking 9am-6pm
    STREET_CLEANING = "street_cleaning"    # CA R32: no parking, specific weekday window
    PERMIT_EXEMPT_LIMIT = "permit_limit"   # RPP: time limit EXCEPT vehicles with area permit
    LOADING_ONLY = "loading_only"          # passenger/commercial loading zone window
    ANGLE_PARKING = "angle_parking"        # informational: "PARK AT 90 DEGREES" etc. No verdict effect.


@dataclass(frozen=True)
class Window:
    """A recurring weekly time window, e.g. Tue 8:00-10:00."""
    days: frozenset[Day]
    start: time
    end: time
    weeks: frozenset[int] = frozenset()  # which weeks of the month (1-5); empty = every week

    def contains(self, dt: datetime) -> bool:
        if Day(dt.weekday()) not in self.days or not (self.start <= dt.time() < self.end):
            return False
        if self.weeks:  # "2nd & 4th Monday" style: nth occurrence of the weekday in the month
            week_of_month = (dt.day - 1) // 7 + 1
            if week_of_month not in self.weeks:
                return False
        return True


@dataclass(frozen=True)
class Restriction:
    kind: Kind
    window: Window
    limit_minutes: int | None = None    # TIME_LIMIT / PERMIT_EXEMPT_LIMIT
    permit_area: str | None = None      # RPP area letter, e.g. "S"
    tow: bool = False                   # tow-away enforcement on violation


@dataclass
class SignStack:
    """Everything on one pole."""
    restrictions: list[Restriction] = field(default_factory=list)
    pole_id: str | None = None


class Verdict(Enum):
    TOW_RISK = "tow_risk"           # parking now risks a tow
    NO = "no"                       # no parking (citation risk)
    LIMITED = "limited"             # ok up to N minutes
    OK = "ok"                       # no restriction applies right now
    ABSTAIN = "abstain"             # not decidable from the sign stack alone


@dataclass
class Answer:
    verdict: Verdict
    limit_minutes: int | None = None   # for LIMITED
    until: datetime | None = None      # next moment the verdict changes (v2)
    reason: str = ""


SEVERITY = [Verdict.TOW_RISK, Verdict.NO, Verdict.LIMITED, Verdict.OK]


def can_park(stack: SignStack, when: datetime, permit_areas: frozenset[str] = frozenset()) -> Answer:
    """The resolver: ground truth for every generated question."""
    verdicts: list[Answer] = []
    for r in stack.restrictions:
        if r.kind is Kind.ANGLE_PARKING:
            continue  # informational ("park at 90 degrees"), never affects the verdict
        if not r.window.contains(when):
            continue
        if r.kind in (Kind.NO_STOPPING, Kind.NO_PARKING, Kind.STREET_CLEANING, Kind.LOADING_ONLY):
            v = Verdict.TOW_RISK if (r.tow or r.kind is Kind.NO_STOPPING) else Verdict.NO
            verdicts.append(Answer(v, reason=f"{r.kind.value} in effect"))
        elif r.kind is Kind.TIME_LIMIT:
            verdicts.append(Answer(Verdict.LIMITED, limit_minutes=r.limit_minutes,
                                   reason=f"{r.limit_minutes}min limit"))
        elif r.kind is Kind.PERMIT_EXEMPT_LIMIT:
            # permit_area must be a hashable scalar; malformed model output (e.g. a list)
            # is treated as "no matching permit"
            if isinstance(r.permit_area, str) and r.permit_area in permit_areas:
                verdicts.append(Answer(Verdict.OK, reason=f"permit {r.permit_area} exempts"))
            else:
                verdicts.append(Answer(Verdict.LIMITED, limit_minutes=r.limit_minutes,
                                       reason=f"{r.limit_minutes}min limit without permit {r.permit_area}"))
    if not verdicts:
        return Answer(Verdict.OK, reason="no restriction in effect")
    verdicts.sort(key=lambda a: SEVERITY.index(a.verdict))
    best = verdicts[0]
    if best.verdict is Verdict.LIMITED:
        # multiple limits: strictest applies. Tolerate None (malformed model reads).
        limits = [a.limit_minutes for a in verdicts
                  if a.verdict is Verdict.LIMITED and a.limit_minutes is not None]
        best.limit_minutes = min(limits) if limits else None
    return best