Coverage for src/data_api/domain/buses/services.py: 92%
276 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-11-13 19:28 +0000
1"""
2Buses domain service.
3Handles data fetching, caching, and processing using the Graph module.
4"""
6from __future__ import annotations
8from datetime import datetime, timedelta
9from itertools import product
10from typing import Any, Literal, Optional, cast
12from pydantic import TypeAdapter
14from data_api.data.manager import nthudata
15from data_api.data.nthudata import FetchFailure, JsonData, NTHUDataManager, Snapshot
16from data_api.domain.buses import enums, graph, models
18# Type lists
19BUS_ROUTE_TYPE = [t.value for t in enums.BusRouteType]
20BUS_ROUTE_TYPE_WITHOUT_ALL = [t for t in BUS_ROUTE_TYPE if t != "all"]
21BUS_DAY = [d.value for d in enums.BusDay]
22BUS_DIRECTION = [d.value for d in enums.BusDirection]
23BUS_DIRECTION_WITHOUT_ALL = [d for d in BUS_DIRECTION if d != "all"]
25ScheduleKey = tuple[str, str, str]
26ScheduleStore = dict[ScheduleKey, list[dict[str, Any]]]
27RouteInfoKey = tuple[str, str]
30# --- Helper Functions ---
31def after_specific_time(target_list: list[dict], time_str: str, time_keys: list[str]) -> list[dict]:
32 """
33 Filter list to keep only items after specified time.
35 Args:
36 target_list: List of dictionaries to filter
37 time_str: Reference time in "HH:MM" format
38 time_keys: Path to the time field (e.g., ["time"] or ["dep_info", "time"])
40 Returns:
41 Filtered list containing only items with time >= time_str
42 """
43 if not time_str or not target_list:
44 return target_list or []
46 try:
47 ref_time = datetime.strptime(time_str, "%H:%M")
48 filtered = []
49 for item in target_list:
50 item_time_str = _extract_nested_value(item, time_keys)
51 if not isinstance(item_time_str, str):
52 continue
54 try:
55 item_time = datetime.strptime(item_time_str, "%H:%M")
56 if item_time >= ref_time:
57 filtered.append(item)
58 except ValueError:
59 continue
60 return filtered
61 except ValueError:
62 return []
65def _extract_nested_value(data: dict, keys: list[str]) -> Any:
66 """Extract value from nested dictionary using key path."""
67 val = data
68 for k in keys:
69 if isinstance(val, dict):
70 val = val.get(k)
71 else:
72 return None
73 return val
76def add_time(time_str: str, minutes: int) -> str:
77 """
78 Add minutes to HH:MM string safely.
80 Args:
81 time_str: Time in "HH:MM" format
82 minutes: Minutes to add (can be negative)
84 Returns:
85 New time string in "HH:MM" format, or original on error
86 """
87 try:
88 dt = datetime.strptime(time_str, "%H:%M") + timedelta(minutes=minutes)
89 return dt.strftime("%H:%M")
90 except ValueError, TypeError:
91 return time_str
94def sort_by_time(target: list[dict], time_keys: list[str]) -> None:
95 """
96 Sort list of dictionaries by time field in-place.
98 Args:
99 target: List to sort
100 time_keys: Path to the time field (e.g., ["time"] or ["dep_info", "time"])
101 """
103 def get_sort_key(x):
104 time_str = _extract_nested_value(x, time_keys)
105 if not isinstance(time_str, str):
106 return datetime.max
107 try:
108 return datetime.strptime(time_str, "%H:%M")
109 except ValueError, TypeError:
110 return datetime.max # Push invalid to end
112 target.sort(key=get_sort_key)
115class PreparedBusData:
116 """Prepared bus indexes and pure queries, without client or refresh ownership."""
118 def __init__(self) -> None:
119 # Schedule data stores: (route_type, day, direction) -> list of schedules
120 self.raw_schedule_data = self._new_schedule_store()
121 self.detailed_schedule_data = self._new_schedule_store()
122 self._route_info: dict[RouteInfoKey, dict[str, Any]] = {}
124 self._res_json: dict[str, Any] = {}
125 # Track Gen2 (綜二館) departures for route calculation
126 self._gen2_departures: set[str] = set()
128 def _new_schedule_store(self) -> ScheduleStore:
129 """Create empty schedule store with all possible keys."""
130 return {
131 (rtype, day, direction): []
132 for rtype, day, direction in product(BUS_ROUTE_TYPE, BUS_DAY, BUS_DIRECTION)
133 }
135 @staticmethod
136 def prepare(raw: JsonData) -> PreparedBusData:
137 from data_api.models.buses import BusInfo, BusSchedule
139 if not isinstance(raw, dict):
140 raise FetchFailure("payload_type")
141 candidate = PreparedBusData()
142 info_keys = {
143 "towardTSMCBuildingInfo",
144 "towardMainGateInfo",
145 "towardNandaInfo",
146 "towardMainCampusInfo",
147 }
148 schedule_keys = {
149 candidate._get_schedule_json_key(rtype, day, direction)
150 for rtype, day, direction in product(
151 BUS_ROUTE_TYPE_WITHOUT_ALL, BUS_DAY, BUS_DIRECTION_WITHOUT_ALL
152 )
153 }
154 if raw and not (info_keys | schedule_keys).intersection(raw):
155 raise FetchFailure("validation")
156 for key in info_keys.intersection(raw):
157 BusInfo.model_validate(raw[key])
158 for key in schedule_keys.intersection(raw):
159 candidate._validate_schedule_rows(raw[key])
160 candidate._res_json = raw
161 candidate._process_all_data()
162 for schedules in candidate.raw_schedule_data.values():
163 for schedule in schedules:
164 BusSchedule.model_validate(schedule)
165 return candidate
167 @staticmethod
168 def _validate_schedule_rows(raw: object) -> None:
169 rows = TypeAdapter(list[dict[str, str]]).validate_python(raw, strict=True)
170 for row in rows:
171 if "time" not in row or "description" not in row:
172 raise FetchFailure("validation")
173 try:
174 datetime.strptime(row["time"], "%H:%M")
175 except ValueError as exc:
176 raise FetchFailure("validation") from exc
178 def _process_all_data(self) -> None:
179 """
180 Main processing pipeline for bus data.
182 Steps:
183 1. Reset all data structures
184 2. Populate route info (metadata)
185 3. Process raw schedules from JSON
186 4. Generate detailed schedules with arrival times
187 5. Sort all schedules by time
188 6. Derive combined views (all routes, all directions)
189 """
190 self._reset_registries()
191 self._populate_info_data()
192 self._populate_raw_schedule()
193 self._generate_detailed_schedule_and_stops()
194 self._sort_schedule_store(self.raw_schedule_data, ["time"])
195 self._sort_schedule_store(self.detailed_schedule_data, ["dep_info", "time"])
196 self._derive_combined_views()
198 def _reset_registries(self) -> None:
199 self._gen2_departures.clear()
200 self.raw_schedule_data = self._new_schedule_store()
201 self.detailed_schedule_data = self._new_schedule_store()
202 self._route_info = {}
204 # --- 1. Info Data ---
205 def _populate_info_data(self) -> None:
206 mapping = {
207 ("main", "up"): "towardTSMCBuildingInfo",
208 ("main", "down"): "towardMainGateInfo",
209 ("nanda", "up"): "towardNandaInfo",
210 ("nanda", "down"): "towardMainCampusInfo",
211 }
212 for (rtype, rdir), json_key in mapping.items():
213 info = self._res_json.get(json_key)
214 if isinstance(info, dict) and info:
215 self._route_info[(rtype, rdir)] = info
217 # --- 2. Raw Schedule ---
218 def _populate_raw_schedule(self) -> None:
219 """Process raw bus schedules from JSON data."""
220 for rtype, day, rdir in product(
221 BUS_ROUTE_TYPE_WITHOUT_ALL, BUS_DAY, BUS_DIRECTION_WITHOUT_ALL
222 ):
223 json_key = self._get_schedule_json_key(rtype, day, rdir)
224 raw_list = self._res_json.get(json_key, [])
226 enhanced_list = []
227 for item in raw_list:
228 bus = self._enhance_schedule_item(item, rtype, day, rdir)
229 enhanced_list.append(bus)
231 self.raw_schedule_data[(rtype, day, rdir)] = enhanced_list
233 def _get_schedule_json_key(self, rtype: str, day: str, rdir: str) -> str:
234 """Generate JSON key for bus schedule lookup."""
235 toward_name = ""
236 if rtype == "main":
237 toward_name = "TSMCBuilding" if rdir == "up" else "MainGate"
238 elif rtype == "nanda":
239 toward_name = "Nanda" if rdir == "up" else "MainCampus"
240 return f"{day}BusScheduleToward{toward_name}"
242 def _enhance_schedule_item(self, item: dict, rtype: str, day: str, rdir: str) -> dict:
243 """Enhance a single schedule item with additional fields."""
244 bus = item.copy()
245 bus["bus_type"] = self._classify_bus_type(rtype, day, bus.get("description", ""))
247 if rtype == "nanda":
248 self._enhance_nanda_schedule(bus, rdir)
249 elif rtype == "main":
250 self._track_gen2_departures_if_needed(bus)
252 return bus
254 def _enhance_nanda_schedule(self, bus: dict, rdir: str) -> None:
255 """Add Nanda-specific fields to schedule item."""
256 bus["dep_stop"] = "校門" if rdir == "up" else "南大"
257 bus["line"] = graph.resolver.get_nanda_line(bus.get("description", ""), bus.get("line", ""))
259 def _track_gen2_departures_if_needed(self, bus: dict) -> None:
260 """Track Gen2 (綜二) departures for main campus route calculation."""
261 if "綜二" in bus.get("dep_stop", ""):
262 line = bus.get("line", "")
263 time = bus.get("time", "")
264 predicted_gate_time = add_time(time, 7)
265 self._gen2_departures.add(f"{predicted_gate_time}{line}")
266 self._gen2_departures.add(f"0{predicted_gate_time}{line}")
268 def _classify_bus_type(self, rtype: str, day: str, desc: str) -> str:
269 """
270 Classify the bus vehicle type based on route, day, and description.
272 Rules:
273 - Nanda buses with "83" in description -> route_83 (city bus)
274 - Main campus buses with "大" (large) in description -> large_sized_bus
275 - Nanda weekday buses -> large_sized_bus
276 - All others -> middle_sized_bus
278 Args:
279 rtype: Route type ('main' or 'nanda')
280 day: Day type ('weekday' or 'weekend')
281 desc: Bus description text
283 Returns:
284 Bus type enum value
285 """
286 if rtype == "nanda" and "83" in desc:
287 return enums.BusType.route_83.value
288 if (rtype == "main" and "大" in desc) or (rtype == "nanda" and day == "weekday"):
289 return enums.BusType.large_sized_bus.value
290 return enums.BusType.middle_sized_bus.value
292 # --- 3. Detailed Schedule & Route Calculation ---
293 def _generate_detailed_schedule_and_stops(self) -> None:
294 """Generate detailed schedules with arrival times for each stop."""
295 for rtype, day, rdir in product(
296 BUS_ROUTE_TYPE_WITHOUT_ALL, BUS_DAY, BUS_DIRECTION_WITHOUT_ALL
297 ):
298 raw_list = self.raw_schedule_data.get((rtype, day, rdir), [])
299 if not isinstance(raw_list, list):
300 continue
302 detailed_list = []
303 for bus in raw_list:
304 detailed_entry = self._create_detailed_schedule_entry(bus, rtype, rdir)
305 detailed_list.append(detailed_entry)
307 self.detailed_schedule_data[(rtype, day, rdir)] = detailed_list
309 def _create_detailed_schedule_entry(self, bus: dict, rtype: str, rdir: str) -> dict:
310 """Create a detailed schedule entry with stop arrival times."""
311 route = self._get_route_from_graph(bus, rtype, rdir)
312 stops_time_info = []
314 if route:
315 start_time = bus.get("time", "00:00")
316 stops_time_info = self._calculate_stop_arrival_times(route, start_time)
318 return {
319 "dep_info": bus,
320 "stops_time": stops_time_info,
321 "bus_type": bus.get("bus_type", ""),
322 }
324 def _calculate_stop_arrival_times(
325 self,
326 route: models.Route,
327 start_time: str,
328 ) -> list[dict]:
329 """Calculate arrival times for each stop on the route."""
330 stops_time_info = []
331 for stop, offset in zip(route.stops, route.time_offsets):
332 arr_time = add_time(start_time, offset)
333 stops_time_info.append({"stop": stop.name, "arrive_time": arr_time})
335 return stops_time_info
337 def _get_route_from_graph(self, bus: dict, rtype: str, rdir: str) -> Optional[models.Route]:
338 """
339 Determine the route for a bus schedule entry.
341 Args:
342 bus: Bus schedule dictionary with time, line, dep_stop, etc.
343 rtype: Route type ('main' or 'nanda')
344 rdir: Direction ('up' or 'down')
346 Returns:
347 Route object if found, None otherwise
348 """
349 if rtype == "main":
350 return self._resolve_main_campus_route(bus)
351 elif rtype == "nanda":
352 return self._resolve_nanda_route(bus, rdir)
353 return None
355 def _resolve_main_campus_route(self, bus: dict) -> Optional[models.Route]:
356 """Resolve main campus route based on line, departure stop, and time."""
357 line = bus.get("line", "")
358 dep_stop = bus.get("dep_stop", "")
359 time_val = bus.get("time", "")
361 # Check if this bus departs from Gen2 (綜二) by inference
362 identifier = f"{time_val}{line}"
363 is_from_gen2 = (
364 identifier in self._gen2_departures or f"0{identifier}" in self._gen2_departures
365 )
367 return graph.resolver.resolve_main_campus_route(line, dep_stop, is_from_gen2)
369 def _resolve_nanda_route(self, bus: dict, rdir: str) -> Optional[models.Route]:
370 """Resolve Nanda route using the same line as the departure schedule."""
371 direction = cast(Literal["up", "down"], rdir)
372 return graph.resolver.resolve_nanda_route(
373 direction, bus.get("description", ""), bus.get("line", "")
374 )
376 def _sort_schedule_store(self, store: ScheduleStore, time_path: list[str]) -> None:
377 for entries in store.values():
378 if isinstance(entries, list) and entries:
379 sort_by_time(entries, time_path)
381 # --- 4. Post-Processing ("All" fields) ---
382 def _derive_combined_views(self) -> None:
383 self._combine_route_types_for_store(self.raw_schedule_data, ["time"])
384 self._combine_route_types_for_store(self.detailed_schedule_data, ["dep_info", "time"])
386 self._combine_directions_for_store(self.raw_schedule_data, ["time"])
387 self._combine_directions_for_store(self.detailed_schedule_data, ["dep_info", "time"])
389 def _combine_route_types_for_store(self, store: ScheduleStore, time_path: list[str]) -> None:
390 for day, rdir in product(BUS_DAY, BUS_DIRECTION_WITHOUT_ALL):
391 combined = [
392 *store.get(("main", day, rdir), []),
393 *store.get(("nanda", day, rdir), []),
394 ]
395 if combined:
396 sort_by_time(combined, time_path)
397 store[("all", day, rdir)] = combined
399 def _combine_directions_for_store(self, store: ScheduleStore, time_path: list[str]) -> None:
400 for rtype, day in product(BUS_ROUTE_TYPE, BUS_DAY):
401 combined = [
402 *store.get((rtype, day, "up"), []),
403 *store.get((rtype, day, "down"), []),
404 ]
405 if combined:
406 sort_by_time(combined, time_path)
407 store[(rtype, day, "all")] = combined
409 def get_route_info(self, route_type: Optional[str], direction: Optional[str]) -> list[dict]:
410 # 用篩選條件取得路線資訊
411 results = []
412 for (rtype, rdir), info in self._route_info.items():
413 if (route_type is None or route_type == rtype) and (
414 direction is None or direction == rdir
415 ):
416 results.append(info)
417 return results
419 @staticmethod
420 def _find_stop_schedule(schedules: list[dict], stop_name: str, detailed: bool) -> list[dict]:
421 matching = []
422 for entry in schedules:
423 for arrival in entry["stops_time"]:
424 if arrival["stop"] == stop_name:
425 item = entry if detailed else entry["dep_info"]
426 matching.append({**item, "arrive_time": arrival["arrive_time"]})
427 break
428 return matching
430 def get_schedule(
431 self,
432 *,
433 route_type: str,
434 day: str,
435 direction: str,
436 detailed: bool = False,
437 stop: enums.BusStopsName | None = None,
438 ) -> list[dict]:
439 if stop is not None:
440 schedules = self.detailed_schedule_data.get((route_type, day, direction), [])
441 return self._find_stop_schedule(schedules, stop.value, detailed)
442 store = self.detailed_schedule_data if detailed else self.raw_schedule_data
443 return store.get((route_type, day, direction), [])
445 def query_schedule(
446 self,
447 *,
448 route_type: str,
449 day: str,
450 direction: str,
451 detailed: bool = False,
452 stop: enums.BusStopsName | None = None,
453 after_time: str = "",
454 limit: int | None = 5,
455 ) -> list[dict]:
456 """Filter by the selected stop's arrival time, or departure time without a stop."""
457 schedules = self.get_schedule(
458 route_type=route_type, day=day, direction=direction, detailed=detailed, stop=stop
459 )
460 if stop is not None:
461 return after_specific_time(schedules, after_time, ["arrive_time"])[:limit]
462 time_path = ["dep_info", "time"] if detailed else ["time"]
463 return after_specific_time(schedules, after_time, time_path)[:limit]
465 @staticmethod
466 def gen_bus_stops_info() -> list[dict]:
467 return [
468 {
469 "name": s.name,
470 "name_en": s.name_en,
471 "latitude": s.latitude,
472 "longitude": s.longitude,
473 }
474 for s in graph.STOPS_DATA.values()
475 ]
478class BusesService:
479 """Refresh orchestration with one atomic, versioned prepared-data view."""
481 prepare = staticmethod(PreparedBusData.prepare)
482 gen_bus_stops_info = staticmethod(PreparedBusData.gen_bus_stops_info)
484 def __init__(self, manager: NTHUDataManager = nthudata) -> None:
485 self._snapshot: Snapshot[PreparedBusData] | None = None
486 self.manager = manager
487 self.state = manager.register("buses.json", self.prepare)
488 self.last_commit_hash: str | None = None
490 async def update_data(self) -> None:
491 self._snapshot = await self.manager.get_snapshot("buses.json", self.state)
492 self.last_commit_hash = self._snapshot.version
494 def reset(self) -> None:
495 self._snapshot = None
496 self.last_commit_hash = None
498 @property
499 def data(self) -> PreparedBusData:
500 if self._snapshot is None:
501 raise RuntimeError("Bus service has not loaded a snapshot")
502 return self._snapshot.data
504 @property
505 def raw_schedule_data(self) -> ScheduleStore:
506 return self.data.raw_schedule_data
508 def get_route_info(self, route_type: str | None, direction: str | None) -> list[dict]:
509 return self.data.get_route_info(route_type, direction)
511 def get_schedule(
512 self,
513 *,
514 route_type: str,
515 day: str,
516 direction: str,
517 detailed: bool = False,
518 stop: enums.BusStopsName | None = None,
519 ) -> list[dict]:
520 return self.data.get_schedule(
521 route_type=route_type, day=day, direction=direction, detailed=detailed, stop=stop
522 )
524 def query_schedule(
525 self,
526 *,
527 route_type: str,
528 day: str,
529 direction: str,
530 detailed: bool = False,
531 stop: enums.BusStopsName | None = None,
532 after_time: str = "",
533 limit: int | None = 5,
534 ) -> list[dict]:
535 return self.data.query_schedule(
536 route_type=route_type,
537 day=day,
538 direction=direction,
539 detailed=detailed,
540 stop=stop,
541 after_time=after_time,
542 limit=limit,
543 )
546buses_service = BusesService(nthudata)