semantics.py

"""Semantic rules across rows (TODS-x4xx).

These checks are deliberately conservative where the spec is permissive. Real
schedules have legitimate edge cases (zero-minute events, overnight times,
overlapping non-trip events), and the spec allows them; only constraints the
spec actually states are errors here.
"""

from __future__ import annotations

from collections.abc import Iterator
from dataclasses import dataclass

from ..findings import Finding, Severity
from ..gtfs_companion import parse_gtfs_date
from ..loader import Row
from ..schema import SPEC_URL
from . import ValidationContext, rule
from .fields import parse_time

_RUN_EVENTS_SECTION = f"{SPEC_URL}#run_eventstxt"


@dataclass(frozen=True)
class _Event:
    row: Row
    service_id: str
    run_id: str
    sequence: int | None
    trip_id: str
    start: int | None  # seconds
    end: int | None

    @property
    def run(self) -> tuple[str, str]:
        return (self.service_id, self.run_id)


def _events(context: ValidationContext) -> list[_Event]:
    feed = context.package.get("run_events.txt")
    if feed is None:
        return []
    events = []
    for row in feed.rows:
        sequence_raw = row.values.get("event_sequence", "")
        events.append(
            _Event(
                row=row,
                service_id=row.values.get("service_id", ""),
                run_id=row.values.get("run_id", ""),
                sequence=int(sequence_raw) if sequence_raw.isdigit() else None,
                trip_id=row.values.get("trip_id", ""),
                start=parse_time(row.values.get("start_time", "")),
                end=parse_time(row.values.get("end_time", "")),
            )
        )
    return events


def _events_by_run(context: ValidationContext) -> dict[tuple[str, str], list[_Event]]:
    runs: dict[tuple[str, str], list[_Event]] = {}
    for event in _events(context):
        if event.service_id and event.run_id:
            runs.setdefault(event.run, []).append(event)
    return runs


@rule(
    id="TODS-E401",
    severity=Severity.ERROR,
    title="Event ends before it starts",
    description=(
        "A run event's end_time is earlier than its start_time. Equal times are fine "
        "(the spec allows zero-duration events such as a report time); for work past "
        "midnight, use hours of 24 or more rather than wrapping around."
    ),
    spec_section=_RUN_EVENTS_SECTION,
    interpretation=(
        "the spec is silent on end<start; this treats it as an error and equal "
        "times as valid (spec-questions #5)."
    ),
)
def event_ends_before_start(context: ValidationContext) -> Iterator[Finding]:
    for event in _events(context):
        if event.start is not None and event.end is not None and event.end < event.start:
            yield Finding(
                rule_id="TODS-E401",
                severity=Severity.ERROR,
                file="run_events.txt",
                row=event.row.line,
                field="end_time",
                message=(
                    f"run_events.txt row {event.row.line}: end_time "
                    f"{event.row.values.get('end_time', '')!r} is earlier than "
                    f"start_time {event.row.values.get('start_time', '')!r}."
                ),
                suggestion=(
                    "If the event runs past midnight, keep counting hours upward: "
                    "write 1:10 AM the next day as '25:10:00'."
                ),
            )


@rule(
    id="TODS-E402",
    severity=Severity.ERROR,
    title="Two trip events in one run overlap in time",
    description=(
        "Within one run, two events that both reference trips overlap in time. The "
        "spec prohibits this: an employee cannot be on two trips at once. Touching "
        "end-to-start (zero-minute overlap) is allowed."
    ),
    spec_section=f"{SPEC_URL}#event_sequence-and-event-times",
)
def overlapping_trip_events(context: ValidationContext) -> Iterator[Finding]:
    for events in _events_by_run(context).values():
        trip_events = [e for e in events if e.trip_id and e.start is not None and e.end is not None]
        trip_events.sort(key=lambda e: (e.start or 0, e.end or 0))
        # Sweep with the latest-ending event seen so far, so an event that
        # spans several later ones is compared against each of them.
        previous: _Event | None = None
        for current in trip_events:
            if (
                previous is not None
                and current.start is not None
                and previous.end is not None
                and current.start < previous.end
            ):
                yield Finding(
                    rule_id="TODS-E402",
                    severity=Severity.ERROR,
                    file="run_events.txt",
                    row=current.row.line,
                    message=(
                        f"run_events.txt row {current.row.line}: trip "
                        f"{current.trip_id!r} starts at "
                        f"{current.row.values.get('start_time', '')} but the same "
                        f"run is still on trip {previous.trip_id!r} until "
                        f"{previous.row.values.get('end_time', '')} (row "
                        f"{previous.row.line}). Trip events in one run must not "
                        "overlap; an employee cannot work two trips at once."
                    ),
                )
            if previous is None or (current.end or 0) > (previous.end or 0):
                previous = current


