Spaces:
Sleeping
Sleeping
| """Tasks and plans. | |
| A task is a window of one real 2025 day on the Groningen - Leeuwarden line after something went wrong. The agent is | |
| the dispatcher: every train (a *run* from one end of the line to the other) runs by default exactly as scheduled and | |
| as early as the physics allow (never ahead of its timetable). The plan says only what to change: | |
| hold keep a train at a station until a minute (to let an opposing train through a single-track section, | |
| or to wait for a section to reopen) | |
| to short-turn: end the run early at a turn point, so the same train can start back from there | |
| from start a run late at a turn point (where its train was short-turned) | |
| cancel do not run it at all (its train stays where it is) | |
| The physics, given a plan, are deterministic (`check.simulate`): a train leaves each point at the latest of when it | |
| is ready, its scheduled departure (never early at a scheduled stop) and its hold; it runs at its minimum running | |
| time. A train held where it has no scheduled stop loses `stop_loss` minutes braking and `stop_loss` accelerating. | |
| Whether the result is legal (single-track and headway conflicts, platform tracks, closed sections, where each train | |
| ends up) is `check.evaluate`. | |
| Times are whole minutes after 00:00 of the service day (00:30 the next night is 24:30, minute 1470); plans may give | |
| them as integers or "HH:MM". | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import re | |
| from dataclasses import asdict, dataclass, field | |
| from typing import Any | |
| RULES = { | |
| "h_follow": 3, # minutes between two trains going the same way, at both ends of a section | |
| "h_opp": 0, # minutes after a train leaves a single-track section before an opposing train may enter it | |
| # (0: in the same minute, as the timetable's crossings are planned) | |
| "turn_min": 4, # minutes for a train to turn back (driver changes ends) | |
| "stop_loss": 1, # minutes lost braking, and again accelerating, at a stop that is not in the timetable | |
| "cancel_cost": 30, # per scheduled stop a train does not serve (the next train of its kind is 30 minutes later) | |
| "max_delay": 120, # no train may reach a stop more than this many minutes late | |
| "gap_k": 100, # reward: added to the avoidable cost (about three unserved stops) | |
| } | |
| class Run: | |
| """One scheduled train from one end of its route to the other.""" | |
| id: str # train number | |
| kind: str # stop (Stoptrein: every station) | fast (Sneltrein) | |
| route: list[str] # every point from origin to destination in travel order, junctions included | |
| stops: list[bool] # scheduled stop at route[i] | |
| arr: list[int | None] # scheduled arrival at a stop, the timetable's passing minute elsewhere; None at origin | |
| dep: list[int | None] # scheduled departure at a stop, the passing minute elsewhere; None at destination | |
| seg: list[int] # minimum running minutes route[i] -> route[i + 1] for this train's stopping pattern | |
| unit: str # the train (rolling-stock unit) that runs it | |
| fixed_to: int = 0 # in progress: route index of the point it is at or approaching when the task starts | |
| fixed_arr: int | None = None # in progress: its arrival there (cannot change) | |
| fixed_from: int | None = None # in progress and between points: the minute it left route[fixed_to - 1] | |
| stuck: bool = False # in progress inside a section that closed: it waits there for the reopening | |
| def origin(self) -> str: | |
| return self.route[0] | |
| def dest(self) -> str: | |
| return self.route[-1] | |
| def started(self) -> bool: | |
| return self.fixed_arr is not None | |
| def n_stops(self) -> int: | |
| return sum(self.stops) | |
| def idx(self, code: str) -> int: | |
| return self.route.index(code) | |
| class Unit: | |
| """A train set and the runs it works through the window, in order.""" | |
| id: str | |
| runs: list[str] | |
| at: str | None # where it stands before its first run (None: its first run is already moving) | |
| ready: int | None # when it may leave `at` | |
| after: dict | None = None # its first run after the window: {"run", "station", "dep", "stops"} | |
| class Closure: | |
| kind: str # closed (no train may be in the sections) | slow (+extra minutes) | track (one | |
| # platform track out of use at a station) | |
| where: list[str] # closed/slow: [a, b], every section between points a and b; track: [station] | |
| start: int | |
| end: int # exclusive | |
| extra: int = 0 | |
| cause: str = "" | |
| real: bool = False # from the 2025 disruption record (else generated) | |
| class Task: | |
| task_id: str | |
| split: str | |
| line: str | |
| day: str | |
| start: int # "now": the minute the agent takes over | |
| end: int # runs leaving their origin before this minute are in the task | |
| network: dict | |
| runs: list[Run] | |
| units: list[Unit] | |
| closures: list[Closure] | |
| notices: list[str] | |
| disruptions: list[dict] | |
| difficulty: str | |
| reference: dict = field(default_factory=dict) # optimum, floor, baselines, the real outcome: never shown | |
| rules: dict = field(default_factory=lambda: dict(RULES)) | |
| def to_dict(self, public: bool = False) -> dict: | |
| d = asdict(self) | |
| if public: | |
| d.pop("reference") | |
| return d | |
| def from_dict(cls, d: dict) -> "Task": | |
| d = dict(d) | |
| d["runs"] = [Run(**r) for r in d["runs"]] | |
| d["units"] = [Unit(**u) for u in d["units"]] | |
| d["closures"] = [Closure(**c) for c in d["closures"]] | |
| d.setdefault("reference", {}) | |
| d["rules"] = {**RULES, **(d.get("rules") or {})} | |
| return cls(**d) | |
| def run(self, rid: str) -> Run: | |
| return self.run_map[rid] | |
| def run_map(self) -> dict[str, Run]: | |
| m = self.__dict__.get("_run_map") | |
| if m is None: | |
| m = self.__dict__["_run_map"] = {r.id: r for r in self.runs} | |
| return m | |
| def rule(self, k: str) -> int: | |
| return int(self.rules.get(k, RULES[k])) | |
| # ---------------------------------------------------------------- plans | |
| class Order: | |
| """What the plan changes for one run (absent from the plan: runs in full, as early as possible).""" | |
| cancel: bool = False | |
| start: str | None = None # "from": start at this turn point instead of the origin | |
| end: str | None = None # "to": end at this turn point instead of the destination | |
| hold: dict[str, int] = field(default_factory=dict) # point -> do not leave before this minute | |
| def to_json(self, rid: str) -> dict: | |
| d: dict[str, Any] = {"train": rid} | |
| if self.cancel: | |
| d["cancel"] = True | |
| return d | |
| if self.start: | |
| d["from"] = self.start | |
| if self.end: | |
| d["to"] = self.end | |
| if self.hold: | |
| d["hold"] = {k: hhmm(v) for k, v in self.hold.items()} | |
| return d | |
| Plan = dict[str, Order] | |
| class PlanError(ValueError): | |
| """The plan cannot be read at all.""" | |
| def hhmm(m: int | None) -> str | None: | |
| if m is None: | |
| return None | |
| return f"{m // 60:02d}:{m % 60:02d}" | |
| _TIME = re.compile(r"^\s*(\d{1,2}):(\d{2})\s*$") | |
| def parse_time(v: Any) -> int: | |
| if isinstance(v, bool): | |
| raise ValueError(f"not a time: {v!r}") | |
| if isinstance(v, int): | |
| m = v | |
| elif isinstance(v, float): | |
| if v != v or v in (float("inf"), float("-inf")) or v != int(v): | |
| raise ValueError(f"not a whole minute: {v!r}") | |
| m = int(v) | |
| elif isinstance(v, str): | |
| s = v.strip() | |
| if re.fullmatch(r"-?\d+", s): | |
| m = int(s) | |
| else: | |
| g = _TIME.match(s) | |
| if not g: | |
| raise ValueError(f'not a time: {v!r} (use "HH:MM" or minutes after midnight)') | |
| h, mi = int(g.group(1)), int(g.group(2)) | |
| if mi > 59: | |
| raise ValueError(f"not a time: {v!r}") | |
| m = h * 60 + mi | |
| else: | |
| raise ValueError(f"not a time: {v!r}") | |
| if not 0 <= m <= 48 * 60: | |
| raise ValueError(f"time out of range: {v!r}") | |
| return m | |
| def _load(raw: Any) -> Any: | |
| if isinstance(raw, str): | |
| s = raw.strip() | |
| fence = re.search(r"```(?:json)?\s*(.*?)```", s, re.S) | |
| if fence: | |
| s = fence.group(1).strip() | |
| if not s.startswith(("[", "{")): | |
| i = min([k for k in (s.find("["), s.find("{")) if k >= 0], default=-1) | |
| if i < 0: | |
| raise PlanError("the plan is not JSON: give a list like " | |
| '[{"train": "37413", "hold": {"HDG": "07:31"}}, {"train": "37324", "cancel": true}]') | |
| s = s[i:] | |
| try: | |
| raw, _ = json.JSONDecoder().raw_decode(s) | |
| except json.JSONDecodeError as e: | |
| raise PlanError(f"the plan is not valid JSON: {e}") from None | |
| if isinstance(raw, dict): | |
| if "plan" in raw and isinstance(raw["plan"], (list, dict)): | |
| return _load(raw["plan"]) | |
| # {"37413": {...}, ...} | |
| raw = [{"train": k, **(v if isinstance(v, dict) else {"cancel": v} if isinstance(v, bool) else {"bad": v})} | |
| for k, v in raw.items()] | |
| if not isinstance(raw, list): | |
| raise PlanError("the plan must be a list of entries, one per train you change") | |
| return raw | |
| def parse_plan(task: Task, raw: Any) -> tuple[Plan, list[str]]: | |
| """(plan, entry problems). Raises PlanError only when nothing can be read. An empty list is a valid plan: | |
| every train runs in full, as early as it can.""" | |
| entries = _load(raw) | |
| runs = task.run_map | |
| plan: Plan = {} | |
| problems: list[str] = [] | |
| for n, e in enumerate(entries): | |
| if not isinstance(e, dict): | |
| problems.append(f"entry {n}: not an object") | |
| continue | |
| rid = e.get("train", e.get("run", e.get("id"))) | |
| if isinstance(rid, bool) or rid is None: | |
| problems.append(f"entry {n}: no train number") | |
| continue | |
| rid = str(rid).strip() | |
| if rid not in runs: | |
| problems.append(f"entry {n}: unknown train {rid!r}") | |
| continue | |
| if rid in plan: | |
| problems.append(f"entry {n}: train {rid} appears twice") | |
| continue | |
| r = runs[rid] | |
| o = Order() | |
| unknown = set(e) - {"train", "run", "id", "cancel", "from", "start", "to", "end", "hold", "note", "reason"} | |
| if unknown: | |
| problems.append(f"train {rid}: unknown field(s) {sorted(unknown)}") | |
| c = e.get("cancel", False) | |
| if c not in (True, False, None): | |
| problems.append(f"train {rid}: cancel must be true or false") | |
| o.cancel = c is True | |
| for key, alt, attr in (("from", "start", "start"), ("to", "end", "end")): | |
| v = e.get(key, e.get(alt)) | |
| if v is None: | |
| continue | |
| if not isinstance(v, str) or v.strip().upper() not in r.route: | |
| problems.append(f"train {rid}: {key} {v!r} is not on its route {'-'.join(r.route)}") | |
| continue | |
| setattr(o, attr, v.strip().upper()) | |
| h = e.get("hold") or {} | |
| if not isinstance(h, dict): | |
| problems.append(f'train {rid}: hold must be an object like {{"HDG": "07:31"}}') | |
| h = {} | |
| for k, v in h.items(): | |
| code = str(k).strip().upper() | |
| if code not in r.route: | |
| problems.append(f"train {rid}: hold at {k!r}, which is not on its route") | |
| continue | |
| try: | |
| o.hold[code] = parse_time(v) | |
| except ValueError as ex: | |
| problems.append(f"train {rid}: hold at {code}: {ex}") | |
| plan[rid] = o | |
| return plan, problems | |
| def plan_to_list(plan: Plan) -> list[dict]: | |
| return [o.to_json(rid) for rid, o in plan.items()] | |
| def plan_from_list(task: Task, entries: list[dict]) -> Plan: | |
| plan, problems = parse_plan(task, entries) | |
| if problems: | |
| raise PlanError("; ".join(problems)) | |
| return plan | |