1"""Zone-to-zone predicted travel-time matrices by hour, and how the planner reads them.2 3A matrix M has shape (24, Z, Z): M[h, i, j] = predicted taxi minutes from zone i to zone j4for a departure during hour h (one weekday). The van model multiplies by VAN_FACTOR (1.2).5 6Departure-time interpolation: the hourly value is placed at the middle of the hour7(h*60+30) and interpolated linearly in between. Linear interpolation keeps travel time8continuous and makes the network FIFO ("leaving later never makes you arrive earlier")9as long as |M[h+1]-M[h]| < 60 min, which `is_fifo` checks. FIFO + waiting allowed is what10makes "earliest clock time dominates" valid in the search (see search.py).11"""12from __future__ import annotations13 14import numpy as np15 16from . import config as C17 18 19def build_matrix(predictor, n_zones: int, weekday: int) -> np.ndarray:20 Z = n_zones21 h, i, j = np.meshgrid(np.arange(24), np.arange(Z), np.arange(Z), indexing="ij")22 pred = predictor.predict(i.ravel(), j.ravel(), h.ravel(), np.full(h.size, weekday))23 return np.asarray(pred, float).reshape(24, Z, Z)24 25 26def build_all_weekdays(predictor, n_zones: int) -> np.ndarray:27 return np.stack([build_matrix(predictor, n_zones, w) for w in range(5)])28 29 30class TimeModel:31 """Travel-time oracle used by the CSP, the search and the simulator."""32 33 def __init__(self, matrix: np.ndarray, factor: float = C.VAN_FACTOR, name: str = "model"):34 M = np.asarray(matrix, float)35 if M.ndim == 2: # constant matrix -> same for every hour36 M = np.repeat(M[None], 24, axis=0)37 self.M = M * factor38 self.factor, self.name = factor, name39 self.Z = M.shape[1]40 # suffix minimum over hour knots: lower bound on travel time for any departure >= t41 self.suffix_min = np.minimum.accumulate(self.M[::-1], axis=0)[::-1]42 self.global_min = self.M.min(axis=0)43 44 @staticmethod45 def _knot(t: float):46 x = (t - 30.0) / 60.047 if x <= 0:48 return 0, 0.049 if x >= 23:50 return 23, 0.051 h0 = int(np.floor(x))52 return h0, x - h053 54 def tt(self, i: int, j: int, t: float) -> float:55 h0, w = self._knot(t)56 if w == 0.0:57 return float(self.M[h0, i, j])58 return float((1 - w) * self.M[h0, i, j] + w * self.M[h0 + 1, i, j])59 60 def lower_bound_matrix(self, t: float) -> np.ndarray:61 """min over all departure times >= t of tt(., ., .) -- admissible edge costs."""62 h0, _ = self._knot(t)63 return self.suffix_min[h0]64 65 def is_fifo(self) -> bool:66 return bool(np.all(np.diff(self.M, axis=0) > -60.0))