@rule(
    id="TODS-W403",
    severity=Severity.WARNING,
    title="Event order disagrees with event times",
    description=(
        "Within one run, event_sequence does not increase with start_time. The spec "
        "says sequence values should increase throughout the day; out-of-order values "
        "usually mean the sequence or a time is wrong."
    ),
    spec_section=f"{SPEC_URL}#event_sequence-and-event-times",
)
def sequence_disagrees_with_time(context: ValidationContext) -> Iterator[Finding]:
    for events in _events_by_run(context).values():
        timed = [e for e in events if e.sequence is not None and e.start is not None]
        timed.sort(key=lambda e: e.sequence or 0)
        for previous, current in zip(timed, timed[1:], strict=False):
            if previous.start is None or current.start is None:
                continue  # filtered above; keeps the type-checker satisfied
            if current.start < previous.start:
                yield Finding(
                    rule_id="TODS-W403",
                    severity=Severity.WARNING,
                    file="run_events.txt",
                    row=current.row.line,
                    field="event_sequence",
                    message=(
                        f"run_events.txt row {current.row.line}: event_sequence "
                        f"{current.sequence} starts at "
                        f"{current.row.values.get('start_time', '')}, earlier than "
                        f"event_sequence {previous.sequence} (row {previous.row.line}, "
                        f"{previous.row.values.get('start_time', '')}) in the same "
                        "run. Sequence values should increase through the day."
                    ),
                )
                break  # one finding per run is enough to point at the problem


@rule(
    id="TODS-W404",
    severity=Severity.WARNING,
    title="Employee is assigned to overlapping runs on the same date",
    description=(
        "An employee is assigned to two runs on the same date whose events overlap in "
        "time. Real schedules have legitimate exceptions, so this is a warning, not "
        "an error."
    ),
    spec_section=f"{SPEC_URL}#employee_run_datestxt",
)
def employee_double_booked(context: ValidationContext) -> Iterator[Finding]:
    assignments = context.package.get("employee_run_dates.txt")
    if assignments is None:
        return
    spans: dict[tuple[str, str], tuple[int, int]] = {}
    for run, events in _events_by_run(context).items():
        times = [(e.start, e.end) for e in events if e.start is not None and e.end is not None]
        if times:
            spans[run] = (min(t[0] for t in times), max(t[1] for t in times))

    by_employee_date: dict[tuple[str, str], list[tuple[tuple[str, str], Row]]] = {}
    for row in assignments.rows:
        employee_id = row.values.get("employee_id", "")
        day = row.values.get("date", "")
        run = (row.values.get("service_id", ""), row.values.get("run_id", ""))
        if employee_id and day and all(run):
            by_employee_date.setdefault((employee_id, day), []).append((run, row))

    for (employee_id, day), entries in by_employee_date.items():
        for i, (run_a, _row_a) in enumerate(entries):
            for run_b, row_b in entries[i + 1 :]:
                if run_a == run_b or run_a not in spans or run_b not in spans:
                    continue
                start_a, end_a = spans[run_a]
                start_b, end_b = spans[run_b]
                if start_b < end_a and start_a < end_b:
                    yield Finding(
                        rule_id="TODS-W404",
                        severity=Severity.WARNING,
                        file="employee_run_dates.txt",
                        row=row_b.line,
                        message=(
                            f"employee_run_dates.txt row {row_b.line}: employee "
                            f"{employee_id!r} is assigned to run "
                            f"(service_id {run_b[0]!r}, run_id {run_b[1]!r}) on "
                            f"{day}, which overlaps in time with their assignment to "
                            f"run (service_id {run_a[0]!r}, run_id {run_a[1]!r}) on "
                            "the same date."
                        ),
                        suggestion=(
                            "If this is intentional (split duties), no change is "
                            "needed; otherwise check for a stale assignment."
                        ),
                    )


@rule(
    id="TODS-E405",
    severity=Severity.ERROR,
    title="Run operates on dates its trip does not",
    description=(
        "A run event works a trip whose service is different from the run's service, "
        "and the run's service operates on dates the trip's service does not. The "
        "spec requires the run's dates to be a subset of the trip's dates."
    ),
    spec_section=f"{SPEC_URL}#service_id-crew-schedules-and-trip-schedules",
    needs_gtfs=True,
)
def run_dates_exceed_trip_dates(context: ValidationContext) -> Iterator[Finding]:
    assert context.gtfs is not None
    dates = context.gtfs.service_dates
    reported: set[tuple[str, str]] = set()
    for event in _events(context):
        if not event.trip_id or not event.service_id:
            continue
        trip_service = context.gtfs.trip_service.get(event.trip_id)
        if not trip_service or trip_service == event.service_id:
            continue
        if event.service_id not in dates or trip_service not in dates:
            continue  # undefined services are TODS-E308's concern
        if (event.service_id, trip_service) in reported:
            continue
        extra = dates[event.service_id] - dates[trip_service]
        if extra:
            reported.add((event.service_id, trip_service))
            sample = min(extra).strftime("%Y%m%d")
            yield Finding(
                rule_id="TODS-E405",
                severity=Severity.ERROR,
                file="run_events.txt",
                row=event.row.line,
                field="service_id",
                message=(
                    f"run_events.txt row {event.row.line}: the run's service "
                    f"{event.service_id!r} operates on {len(extra)} date(s) (e.g. "
                    f"{sample}) when trip {event.trip_id!r}'s service "
                    f"{trip_service!r} does not. A run may only work a trip on dates "
                    "the trip actually operates."
                ),
                suggestion=(
                    "Adjust the run's service days in calendar_supplement.txt or "
                    "calendar_dates_supplement.txt so they are a subset of the "
                    "trip's."
                ),
            )


@rule(
    id="TODS-W406",
    severity=Severity.WARNING,
    title="Employee assignment date is outside the run's service days",
    description=(
        "An employee_run_dates.txt row assigns a run on a date when the run's "
        "service_id does not operate, according to the supplemented calendars."
    ),
    spec_section=f"{SPEC_URL}#employee_run_datestxt",
    needs_gtfs=True,
)
def assignment_outside_service(context: ValidationContext) -> Iterator[Finding]:
    assert context.gtfs is not None
    assignments = context.package.get("employee_run_dates.txt")
    if assignments is None:
        return
    dates = context.gtfs.service_dates
    for row in assignments.rows:
        service_id = row.values.get("service_id", "")
        day = parse_gtfs_date(row.values.get("date", ""))
        if not service_id or day is None or service_id not in dates:
            continue
        if day not in dates[service_id]:
            yield Finding(
                rule_id="TODS-W406",
                severity=Severity.WARNING,
                file="employee_run_dates.txt",
                row=row.line,
                field="date",
                message=(
                    f"employee_run_dates.txt row {row.line}: "
                    f"{row.values.get('date', '')} is not a day that service "
                    f"{service_id!r} operates, so run "
                    f"(service_id {service_id!r}, run_id "
                    f"{row.values.get('run_id', '')!r}) has nothing scheduled then."
                ),
                suggestion=(
                    "Check the date, or extend the service via the calendar supplement files."
                ),
            )


@rule(
    id="TODS-W407",
    severity=Severity.WARNING,
    title="Vehicle assignment date is outside the service days",
    description=(
        "A vehicle_assignments.txt row names a service_id and a date when that "
        "service does not operate, according to the supplemented calendars."
    ),
    spec_section=f"{SPEC_URL}#vehicle_assignmentstxt",
    needs_gtfs=True,
)
def vehicle_assignment_outside_service(context: ValidationContext) -> Iterator[Finding]:
    assert context.gtfs is not None
    assignments = context.package.get("vehicle_assignments.txt")
    if assignments is None:
        return
    dates = context.gtfs.service_dates
    for row in assignments.rows:
        service_id = row.values.get("service_id", "")
        day = parse_gtfs_date(row.values.get("date", ""))
        if not service_id or day is None or service_id not in dates:
            continue
        if day not in dates[service_id]:
            yield Finding(
                rule_id="TODS-W407",
                severity=Severity.WARNING,
                file="vehicle_assignments.txt",
                row=row.line,
                field="date",
                message=(
                    f"vehicle_assignments.txt row {row.line}: "
                    f"{row.values.get('date', '')} is not a day that service "
                    f"{service_id!r} operates, so block "
                    f"{row.values.get('block_id', '')!r} does not run then."
                ),
            )


@rule(
    id="TODS-W408",
    severity=Severity.WARNING,
    title="Identical employee assignment appears twice",
    description=(
        "Two rows in employee_run_dates.txt are exactly identical (same date, "
        "service, run, and employee). Multiple employees per run are fine; the same "
        "employee twice is usually an export bug."
    ),
    spec_section=f"{SPEC_URL}#employee_run_datestxt",
    interpretation=(
        "permissive: the spec's 'Primary Key: *' is read as not forbidding an exact "
        "duplicate row, so this is a warning rather than an error (spec-questions #6)."
    ),
)
def duplicate_assignment(context: ValidationContext) -> Iterator[Finding]:
    assignments = context.package.get("employee_run_dates.txt")
    if assignments is None:
        return
    seen: dict[tuple[str, str, str, str], int] = {}
    for row in assignments.rows:
        key = (
            row.values.get("date", ""),
            row.values.get("service_id", ""),
            row.values.get("run_id", ""),
            row.values.get("employee_id", ""),
        )
        if not all(key):
            continue
        if key in seen:
            yield Finding(
                rule_id="TODS-W408",
                severity=Severity.WARNING,
                file="employee_run_dates.txt",
                row=row.line,
                message=(
                    f"employee_run_dates.txt row {row.line} repeats the assignment "
                    f"on row {seen[key]} exactly (date {key[0]}, service_id "
                    f"{key[1]!r}, run_id {key[2]!r}, employee_id {key[3]!r})."
                ),
                suggestion="Remove the duplicate row.",
            )
        else:
            seen[key] = row.line