import json import math import os import ssl import urllib.parse import urllib.request from copy import deepcopy from datetime import datetime, timedelta MODE_NAMES = ("출근", "퇴근") PRESET_LIMIT = 10 TRANSIT_DB_FILE = "transit_db.json" DEFAULT_API_KEYS = { "odsay": "pJAzwA1O0a1Om/kUegbS/vqK8jD+ProL+Ph2RhIl7Vw", "subway": "7142474d556b696d383570414f6a52", "busArrive": "b416f73a5840954c79b8b4e34005cb18df555284583a3a2bff7c304b1449b980", "googleDirections": "AIzaSyC7f5hegysRFSXHxwurFwLl-BUf0RuYTPo", "kakao": "4213a733f9c3b7e3b8824674aebff659", } SUBWAY_LINE_KEYS = [ "1호선", "2호선", "3호선", "4호선", "5호선", "6호선", "7호선", "8호선", "9호선", "경의중앙", "수인분당", "경춘", "공항철도", "신분당", "우이신설", "서해", "신림", "김포골드", ] SUBWAY_ID_TO_KEY = { "1001": "1호선", "1002": "2호선", "1003": "3호선", "1004": "4호선", "1005": "5호선", "1006": "6호선", "1007": "7호선", "1008": "8호선", "1009": "9호선", "1063": "경의중앙", "1065": "공항철도", "1067": "경춘", "1075": "수인분당", "1077": "신분당", "1092": "우이신설", "1093": "서해", "1094": "신림", } def default_api_keys(): return deepcopy(DEFAULT_API_KEYS) def merge_api_keys(value): merged = default_api_keys() if isinstance(value, dict): for key in merged: candidate = _normalize_label(value.get(key)) if candidate: merged[key] = candidate return merged def _ssl_ctx(): ctx = ssl.create_default_context() ctx.check_hostname = False ctx.verify_mode = ssl.CERT_NONE return ctx def _request_json(url, timeout=10, headers=None, use_ssl=False): req = urllib.request.Request(url, headers=headers or {"User-Agent": "Mozilla/5.0"}) kwargs = {"timeout": timeout} if use_ssl: kwargs["context"] = _ssl_ctx() with urllib.request.urlopen(req, **kwargs) as resp: raw = resp.read().decode("utf-8", errors="replace") return json.loads(raw) def _now_iso(): return datetime.now().isoformat() def _parse_float(value, default=None): try: return float(value) except (TypeError, ValueError): return default def _normalize_label(value, default=""): return str(value or default).strip() def _normalize_transport_type(value, default="bus"): return "subway" if value == "subway" else default def _point(label="", lat=None, lng=None, point_type="point", meta=None): point = { "label": _normalize_label(label), "lat": _parse_float(lat), "lng": _parse_float(lng), "type": point_type, } if meta: point["meta"] = meta return point def _valid_point(point): return isinstance(point, dict) and _parse_float(point.get("lat")) is not None and _parse_float(point.get("lng")) is not None def default_runtime_state(): return { "journey_phase": "idle", "start_triggered_at": None, "boarding_confirmed_at": None, "last_known_position": None, "last_known_position_updated_at": None, "nearest_segment_ref": None, "live_candidate_trips": [], "live_boardable_trip": None, "live_vehicle_marker": None, "live_position_marker": None, "live_train_position": None, } def default_mode_bucket(): return { "config": { "transport_type": "bus", "display_start_name": "", "display_end_name": "", "bus": { "origin_mode": "manual", "from_address": None, "from_stop": None, "to_stop": None, "to_address": None, "shared_routes": [], }, "subway": { "origin_mode": "manual", "from_address": None, "from_station": None, "via_station": None, "to_station": None, "to_address": None, }, }, "last_plan": None, "runtime_state": default_runtime_state(), } def default_preset(index=1, label=None): return { "id": f"preset-{index}", "label": _normalize_label(label, f"프리셋 {index}"), "modes": { "출근": default_mode_bucket(), "퇴근": default_mode_bucket(), }, } def default_commute_state(): return { "version": 2, "activePresetId": "preset-1", "activeMode": "출근", "theme": "dark", "showOnDashboard": True, "apiKeys": default_api_keys(), "presets": [default_preset(1)], } def _normalize_mode_name(value): return "퇴근" if value == "퇴근" else "출근" def _clean_object(value): if isinstance(value, dict): return {k: _clean_object(v) for k, v in value.items()} if isinstance(value, list): return [_clean_object(v) for v in value] return value def normalize_commute_state(state): root = deepcopy(state or {}) default_root = default_commute_state() root["version"] = 2 root["theme"] = root.get("theme") or default_root["theme"] root["showOnDashboard"] = root.get("showOnDashboard", True) root["apiKeys"] = merge_api_keys(root.get("apiKeys")) presets = [] seen_ids = set() for idx, preset in enumerate(root.get("presets") or [], start=1): preset = deepcopy(preset or {}) preset_id = _normalize_label(preset.get("id"), f"preset-{idx}") if preset_id in seen_ids: preset_id = f"{preset_id}-{idx}" seen_ids.add(preset_id) normalized = default_preset(idx, preset.get("label")) normalized["id"] = preset_id normalized["label"] = _normalize_label(preset.get("label"), normalized["label"]) modes = preset.get("modes") or {} for mode_name in MODE_NAMES: raw_bucket = deepcopy(modes.get(mode_name) or {}) bucket = default_mode_bucket() if isinstance(raw_bucket.get("config"), dict): bucket["config"]["transport_type"] = _normalize_transport_type( raw_bucket["config"].get("transport_type"), bucket["config"]["transport_type"], ) bucket["config"]["display_start_name"] = _normalize_label( raw_bucket["config"].get("display_start_name"), "", ) bucket["config"]["display_end_name"] = _normalize_label( raw_bucket["config"].get("display_end_name"), "", ) bucket["config"]["bus"].update(raw_bucket["config"].get("bus") or {}) bucket["config"]["subway"].update(raw_bucket["config"].get("subway") or {}) if not raw_bucket["config"].get("transport_type"): if bucket["config"]["subway"].get("from_station") or bucket["config"]["subway"].get("to_station"): bucket["config"]["transport_type"] = "subway" if raw_bucket.get("last_plan"): bucket["last_plan"] = raw_bucket.get("last_plan") runtime = raw_bucket.get("runtime_state") or {} merged_runtime = default_runtime_state() merged_runtime.update(runtime) bucket["runtime_state"] = merged_runtime normalized["modes"][mode_name] = bucket presets.append(normalized) if len(presets) >= PRESET_LIMIT: break if not presets: presets = default_root["presets"] root["presets"] = presets active_preset = _normalize_label(root.get("activePresetId"), presets[0]["id"]) if active_preset not in {p["id"] for p in presets}: active_preset = presets[0]["id"] root["activePresetId"] = active_preset root["activeMode"] = _normalize_mode_name(root.get("activeMode")) return _clean_object(root) def migrate_legacy_payload(payload): if not isinstance(payload, dict): return default_commute_state() if isinstance(payload.get("commuteConfig"), dict) and payload["commuteConfig"].get("presets"): return normalize_commute_state(payload["commuteConfig"]) if payload.get("presets"): return normalize_commute_state(payload) legacy = payload.get("commuteConfig") if isinstance(payload.get("commuteConfig"), dict) else payload if legacy.get("presets"): return normalize_commute_state(legacy) root = default_commute_state() root["theme"] = legacy.get("theme") or root["theme"] root["showOnDashboard"] = legacy.get("showOnDashboard", True) root["activeMode"] = _normalize_mode_name(legacy.get("commuteMode")) root["apiKeys"] = merge_api_keys(legacy.get("apiKeys")) members = legacy.get("members") or [] if members: root["presets"] = [] for idx, member in enumerate(members[:PRESET_LIMIT], start=1): preset = default_preset(idx, member.get("label")) depart = member.get("depart") or {} arrive = member.get("arrive") or {} routes = member.get("routes") or {} for mode_name in MODE_NAMES: bucket = preset["modes"][mode_name] bus_route = next((r for r in routes.get(mode_name, []) if r.get("type") == "bus"), None) subway_route = next((r for r in routes.get(mode_name, []) if r.get("type") == "subway"), None) if bus_route: bus_cfg = bucket["config"]["bus"] bucket["config"]["transport_type"] = "bus" bus_cfg["from_address"] = _point(depart.get(mode_name), bus_route.get("fromY"), bus_route.get("fromX"), "address") bus_cfg["from_stop"] = _point( bus_route.get("busStopName"), bus_route.get("busStopY"), bus_route.get("busStopX"), "bus_stop", {"id": bus_route.get("busStopId"), "arsId": bus_route.get("busStopId")}, ) bus_cfg["to_address"] = _point(arrive.get(mode_name), None, None, "address") bus_cfg["shared_routes"] = deepcopy(bus_route.get("busRoutes") or []) if subway_route: subway_cfg = bucket["config"]["subway"] if not bus_route: bucket["config"]["transport_type"] = "subway" subway_cfg["from_address"] = _point(depart.get(mode_name), subway_route.get("fromY"), subway_route.get("fromX"), "address") subway_cfg["from_station"] = _point( subway_route.get("stationName"), subway_route.get("stationY"), subway_route.get("stationX"), "subway_station", {"id": subway_route.get("stationId")}, ) via_list = subway_route.get("waypoints") or [] if via_list: subway_cfg["via_station"] = _point(via_list[0], None, None, "subway_station") subway_cfg["to_station"] = _point(subway_route.get("destStation"), None, None, "subway_station") subway_cfg["to_address"] = _point(arrive.get(mode_name), None, None, "address") root["presets"].append(preset) active_index = min(max(int(legacy.get("activeMember", 0) or 0), 0), len(root["presets"]) - 1) root["activePresetId"] = root["presets"][active_index]["id"] return normalize_commute_state(root) def get_api_keys(root): return merge_api_keys((root or {}).get("apiKeys")) def _transit_db_path(script_dir): return os.path.join(script_dir, "transit_data", TRANSIT_DB_FILE) def load_transit_db(script_dir): db_path = _transit_db_path(script_dir) default_db = { "metadata": {"version": 2, "last_sync": None, "last_manual_update": None}, "address_search": {}, "address_geocode": {}, "bus_stop_search": {}, "subway_station_search": {}, "route_search": {}, "common_routes": {}, } if os.path.exists(db_path): try: with open(db_path, "r", encoding="utf-8") as f: raw = json.load(f) if isinstance(raw, dict): for key, value in default_db.items(): if key not in raw or not isinstance(raw[key], type(value)): raw[key] = deepcopy(value) raw["metadata"]["version"] = 2 return raw except Exception: pass return default_db def save_transit_db(script_dir, db): db_path = _transit_db_path(script_dir) os.makedirs(os.path.dirname(db_path), exist_ok=True) with open(db_path, "w", encoding="utf-8") as f: json.dump(db, f, ensure_ascii=False, indent=2) def touch_transit_db(script_dir, manual=False): db = load_transit_db(script_dir) db["metadata"]["last_sync"] = _now_iso() if manual: db["metadata"]["last_manual_update"] = db["metadata"]["last_sync"] save_transit_db(script_dir, db) return db def _cache_key(text): return _normalize_label(text).lower() def _get_cached_section(db, section, key): return deepcopy((db.get(section) or {}).get(_cache_key(key))) def _put_cached_section(db, section, key, value): db.setdefault(section, {})[_cache_key(key)] = deepcopy(value) def _normalize_address_item(doc): road = doc.get("road_address") or {} address = doc.get("address") or {} return { "label": doc.get("place_name") or road.get("address_name") or doc.get("road_address_name") or address.get("address_name") or doc.get("address_name") or "", "address": road.get("address_name") or doc.get("road_address_name") or address.get("address_name") or doc.get("address_name") or "", "lat": _parse_float(doc.get("y") or road.get("y") or address.get("y")), "lng": _parse_float(doc.get("x") or road.get("x") or address.get("x")), } def _dedupe_address_items(items): deduped = [] seen = set() for item in items: key = ( _normalize_label(item.get("label")).lower(), _normalize_label(item.get("address")).lower(), item.get("lat"), item.get("lng"), ) if key in seen: continue seen.add(key) deduped.append(item) return deduped def search_addresses(script_dir, query, api_keys=None): db = load_transit_db(script_dir) cached = _get_cached_section(db, "address_search", query) if cached is not None: return cached headers = { "Authorization": f"KakaoAK {(api_keys or DEFAULT_API_KEYS)['kakao']}", "User-Agent": "Mozilla/5.0", } items = [] address_url = "https://dapi.kakao.com/v2/local/search/address.json?query={}&size=8".format( urllib.parse.quote(query) ) keyword_url = "https://dapi.kakao.com/v2/local/search/keyword.json?query={}&size=8".format( urllib.parse.quote(query) ) try: address_data = _request_json(address_url, timeout=8, headers=headers) items.extend(_normalize_address_item(doc) for doc in address_data.get("documents", [])) except Exception: pass try: keyword_data = _request_json(keyword_url, timeout=8, headers=headers) items.extend(_normalize_address_item(doc) for doc in keyword_data.get("documents", [])) except Exception: pass items = [item for item in _dedupe_address_items(items) if _valid_point(item)] _put_cached_section(db, "address_search", query, items) save_transit_db(script_dir, db) return items def geocode_address(script_dir, query, api_keys=None): db = load_transit_db(script_dir) cached = _get_cached_section(db, "address_geocode", query) if cached is not None: return cached items = search_addresses(script_dir, query, api_keys=api_keys) best = items[0] if items else None _put_cached_section(db, "address_geocode", query, best) save_transit_db(script_dir, db) return best def search_bus_stops(script_dir, name, api_keys=None): db = load_transit_db(script_dir) cached = _get_cached_section(db, "bus_stop_search", name) if cached is not None: return cached url = ( "https://api.odsay.com/v1/api/searchStation" f"?stationName={urllib.parse.quote(name)}" "&stationClass=1" f"&apiKey={urllib.parse.quote((api_keys or DEFAULT_API_KEYS)['odsay'])}" ) data = _request_json(url, timeout=8, use_ssl=True) items = [] for station in (data.get("result") or {}).get("station") or []: items.append( { "name": station.get("stationName") or "", "id": station.get("stationID") or station.get("localStationID") or "", "arsId": station.get("arsID") or station.get("stationID") or "", "lat": _parse_float(station.get("y")), "lng": _parse_float(station.get("x")), } ) _put_cached_section(db, "bus_stop_search", name, items) save_transit_db(script_dir, db) return items def search_subway_stations(script_dir, name, api_keys=None): db = load_transit_db(script_dir) cached = _get_cached_section(db, "subway_station_search", name) if cached is not None: return cached url = ( "https://api.odsay.com/v1/api/searchStation" f"?stationName={urllib.parse.quote(name)}" "&stationClass=2" f"&apiKey={urllib.parse.quote((api_keys or DEFAULT_API_KEYS)['odsay'])}" ) data = _request_json(url, timeout=8, use_ssl=True) items = [] for station in (data.get("result") or {}).get("station") or []: items.append( { "name": station.get("stationName") or "", "id": station.get("stationID") or station.get("localStationID") or "", "lat": _parse_float(station.get("y")), "lng": _parse_float(station.get("x")), "lineNames": station.get("laneName") or "", } ) _put_cached_section(db, "subway_station_search", name, items) save_transit_db(script_dir, db) return items def search_bus_routes(script_dir, name, api_keys=None): db = load_transit_db(script_dir) cached = _get_cached_section(db, "route_search", name) if cached is not None: return cached url = ( "https://api.odsay.com/v1/api/searchLane" f"?busNo={urllib.parse.quote(name)}" f"&apiKey={urllib.parse.quote((api_keys or DEFAULT_API_KEYS)['odsay'])}" ) data = _request_json(url, timeout=8, use_ssl=True) items = [] for lane in (data.get("result") or {}).get("lane") or []: items.append( { "busRouteNm": lane.get("busNo") or "", "busRouteId": lane.get("busID") or lane.get("laneID") or "", "type": lane.get("type") or "", } ) _put_cached_section(db, "route_search", name, items) save_transit_db(script_dir, db) return items def _haversine_m(lat1, lng1, lat2, lng2): radius = 6371000 p1 = math.radians(lat1) p2 = math.radians(lat2) d1 = math.radians(lat2 - lat1) d2 = math.radians(lng2 - lng1) a = math.sin(d1 / 2) ** 2 + math.cos(p1) * math.cos(p2) * math.sin(d2 / 2) ** 2 return radius * 2 * math.atan2(math.sqrt(a), math.sqrt(1 - a)) def _minutes_from_distance(distance_m, speed_mpm=75): return max(1, int(math.ceil((distance_m or 0) / speed_mpm))) def fetch_walking_directions(origin, destination, api_keys=None): if not (_valid_point(origin) and _valid_point(destination)): raise ValueError("도보 경로를 계산할 좌표가 부족합니다.") direct_distance = _haversine_m(origin["lat"], origin["lng"], destination["lat"], destination["lng"]) if direct_distance <= 25: return {"distance_m": int(round(direct_distance)), "duration_minutes": 0, "polyline": None} url = ( "https://maps.googleapis.com/maps/api/directions/json" f"?origin={origin['lat']},{origin['lng']}" f"&destination={destination['lat']},{destination['lng']}" "&mode=walking" f"&key={urllib.parse.quote((api_keys or DEFAULT_API_KEYS)['googleDirections'])}" ) try: data = _request_json(url, timeout=12) routes = data.get("routes") or [] if routes: leg = (routes[0].get("legs") or [{}])[0] distance_m = int((leg.get("distance") or {}).get("value") or 0) duration_s = int((leg.get("duration") or {}).get("value") or 0) return { "distance_m": distance_m, "duration_minutes": max(1, int(math.ceil(duration_s / 60))) if distance_m > 0 else 0, "polyline": routes[0].get("overview_polyline", {}).get("points"), } except Exception: pass return { "distance_m": int(round(direct_distance)), "duration_minutes": 0 if direct_distance <= 25 else _minutes_from_distance(direct_distance), "polyline": None, } def _estimate_ride_eta_minutes(board_point, alight_point, transport_type, api_keys=None): if not (_valid_point(board_point) and _valid_point(alight_point)): return 15 if transport_type == "bus" else 20 try: url = ( "https://api.odsay.com/v1/api/searchPubTransPathT" f"?SX={board_point['lng']}&SY={board_point['lat']}" f"&EX={alight_point['lng']}&EY={alight_point['lat']}" f"&apiKey={urllib.parse.quote((api_keys or DEFAULT_API_KEYS)['odsay'])}" ) data = _request_json(url, timeout=10, use_ssl=True) path = ((data.get("result") or {}).get("path") or [{}])[0] total_time = int(((path.get("info") or {}).get("totalTime")) or 0) if total_time > 0: return total_time except Exception: pass speed = 250 if transport_type == "bus" else 550 distance = _haversine_m(board_point["lat"], board_point["lng"], alight_point["lat"], alight_point["lng"]) return max(3, _minutes_from_distance(distance, speed)) def _as_list(value): if isinstance(value, list): return value if value is None: return [] return [value] def _normalized_name(value): text = _normalize_label(value).lower() for token in (" ", "역", "정류장", "(중)", "(지하)", "(", ")", "-", "_"): text = text.replace(token, "") return text def _name_matches(left, right): a = _normalized_name(left) b = _normalized_name(right) return bool(a and b and (a in b or b in a)) def _point_kind(point): point_type = (point or {}).get("type") if point_type == "subway_station": return "역" if point_type == "bus_stop": return "정류장" if point_type in ("origin", "destination", "address", "gps"): return "도보구간" return "지점" def _clone_point(point, point_type=None, meta=None): if not _valid_point(point): return None merged_meta = deepcopy(point.get("meta") or {}) if meta: merged_meta.update(meta) return _point( point.get("label"), point.get("lat"), point.get("lng"), point_type or point.get("type") or "point", merged_meta or None, ) def _fetch_path_result(board_point, alight_point, api_keys=None): if not (_valid_point(board_point) and _valid_point(alight_point)): return [] url = ( "https://api.odsay.com/v1/api/searchPubTransPathT" f"?SX={board_point['lng']}&SY={board_point['lat']}" f"&EX={alight_point['lng']}&EY={alight_point['lat']}" f"&apiKey={urllib.parse.quote((api_keys or DEFAULT_API_KEYS)['odsay'])}" ) data = _request_json(url, timeout=10, use_ssl=True) return ((data.get("result") or {}).get("path") or []) def _extract_subpath_points(subpath, transport_type): point_type = "bus_stop" if transport_type == "bus" else "subway_station" points = [] stations = (((subpath or {}).get("passStopList") or {}).get("stations") or []) for station in stations: label = station.get("stationName") or station.get("name") or "" point = _point( label, station.get("y"), station.get("x"), point_type, { "id": station.get("stationID") or station.get("stationId"), "station_index": station.get("index"), "source": "odsay_pass_stop", }, ) if _valid_point(point): points.append(point) return points def _subpath_endpoint_point(subpath, prefix, fallback_type): return _point( (subpath or {}).get(f"{prefix}Name"), (subpath or {}).get(f"{prefix}Y"), (subpath or {}).get(f"{prefix}X"), fallback_type, {"source": "odsay_subpath_endpoint"}, ) def _build_subway_subpath_segment(subpath): route_label = " ".join(_normalize_label(lane.get("name") or lane.get("busNo")) for lane in _lane_items(subpath)) points = _extract_subpath_points(subpath, "subway") start_point = _subpath_endpoint_point(subpath, "start", "subway_station") end_point = _subpath_endpoint_point(subpath, "end", "subway_station") if points: points = _ensure_named_endpoint(points, start_point, 0) points = _ensure_named_endpoint(points, end_point, max(0, len(points) - 1)) else: points = [point for point in (_clone_point(start_point, "subway_station"), _clone_point(end_point, "subway_station")) if point] return { "kind": "subway", "route_label": route_label or _normalize_label((subpath or {}).get("name"), "지하철"), "line_key": _extract_subway_line_key(route_label, (subpath or {}).get("name")), "duration_minutes": int((subpath or {}).get("sectionTime") or 0), "points": points, "start_label": (points[0] if points else start_point or {}).get("label") or "", "end_label": (points[-1] if points else end_point or {}).get("label") or "", } def _append_unique_points(target_points, new_points): for point in new_points or []: if not point: continue cloned = _clone_point(point, point.get("type") or "point") if not cloned: continue if target_points: prev = target_points[-1] same_coord = cloned.get("lat") == prev.get("lat") and cloned.get("lng") == prev.get("lng") same_name = _name_matches(cloned.get("label"), prev.get("label")) if same_coord or same_name: prev.setdefault("meta", {}).update(deepcopy(cloned.get("meta") or {})) continue target_points.append(cloned) return target_points def _select_subway_path(board_point, alight_point, via_point=None, api_keys=None): best = None for path in _fetch_path_result(board_point, alight_point, api_keys=api_keys): subpaths = _as_list(path.get("subPath")) subway_subpaths = [subpath for subpath in subpaths if int(subpath.get("trafficType") or 0) == 1] if not subway_subpaths: continue score = 0 first_subway = subway_subpaths[0] last_subway = subway_subpaths[-1] if _name_matches(first_subway.get("startName"), board_point.get("label")): score += 8 if _name_matches(last_subway.get("endName"), alight_point.get("label")): score += 8 if via_point: via_hits = 0 for subpath in subway_subpaths: pass_points = _extract_subpath_points(subpath, "subway") if any(_name_matches(point.get("label"), via_point.get("label")) for point in pass_points): via_hits += 1 if _name_matches(subpath.get("startName"), via_point.get("label")) or _name_matches(subpath.get("endName"), via_point.get("label")): via_hits += 1 score += min(via_hits, 2) * 4 total_time = sum(int(subpath.get("sectionTime") or 0) for subpath in subpaths if int(subpath.get("trafficType") or 0) in (1, 3)) candidate = { "score": score, "subpaths": subpaths, "total_time": total_time, } if best is None or candidate["score"] > best["score"] or (candidate["score"] == best["score"] and candidate["total_time"] < best["total_time"]): best = candidate return best def _build_subway_journey(board_point, alight_point, via_point=None, api_keys=None): path_match = _select_subway_path(board_point, alight_point, via_point=via_point, api_keys=api_keys) if not path_match: fallback_points = [_clone_point(board_point, "subway_station"), _clone_point(alight_point, "subway_station")] return { "journey_segments": [], "points": [point for point in fallback_points if point], "section_time": 0, "lane_label": "", } raw_segments = [] for subpath in path_match.get("subpaths") or []: traffic_type = int(subpath.get("trafficType") or 0) if traffic_type == 1: raw_segments.append({"kind": "subway", "subpath": subpath}) elif traffic_type == 3: raw_segments.append({"kind": "walk", "subpath": subpath}) built_subway = {} for idx, entry in enumerate(raw_segments): if entry["kind"] == "subway": built_subway[idx] = _build_subway_subpath_segment(entry["subpath"]) subway_indexes = sorted(built_subway.keys()) if subway_indexes: first_segment = built_subway[subway_indexes[0]] last_segment = built_subway[subway_indexes[-1]] first_segment["points"] = _ensure_named_endpoint(first_segment.get("points") or [], board_point, 0) last_segment["points"] = _ensure_named_endpoint(last_segment.get("points") or [], alight_point, max(0, len(last_segment.get("points") or []) - 1)) first_segment["start_label"] = (first_segment.get("points") or [board_point])[0].get("label") if (first_segment.get("points") or []) else board_point.get("label") last_segment["end_label"] = (last_segment.get("points") or [alight_point])[-1].get("label") if (last_segment.get("points") or []) else alight_point.get("label") journey_segments = [] total_minutes = 0 lane_labels = [] combined_points = [] for idx, entry in enumerate(raw_segments): if entry["kind"] == "subway": segment = deepcopy(built_subway.get(idx) or {}) if not segment: continue segment_points = [] start_index = None end_index = None for point in segment.get("points") or []: cloned = _clone_point(point, point.get("type") or "subway_station") if not cloned: continue if combined_points: prev = combined_points[-1] same_coord = cloned.get("lat") == prev.get("lat") and cloned.get("lng") == prev.get("lng") same_name = _name_matches(cloned.get("label"), prev.get("label")) if same_coord or same_name: segment_points.append(deepcopy(prev)) reused_index = (prev.get("meta") or {}).get("global_index") if start_index is None: start_index = reused_index end_index = reused_index continue cloned.setdefault("meta", {})["global_index"] = len(combined_points) combined_points.append(cloned) segment_points.append(deepcopy(cloned)) if start_index is None: start_index = (cloned.get("meta") or {}).get("global_index") end_index = (cloned.get("meta") or {}).get("global_index") segment["points"] = segment_points segment["start_index"] = start_index segment["end_index"] = end_index journey_segments.append(segment) total_minutes += int(segment.get("duration_minutes") or 0) if segment.get("route_label"): lane_labels.append(segment.get("route_label")) continue prev_subway = None next_subway = None for candidate_idx in range(idx - 1, -1, -1): if candidate_idx in built_subway: prev_subway = built_subway[candidate_idx] break for candidate_idx in range(idx + 1, len(raw_segments)): if candidate_idx in built_subway: next_subway = built_subway[candidate_idx] break if not (prev_subway and next_subway): continue from_point = ((prev_subway.get("points") or [])[-1]) if prev_subway.get("points") else None to_point = ((next_subway.get("points") or [])[0]) if next_subway.get("points") else None walk_info = {"distance_m": 0, "duration_minutes": 0, "polyline": None} if _valid_point(from_point) and _valid_point(to_point): try: walk_info = fetch_walking_directions(from_point, to_point, api_keys=api_keys) except Exception: walk_info = { "distance_m": int((entry.get("subpath") or {}).get("distance") or 0), "duration_minutes": int((entry.get("subpath") or {}).get("sectionTime") or 0), "polyline": None, } walk_segment = { "kind": "walk", "from_label": (from_point or {}).get("label") or (prev_subway.get("end_label") or ""), "to_label": (to_point or {}).get("label") or (next_subway.get("start_label") or ""), "duration_minutes": int(walk_info.get("duration_minutes") or (entry.get("subpath") or {}).get("sectionTime") or 0), "distance_m": int(walk_info.get("distance_m") or (entry.get("subpath") or {}).get("distance") or 0), } journey_segments.append(walk_segment) total_minutes += int(walk_segment.get("duration_minutes") or 0) return { "journey_segments": journey_segments, "points": combined_points, "section_time": total_minutes, "lane_label": " / ".join(label for label in lane_labels if label), } def _ensure_named_endpoint(points, target_point, index_hint): if not _valid_point(target_point): return points target_clone = _clone_point(target_point, target_point.get("type")) if not points: return [target_clone] compare_index = max(0, min(index_hint, len(points) - 1)) if _name_matches(points[compare_index].get("label"), target_point.get("label")): points[compare_index]["label"] = target_point.get("label") or points[compare_index].get("label") points[compare_index]["lat"] = target_point.get("lat") points[compare_index]["lng"] = target_point.get("lng") points[compare_index].setdefault("meta", {}).update(deepcopy(target_point.get("meta") or {})) return points if index_hint <= 0: return [target_clone] + points return points + [target_clone] def _lane_items(subpath): return [lane for lane in _as_list((subpath or {}).get("lane")) if isinstance(lane, dict)] def _select_transit_subpath(board_point, alight_point, transport_type, route_name="", route_id="", via_label="", api_keys=None): target_traffic = 2 if transport_type == "bus" else 1 best = None for path in _fetch_path_result(board_point, alight_point, api_keys=api_keys): for subpath in _as_list(path.get("subPath")): if int(subpath.get("trafficType") or 0) != target_traffic: continue lane_names = " ".join(_normalize_label(lane.get("name") or lane.get("busNo")) for lane in _lane_items(subpath)) lane_ids = { str(lane.get("busID") or lane.get("busRouteId") or lane.get("laneID") or "") for lane in _lane_items(subpath) if lane.get("busID") or lane.get("busRouteId") or lane.get("laneID") } pass_points = _extract_subpath_points(subpath, transport_type) score = 0 if _name_matches(subpath.get("startName"), board_point.get("label")): score += 4 if _name_matches(subpath.get("endName"), alight_point.get("label")): score += 4 if route_name and route_name in lane_names: score += 8 if route_id and str(route_id) in lane_ids: score += 10 if via_label and any(_name_matches(point.get("label"), via_label) for point in pass_points): score += 6 if pass_points: if any(_name_matches(point.get("label"), board_point.get("label")) for point in pass_points): score += 2 if any(_name_matches(point.get("label"), alight_point.get("label")) for point in pass_points): score += 2 candidate = { "score": score, "section_time": int(subpath.get("sectionTime") or 0), "subpath": subpath, "lane_names": lane_names, "pass_points": pass_points, } if best is None: best = candidate continue if candidate["score"] > best["score"]: best = candidate continue if candidate["score"] == best["score"] and candidate["section_time"] > 0: best_time = best["section_time"] or 999999 if candidate["section_time"] < best_time: best = candidate return best def _select_transit_path(board_point, alight_point, transport_type, route_name="", route_id="", via_label="", api_keys=None): target_traffic = 2 if transport_type == "bus" else 1 relevant_traffic = {target_traffic, 3} if transport_type == "subway" else {target_traffic} best = None for path in _fetch_path_result(board_point, alight_point, api_keys=api_keys): subpaths = [subpath for subpath in _as_list(path.get("subPath")) if int(subpath.get("trafficType") or 0) in relevant_traffic] transit_subpaths = [subpath for subpath in subpaths if int(subpath.get("trafficType") or 0) == target_traffic] if not transit_subpaths: continue lane_names = " ".join( _normalize_label(lane.get("name") or lane.get("busNo")) for subpath in transit_subpaths for lane in _lane_items(subpath) if lane.get("name") or lane.get("busNo") ) lane_ids = { str(lane.get("busID") or lane.get("busRouteId") or lane.get("laneID") or "") for subpath in transit_subpaths for lane in _lane_items(subpath) if lane.get("busID") or lane.get("busRouteId") or lane.get("laneID") } pass_points = [] for subpath in transit_subpaths: pass_points.extend(_extract_subpath_points(subpath, transport_type)) score = 0 if _name_matches(transit_subpaths[0].get("startName"), board_point.get("label")): score += 4 if _name_matches(transit_subpaths[-1].get("endName"), alight_point.get("label")): score += 4 if route_name and route_name in lane_names: score += 8 if route_id and str(route_id) in lane_ids: score += 10 if via_label and any(_name_matches(point.get("label"), via_label) for point in pass_points): score += 6 if pass_points: if any(_name_matches(point.get("label"), board_point.get("label")) for point in pass_points): score += 2 if any(_name_matches(point.get("label"), alight_point.get("label")) for point in pass_points): score += 2 candidate = { "score": score, "section_time": sum(int(subpath.get("sectionTime") or 0) for subpath in subpaths), "path": path, "subpaths": subpaths, "lane_names": lane_names, "pass_points": pass_points, } if best is None: best = candidate continue if candidate["score"] > best["score"]: best = candidate continue if candidate["score"] == best["score"] and candidate["section_time"] > 0: best_time = best["section_time"] or 999999 if candidate["section_time"] < best_time: best = candidate return best def _path_total_time(path): info = (path or {}).get("info") or {} total_time = int(info.get("totalTime") or 0) if total_time > 0: return total_time return sum(int((subpath or {}).get("sectionTime") or 0) for subpath in _as_list((path or {}).get("subPath"))) def _subpath_lane_label(subpath): lane_labels = [ _normalize_label(lane.get("name") or lane.get("busNo")) for lane in _lane_items(subpath) if lane.get("name") or lane.get("busNo") ] return " ".join(label for label in lane_labels if label) def _subpath_line_key(subpath): for lane in _lane_items(subpath): line_key = _extract_subway_line_key( lane.get("subwayCode"), lane.get("subwayID"), lane.get("name"), ) if line_key: return line_key return _extract_subway_line_key(_subpath_lane_label(subpath)) def _merge_unique_points_with_indexes(existing_points, new_points): merged = list(existing_points or []) start_index = len(merged) for idx, point in enumerate(new_points or []): candidate = _clone_point(point, point.get("type")) if not candidate: continue if merged: prev = merged[-1] same_coord = prev.get("lat") == candidate.get("lat") and prev.get("lng") == candidate.get("lng") same_name = _name_matches(prev.get("label"), candidate.get("label")) if same_coord or same_name: prev.setdefault("meta", {}).update(candidate.get("meta") or {}) if idx == 0: start_index = len(merged) - 1 continue merged.append(candidate) return merged, start_index, len(merged) - 1 if merged else None def _subpath_endpoint_point_from_points(subpath, key_name, fallback_points, fallback_index): label = (subpath or {}).get(key_name) or "" if fallback_points and 0 <= fallback_index < len(fallback_points): point = _clone_point(fallback_points[fallback_index], fallback_points[fallback_index].get("type")) if point and label: point["label"] = label return point point_type = "subway_station" if int((subpath or {}).get("trafficType") or 0) == 1 else "point" return _point(label, None, None, point_type) if label else None def _select_subway_journey_path(board_point, alight_point, via_label="", api_keys=None): best = None for path in _fetch_path_result(board_point, alight_point, api_keys=api_keys): subpaths = _as_list(path.get("subPath")) subway_subpaths = [subpath for subpath in subpaths if int(subpath.get("trafficType") or 0) == 1] if not subway_subpaths: continue score = 0 if _name_matches((subway_subpaths[0] or {}).get("startName"), board_point.get("label")): score += 8 if _name_matches((subway_subpaths[-1] or {}).get("endName"), alight_point.get("label")): score += 8 all_points = [] for subpath in subway_subpaths: pass_points = _extract_subpath_points(subpath, "subway") all_points.extend(pass_points) lane_label = _subpath_lane_label(subpath) if lane_label: score += 1 if any(_name_matches(point.get("label"), board_point.get("label")) for point in pass_points): score += 2 if any(_name_matches(point.get("label"), alight_point.get("label")) for point in pass_points): score += 2 if via_label and any(_name_matches(point.get("label"), via_label) for point in pass_points): score += 5 candidate = { "path": path, "subpaths": subpaths, "score": score, "total_time": _path_total_time(path), } if best is None or candidate["score"] > best["score"] or ( candidate["score"] == best["score"] and candidate["total_time"] < best["total_time"] ): best = candidate return best def _build_subway_journey_meta(board_point, alight_point, destination_point, via_point=None, api_keys=None): selected = _select_subway_journey_path( board_point, alight_point, via_label=(via_point or {}).get("label") if via_point else "", api_keys=api_keys, ) if not selected: return None journey_segments = [] merged_transit_points = [] previous_subway = None subpaths = selected.get("subpaths") or [] subway_indices = [idx for idx, subpath in enumerate(subpaths) if int((subpath or {}).get("trafficType") or 0) == 1] if not subway_indices: return None first_subway_index = subway_indices[0] last_subway_index = subway_indices[-1] for idx, subpath in enumerate(subpaths): traffic_type = int(subpath.get("trafficType") or 0) if traffic_type == 1: segment_points = _extract_subpath_points(subpath, "subway") start_hint = board_point if idx == first_subway_index else _subpath_endpoint_point_from_points(subpath, "startName", segment_points, 0) end_hint = alight_point if idx == last_subway_index else _subpath_endpoint_point_from_points(subpath, "endName", segment_points, max(0, len(segment_points) - 1)) segment_points = _ensure_named_endpoint(segment_points, start_hint, 0) if start_hint else segment_points segment_points = _ensure_named_endpoint(segment_points, end_hint, max(0, len(segment_points) - 1)) if end_hint else segment_points merged_transit_points, start_index, end_index = _merge_unique_points_with_indexes(merged_transit_points, segment_points) line_label = _subpath_lane_label(subpath) or "지하철" journey_segments.append({ "kind": "subway", "line_label": line_label, "line_key": _subpath_line_key(subpath), "board_label": (start_hint or {}).get("label") or subpath.get("startName") or (segment_points[0].get("label") if segment_points else ""), "alight_label": (end_hint or {}).get("label") or subpath.get("endName") or (segment_points[-1].get("label") if segment_points else ""), "duration_minutes": int(subpath.get("sectionTime") or 0), "distance_m": int(subpath.get("distance") or subpath.get("sectionDistance") or 0), "points": segment_points, "global_start_index": start_index, "global_end_index": end_index, }) previous_subway = journey_segments[-1] elif traffic_type == 3 and previous_subway: next_subway = None for next_subpath in subpaths[idx + 1:]: if int((next_subpath or {}).get("trafficType") or 0) == 1: next_subway = next_subpath break if not next_subway: continue next_points = _extract_subpath_points(next_subpath, "subway") from_point = ((previous_subway.get("points") or [])[-1]) if previous_subway.get("points") else None to_point = _subpath_endpoint_point_from_points(next_subpath, "startName", next_points, 0) walk_duration = int(subpath.get("sectionTime") or 0) walk_distance = int(subpath.get("distance") or subpath.get("sectionDistance") or 0) if walk_distance <= 0 and _valid_point(from_point) and _valid_point(to_point): try: walking = fetch_walking_directions(from_point, to_point, api_keys=api_keys) walk_duration = int(walking.get("duration_minutes") or walk_duration or 0) walk_distance = int(walking.get("distance_m") or 0) except Exception: pass to_label = next_subway.get("startName") or previous_subway.get("alight_label") or "" journey_segments.append({ "kind": "walk", "from_label": previous_subway.get("alight_label") or "", "to_label": to_label, "duration_minutes": walk_duration, "distance_m": walk_distance, }) if not merged_transit_points: return None first_subway = next((segment for segment in journey_segments if segment.get("kind") == "subway"), None) return { "journey_segments": journey_segments, "points": _compose_segment_points( board_point, merged_transit_points, destination_point, via_label=(via_point or {}).get("label") if via_point else "", ), "section_time": sum(int((segment or {}).get("duration_minutes") or 0) for segment in journey_segments), "lane_label": (first_subway or {}).get("line_label") or "", "primary_line_key": (first_subway or {}).get("line_key") or "", } def _compose_segment_points(origin_point, transit_points, destination_point, via_label=""): points = [] if _valid_point(origin_point): points.append(_clone_point(origin_point, "origin")) normalized_transit = [] for idx, point in enumerate(transit_points or []): cloned = _clone_point(point, point.get("type") or "point", {"segment_index": idx}) if not cloned: continue role = "pass" if idx == 0: role = "board" elif idx == len(transit_points) - 1: role = "alight" if via_label and _name_matches(cloned.get("label"), via_label): role = "via" cloned.setdefault("meta", {})["role"] = role normalized_transit.append(cloned) points.extend(normalized_transit) if _valid_point(destination_point): points.append(_clone_point(destination_point, "destination")) deduped = [] for point in points: if deduped: prev = deduped[-1] same_coord = point.get("lat") == prev.get("lat") and point.get("lng") == prev.get("lng") same_name = _name_matches(point.get("label"), prev.get("label")) if same_coord or same_name: if prev.get("type") in ("origin", "destination") and point.get("type") not in ("origin", "destination"): deduped[-1] = point else: prev_meta = prev.setdefault("meta", {}) prev_meta.update(point.get("meta") or {}) continue deduped.append(point) return deduped def _subpath_route_label(subpath, transport_type): labels = [ _normalize_label(lane.get("name") or lane.get("busNo")) for lane in _lane_items(subpath) if lane.get("name") or lane.get("busNo") ] if labels: return " ".join(labels) return "버스" if transport_type == "bus" else "지하철" def _build_transit_sections(origin_point, board_point, alight_point, destination_point, transport_type, via_point=None, route_name="", route_id="", api_keys=None): path_match = _select_transit_path( board_point, alight_point, transport_type, route_name=route_name, route_id=route_id, via_label=(via_point or {}).get("label") if via_point else "", api_keys=api_keys, ) if not path_match: fallback = _segment_points( origin_point, board_point, alight_point, destination_point, transport_type, via_point=via_point, route_name=route_name, route_id=route_id, api_keys=api_keys, ) return { "sections": [], "points": fallback.get("points") or [], "section_time": fallback.get("section_time") or 0, "lane_label": fallback.get("lane_label") or route_name or ("버스" if transport_type == "bus" else "지하철"), } target_traffic = 2 if transport_type == "bus" else 1 relevant_subpaths = path_match.get("subpaths") or [] transit_subpaths = [subpath for subpath in relevant_subpaths if int(subpath.get("trafficType") or 0) == target_traffic] transit_total = len(transit_subpaths) transit_seen = 0 sections = [] all_transit_points = [] section_index = 0 for subpath in relevant_subpaths: traffic_type = int(subpath.get("trafficType") or 0) if traffic_type == 3 and transport_type == "subway": from_label = subpath.get("startName") or (sections[-1].get("alight_label") if sections else board_point.get("label")) to_label = subpath.get("endName") or "" sections.append( { "type": "walk", "section_index": section_index, "from_label": from_label, "to_label": to_label, "duration_minutes": int(subpath.get("sectionTime") or 0), "distance_m": int(float(subpath.get("distance") or 0)), } ) section_index += 1 continue if traffic_type != target_traffic: continue transit_seen += 1 route_label = _subpath_route_label(subpath, transport_type) or route_name or ("버스" if transport_type == "bus" else "지하철") line_key = _extract_subway_line_key(route_label) if transport_type == "subway" else "" points = _extract_subpath_points(subpath, transport_type) if transit_seen == 1: points = _ensure_named_endpoint(points, board_point, 0) if transit_seen == transit_total: points = _ensure_named_endpoint(points, alight_point, max(0, len(points) - 1)) annotated_points = [] for point_index, point in enumerate(points): annotated = _clone_point( point, point.get("type"), { "section_index": section_index, "section_point_index": point_index, "line_key": line_key, "section_route_label": route_label, }, ) if not annotated: continue annotated_points.append(annotated) all_transit_points.append(_clone_point(annotated, annotated.get("type"))) board_label = (annotated_points[0].get("label") if annotated_points else subpath.get("startName")) or board_point.get("label") alight_label = (annotated_points[-1].get("label") if annotated_points else subpath.get("endName")) or alight_point.get("label") sections.append( { "type": transport_type, "section_index": section_index, "route_label": route_label, "line_key": line_key, "board_label": board_label, "alight_label": alight_label, "duration_minutes": int(subpath.get("sectionTime") or 0), "points": annotated_points, } ) section_index += 1 return { "sections": sections, "points": _compose_segment_points(origin_point, [point for point in all_transit_points if point], destination_point, via_label=(via_point or {}).get("label") if via_point else ""), "section_time": path_match.get("section_time") or 0, "lane_label": path_match.get("lane_names") or route_name or ("버스" if transport_type == "bus" else "지하철"), } def _bus_arrival_msg_to_seconds(msg): if not msg: return 9999 if any(word in msg for word in ("운행종료", "운행중단", "막차", "출발대기")): return 9999 import re minute_match = re.search(r"(\d+)분", msg) second_match = re.search(r"(\d+)초", msg) minutes = int(minute_match.group(1)) if minute_match else 0 seconds = int(second_match.group(1)) if second_match else 0 if minutes or seconds: return minutes * 60 + seconds if "곧" in msg or "즉시" in msg: return 30 return 9999 def _raise_bus_api_error(data): header = data.get("msgHeader") or {} header_code = str(header.get("headerCd") or "") if header_code and header_code != "0": raise ValueError(header.get("headerMsg") or f"버스 도착정보 조회 실패 (headerCd={header_code})") def _normalize_bus_arrival_items(items): trips = [] for item in items: route_name = item.get("rtNm") or item.get("busRouteAbrv") or item.get("busRouteNm") or "" route_id = item.get("busRouteId") or item.get("routeId") or "" for order, key in enumerate(("arrmsg1", "arrmsg2"), start=1): eta_text = item.get(key) or "" eta_seconds = _bus_arrival_msg_to_seconds(eta_text) if eta_seconds >= 9999: continue trips.append( { "route_name": route_name, "route_id": route_id, "eta_seconds": eta_seconds, "eta_text": eta_text, "vehicle_order": order, "direction": item.get("dir") or item.get("adirection") or "", "station_label": item.get(f"stationNm{order}") or item.get("stNm") or item.get("staNm") or "", } ) trips.sort(key=lambda x: x["eta_seconds"]) return trips def _fetch_bus_arrivals_by_ars(stop, api_keys): ars_id = (stop.get("meta") or {}).get("arsId") or stop.get("arsId") if not ars_id: return [] url = ( "http://ws.bus.go.kr/api/rest/stationinfo/getStationByUid" f"?ServiceKey={urllib.parse.quote(api_keys['busArrive'], safe='')}" f"&arsId={urllib.parse.quote(str(ars_id), safe='')}&resultType=json" ) data = _request_json(url, timeout=8) _raise_bus_api_error(data) return _normalize_bus_arrival_items((data.get("msgBody") or {}).get("itemList") or []) def _fetch_bus_arrivals_by_station_id(stop, api_keys): station_id = (stop.get("meta") or {}).get("id") or stop.get("id") if not station_id: return [] url = ( "http://ws.bus.go.kr/api/rest/arrive/getLowArrInfoByStId" f"?ServiceKey={urllib.parse.quote(api_keys['busArrive'], safe='')}" f"&stId={urllib.parse.quote(str(station_id), safe='')}&resultType=json" ) data = _request_json(url, timeout=8) _raise_bus_api_error(data) return _normalize_bus_arrival_items((data.get("msgBody") or {}).get("itemList") or []) def fetch_bus_arrivals(stop, api_keys=None): api_keys = api_keys or DEFAULT_API_KEYS has_ars = (stop.get("meta") or {}).get("arsId") or stop.get("arsId") has_station_id = (stop.get("meta") or {}).get("id") or stop.get("id") if not (has_ars or has_station_id): return [] errors = [] for fetcher in (_fetch_bus_arrivals_by_ars, _fetch_bus_arrivals_by_station_id): try: trips = fetcher(stop, api_keys) if trips: return trips except Exception as exc: errors.append(str(exc)) if errors: raise ValueError(errors[0]) return [] def _subway_eta_seconds(item): eta_seconds = int(item.get("barvlDt") or 0) if eta_seconds > 0: return eta_seconds message = str(item.get("arvlMsg2") or "") code = str(item.get("arvlCd") or "") if "진입" in message or code == "0": return 30 if "도착" in message or code == "1": return 45 if "전역출발" in message: return 90 if "전역진입" in message: return 60 if "출발" in message or code == "2": return 180 return 0 def _extract_subway_line_key(*values): for raw in values: text = str(raw or "").strip() if not text: continue mapped = SUBWAY_ID_TO_KEY.get(text) if mapped: return mapped normalized = text.replace(" ", "") for key in SUBWAY_LINE_KEYS: if key.replace(" ", "") in normalized: return key return "" def fetch_subway_arrivals(station_name, via_name="", dest_name="", api_keys=None): if not station_name: return [] url = ( "http://swopenapi.seoul.go.kr/api/subway/" f"{urllib.parse.quote((api_keys or DEFAULT_API_KEYS)['subway'])}" f"/json/realtimeStationArrival/0/20/{urllib.parse.quote(station_name)}" ) data = _request_json(url, timeout=8) items = data.get("realtimeArrivalList") or [] all_trips = [] filtered_trips = [] filters = [value for value in (via_name, dest_name) if value] for item in items: haystack = " ".join(str(item.get(key) or "") for key in ("trainLineNm", "bstatnNm", "arvlMsg2", "arvlMsg3")) eta_seconds = _subway_eta_seconds(item) if eta_seconds <= 0: continue trip = { "route_name": item.get("trainLineNm") or "", "eta_seconds": eta_seconds, "eta_text": item.get("arvlMsg2") or "", "station_label": item.get("arvlMsg3") or "", "destination": item.get("bstatnNm") or "", "train_no": item.get("btrainNo") or item.get("trainNo") or "", "arrival_code": str(item.get("arvlCd") or ""), "received_at": item.get("recptnDt") or "", "updn_line": item.get("updnLine") or "", "is_express": str(item.get("btrainSttus") or item.get("directAt") or "") in ("1", "7") or "급행" in str(item.get("trainLineNm") or ""), "line_key": _extract_subway_line_key( item.get("subwayId"), item.get("subwayNm"), item.get("trainLineNm"), ), } all_trips.append(trip) if not filters or any(keyword in haystack for keyword in filters): filtered_trips.append(trip) trips = filtered_trips or all_trips trips.sort(key=lambda x: x["eta_seconds"]) return trips def fetch_subway_positions(line_name, api_keys=None): line_name = _extract_subway_line_key(line_name) if not line_name: return [] url = ( "http://swopenapi.seoul.go.kr/api/subway/" f"{urllib.parse.quote((api_keys or DEFAULT_API_KEYS)['subway'])}" f"/json/realtimePosition/0/60/{urllib.parse.quote(line_name)}" ) data = _request_json(url, timeout=8) items = data.get("realtimePositionList") or [] positions = [] for item in items: positions.append( { "train_no": item.get("trainNo") or item.get("btrainNo") or "", "station_label": item.get("statnNm") or "", "station_id": item.get("statnId") or "", "line_key": _extract_subway_line_key(item.get("subwayId"), item.get("subwayNm"), line_name), "status_code": str(item.get("trainSttus") or ""), "status_text": item.get("trainSttus") or "", "updn_line": item.get("updnLine") or "", "direct_at": item.get("directAt") or "", "received_at": item.get("recptnDt") or "", } ) return positions def _point_from_config(value, fallback_type): if isinstance(value, dict): meta = deepcopy(value.get("meta") or {}) if value.get("id") and "id" not in meta: meta["id"] = value.get("id") if value.get("arsId") and "arsId" not in meta: meta["arsId"] = value.get("arsId") if value.get("lineNames") and "lineNames" not in meta: meta["lineNames"] = value.get("lineNames") if value.get("lineName") and "lineName" not in meta: meta["lineName"] = value.get("lineName") return _point(value.get("label") or value.get("name"), value.get("lat"), value.get("lng"), value.get("type") or fallback_type, meta) return None def _choose_origin_point(origin_mode, manual_point, current_position): if origin_mode == "gps" and _valid_point(current_position): return _point("GPS 현재위치 사용", current_position["lat"], current_position["lng"], "gps") if _valid_point(manual_point): return manual_point if _valid_point(current_position): return _point("GPS 현재위치 사용", current_position["lat"], current_position["lng"], "gps") raise ValueError("출발지 좌표를 확인할 수 없습니다.") def _filter_bus_trips(trips, shared_routes): if not shared_routes: return trips route_names = {item.get("busRouteNm") for item in shared_routes if item.get("busRouteNm")} route_ids = {item.get("busRouteId") for item in shared_routes if item.get("busRouteId")} filtered = [trip for trip in trips if trip.get("route_name") in route_names or trip.get("route_id") in route_ids] return filtered or trips def _pick_boardable_trip(candidates, ready_seconds): if not candidates: return None for trip in candidates: if trip["eta_seconds"] >= ready_seconds: return trip return candidates[0] def _annotate_candidate_trips(candidates, ready_seconds): annotated = [] selected_index = 0 found = False for idx, trip in enumerate(candidates or []): annotated_trip = deepcopy(trip) selectable = int(annotated_trip.get("eta_seconds") or 0) >= int(ready_seconds or 0) annotated_trip["candidate_index"] = idx annotated_trip["selectable"] = selectable annotated.append(annotated_trip) if selectable and not found: selected_index = idx found = True return annotated, selected_index def _retain_candidate_window(candidates, selected_index, minimum_size=8): if not candidates: return [], 0 keep_count = max(int(minimum_size or 0), int(selected_index or 0) + 1) kept = deepcopy(candidates[:keep_count]) if not kept: return [], 0 return kept, max(0, min(int(selected_index or 0), len(kept) - 1)) def _attach_subway_segment_candidates(journey_segments, api_keys=None): attached = [] for idx, segment in enumerate(deepcopy(journey_segments or [])): if segment.get("kind") != "subway": attached.append(segment) continue ready_seconds = 0 if idx > 0: previous = (journey_segments or [])[idx - 1] or {} if previous.get("kind") == "walk": ready_seconds = int(previous.get("duration_minutes") or 0) * 60 try: trips = fetch_subway_arrivals(segment.get("board_label") or "", "", segment.get("alight_label") or "", api_keys=api_keys) except Exception: trips = [] candidate_trips, selected_index = _annotate_candidate_trips(trips, ready_seconds) candidate_trips, selected_index = _retain_candidate_window(candidate_trips, selected_index, minimum_size=6) segment["candidate_trips"] = candidate_trips segment["selected_trip_index"] = selected_index attached.append(segment) return attached def _plan_time_base(plan): snapshot = plan.get("snapshot_created_at") if snapshot: try: return datetime.fromisoformat(snapshot) except Exception: pass return datetime.now() def _apply_selected_trip(plan, selected_index=None): candidates = deepcopy(plan.get("candidate_trips") or []) if not candidates: return plan if selected_index is None: selected_index = int(plan.get("selected_trip_index") or 0) selected_index = max(0, min(int(selected_index), len(candidates) - 1)) selected_trip = deepcopy(candidates[selected_index]) time_base = _plan_time_base(plan) board_time = time_base + timedelta(seconds=int(selected_trip.get("eta_seconds") or 0)) arrival_time = board_time + timedelta(minutes=int((plan.get("ride_eta_minutes") or 0) + (plan.get("walk_after_alight_minutes") or 0))) route_label = selected_trip.get("route_name") or (plan.get("summary") or {}).get("route_label") or ("버스" if plan.get("transport_type") == "bus" else "지하철") subway_line_key = "" if plan.get("transport_type") == "subway": subway_line_key = _extract_subway_line_key( selected_trip.get("line_key"), (plan.get("summary") or {}).get("subway_line_key"), route_label, ((plan.get("board_point") or {}).get("meta") or {}).get("lineName"), ((plan.get("board_point") or {}).get("meta") or {}).get("lineNames"), ((plan.get("alight_point") or {}).get("meta") or {}).get("lineName"), ((plan.get("alight_point") or {}).get("meta") or {}).get("lineNames"), ) if subway_line_key: selected_trip["line_key"] = subway_line_key plan["selected_trip_index"] = selected_index plan["boardable_trip"] = selected_trip plan["board_time"] = board_time.isoformat() plan["arrival_time"] = arrival_time.isoformat() if plan.get("transport_type") == "bus": plan["vehicle_marker"] = { "type": "bus", "label": route_label, "eta_text": selected_trip.get("eta_text") or "", "direction": selected_trip.get("direction") or "", } else: plan["vehicle_marker"] = { "type": "subway", "label": route_label, "eta_text": selected_trip.get("eta_text") or "", "destination": selected_trip.get("destination") or "", "line_key": subway_line_key, } plan["summary"] = { "route_label": route_label, "board_time_hhmm": _format_hhmm(plan["board_time"]), "arrival_time_hhmm": _format_hhmm(plan["arrival_time"]), } if subway_line_key: plan["summary"]["subway_line_key"] = subway_line_key return plan def _format_hhmm(iso_value): if not iso_value: return "--:--" try: return datetime.fromisoformat(iso_value).strftime("%H:%M") except Exception: return "--:--" def _segment_points(origin_point, board_point, alight_point, destination_point, transport_type, via_point=None, route_name="", route_id="", api_keys=None): path_match = _select_transit_subpath( board_point, alight_point, transport_type, route_name=route_name, route_id=route_id, via_label=(via_point or {}).get("label") if via_point else "", api_keys=api_keys, ) transit_points = [] section_time = 0 lane_label = route_name or "" if path_match: transit_points = path_match.get("pass_points") or [] transit_points = _ensure_named_endpoint(transit_points, board_point, 0) transit_points = _ensure_named_endpoint(transit_points, alight_point, max(0, len(transit_points) - 1)) section_time = path_match.get("section_time") or 0 lane_label = path_match.get("lane_names") or lane_label if not transit_points: transit_points = [_clone_point(board_point, board_point.get("type")), _clone_point(alight_point, alight_point.get("type"))] return { "points": _compose_segment_points(origin_point, [point for point in transit_points if point], destination_point, via_label=(via_point or {}).get("label") if via_point else ""), "section_time": section_time, "lane_label": lane_label, } def _build_bus_plan(bus_cfg, current_position, api_keys=None, transfer_context=None): from_address = _point_from_config(bus_cfg.get("from_address"), "address") from_stop = _point_from_config(bus_cfg.get("from_stop"), "bus_stop") to_stop = _point_from_config(bus_cfg.get("to_stop"), "bus_stop") to_address = _point_from_config(bus_cfg.get("to_address"), "address") if not (_valid_point(from_stop) and _valid_point(to_stop)): return None origin_point = _choose_origin_point(bus_cfg.get("origin_mode"), from_address, current_position) destination_point = to_address if _valid_point(to_address) else to_stop walk_to_board = fetch_walking_directions(origin_point, from_stop, api_keys=api_keys) walk_after_alight = fetch_walking_directions(to_stop, destination_point, api_keys=api_keys) if _valid_point(destination_point) and ( to_stop["lat"] != destination_point["lat"] or to_stop["lng"] != destination_point["lng"] ) else {"distance_m": 0, "duration_minutes": 0, "polyline": None} candidates = _filter_bus_trips(fetch_bus_arrivals(from_stop, api_keys=api_keys), bus_cfg.get("shared_routes") or []) if not candidates: raise ValueError("버스 실시간 도착정보를 찾을 수 없습니다.") candidate_trips, selected_index = _annotate_candidate_trips(candidates, walk_to_board["duration_minutes"] * 60) candidate_trips, selected_index = _retain_candidate_window(candidate_trips, selected_index) boardable = deepcopy(candidate_trips[selected_index]) segment_meta = _segment_points( origin_point, from_stop, to_stop, destination_point, "bus", route_name=boardable.get("route_name") or "", route_id=boardable.get("route_id") or "", api_keys=api_keys, ) ride_eta = segment_meta.get("section_time") or _estimate_ride_eta_minutes(from_stop, to_stop, "bus", api_keys=api_keys) now_dt = datetime.now() board_time = now_dt + timedelta(seconds=int(boardable["eta_seconds"])) arrival_time = board_time + timedelta(minutes=int(ride_eta + walk_after_alight["duration_minutes"])) route_label = boardable.get("route_name") or segment_meta.get("lane_label") or "버스" return { "transport_type": "bus", "snapshot_created_at": _now_iso(), "origin_point": origin_point, "board_point": from_stop, "alight_point": to_stop, "destination_point": destination_point, "walk_to_board_minutes": walk_to_board["duration_minutes"], "walk_to_board_distance_m": walk_to_board["distance_m"], "ride_eta_minutes": ride_eta, "walk_after_alight_minutes": walk_after_alight["duration_minutes"], "walk_after_alight_distance_m": walk_after_alight["distance_m"], "board_time": board_time.isoformat(), "arrival_time": arrival_time.isoformat(), "boardable_trip": boardable, "candidate_trips": candidate_trips, "selected_trip_index": selected_index, "segment_points": segment_meta.get("points") or [], "vehicle_marker": { "type": "bus", "label": route_label, "eta_text": boardable.get("eta_text") or "", "direction": boardable.get("direction") or "", }, "summary": { "route_label": route_label, "board_time_hhmm": _format_hhmm(board_time.isoformat()), "arrival_time_hhmm": _format_hhmm(arrival_time.isoformat()), }, "transfer_walk": { "enabled": bool(transfer_context), "from_label": (transfer_context or {}).get("source_label") or origin_point.get("label"), "from_card_label": (transfer_context or {}).get("preset_label") or "", "to_label": from_stop.get("label"), "duration_minutes": walk_to_board["duration_minutes"], "distance_m": walk_to_board["distance_m"], }, } def _build_subway_plan(subway_cfg, current_position, api_keys=None, transfer_context=None): from_address = _point_from_config(subway_cfg.get("from_address"), "address") from_station = _point_from_config(subway_cfg.get("from_station"), "subway_station") via_station = _point_from_config(subway_cfg.get("via_station"), "subway_station") if subway_cfg.get("via_station") else None to_station = _point_from_config(subway_cfg.get("to_station"), "subway_station") to_address = _point_from_config(subway_cfg.get("to_address"), "address") if not (_valid_point(from_station) and _valid_point(to_station)): return None origin_point = _choose_origin_point(subway_cfg.get("origin_mode"), from_address, current_position) destination_point = to_address if _valid_point(to_address) else to_station walk_to_board = fetch_walking_directions(origin_point, from_station, api_keys=api_keys) walk_after_alight = fetch_walking_directions(to_station, destination_point, api_keys=api_keys) if _valid_point(destination_point) and ( to_station["lat"] != destination_point["lat"] or to_station["lng"] != destination_point["lng"] ) else {"distance_m": 0, "duration_minutes": 0, "polyline": None} candidates = fetch_subway_arrivals(from_station.get("label"), via_station.get("label") if via_station else "", to_station.get("label"), api_keys=api_keys) if not candidates: raise ValueError("지하철 실시간 도착정보를 찾을 수 없습니다.") candidate_trips, selected_index = _annotate_candidate_trips(candidates, walk_to_board["duration_minutes"] * 60) candidate_trips, selected_index = _retain_candidate_window(candidate_trips, selected_index) boardable = deepcopy(candidate_trips[selected_index]) segment_meta = _build_subway_journey_meta( from_station, to_station, destination_point, via_point=via_station, api_keys=api_keys, ) if not segment_meta: segment_meta = _segment_points( origin_point, from_station, to_station, destination_point, "subway", via_point=via_station, route_name=boardable.get("route_name") or "", api_keys=api_keys, ) ride_eta = segment_meta.get("section_time") or _estimate_ride_eta_minutes(from_station, to_station, "subway", api_keys=api_keys) now_dt = datetime.now() board_time = now_dt + timedelta(seconds=int(boardable["eta_seconds"])) arrival_time = board_time + timedelta(minutes=int(ride_eta + walk_after_alight["duration_minutes"])) route_label = boardable.get("route_name") or segment_meta.get("lane_label") or "지하철" subway_line_key = _extract_subway_line_key( segment_meta.get("primary_line_key"), segment_meta.get("lane_label"), boardable.get("line_key"), ((from_station.get("meta") or {}).get("lineName")), ((from_station.get("meta") or {}).get("lineNames")), ((to_station.get("meta") or {}).get("lineName")), ((to_station.get("meta") or {}).get("lineNames")), route_label, ) if subway_line_key: boardable["line_key"] = subway_line_key journey_segments = _attach_subway_segment_candidates(segment_meta.get("journey_segments") or [], api_keys=api_keys) return { "transport_type": "subway", "snapshot_created_at": _now_iso(), "origin_point": origin_point, "board_point": from_station, "alight_point": to_station, "destination_point": destination_point, "walk_to_board_minutes": walk_to_board["duration_minutes"], "walk_to_board_distance_m": walk_to_board["distance_m"], "ride_eta_minutes": ride_eta, "walk_after_alight_minutes": walk_after_alight["duration_minutes"], "walk_after_alight_distance_m": walk_after_alight["distance_m"], "board_time": board_time.isoformat(), "arrival_time": arrival_time.isoformat(), "boardable_trip": boardable, "candidate_trips": candidate_trips, "selected_trip_index": selected_index, "segment_points": segment_meta.get("points") or [], "journey_segments": journey_segments, "vehicle_marker": { "type": "subway", "label": route_label, "eta_text": boardable.get("eta_text") or "", "destination": boardable.get("destination") or "", "line_key": subway_line_key, }, "summary": { "route_label": route_label, "board_time_hhmm": _format_hhmm(board_time.isoformat()), "arrival_time_hhmm": _format_hhmm(arrival_time.isoformat()), "subway_line_key": subway_line_key, }, "transfer_walk": { "enabled": bool(transfer_context), "from_label": (transfer_context or {}).get("source_label") or origin_point.get("label"), "from_card_label": (transfer_context or {}).get("preset_label") or "", "to_label": from_station.get("label"), "duration_minutes": walk_to_board["duration_minutes"], "distance_m": walk_to_board["distance_m"], }, } def _get_preset(root, preset_id): for preset in root.get("presets") or []: if preset.get("id") == preset_id: return preset return (root.get("presets") or [default_preset(1)])[0] def _get_preset_index(root, preset_id): for idx, preset in enumerate(root.get("presets") or []): if preset.get("id") == preset_id: return idx return 0 def _get_previous_card_context(root, preset_id, mode_name): idx = _get_preset_index(root, preset_id) if idx <= 0: return None presets = root.get("presets") or [] prev_preset = presets[idx - 1] if idx - 1 < len(presets) else None if not prev_preset: return None prev_bucket = ((prev_preset.get("modes") or {}).get(mode_name) or {}) prev_plan = prev_bucket.get("last_plan") or {} prev_runtime = prev_bucket.get("runtime_state") or {} source_point = None for candidate in ( prev_runtime.get("last_known_position"), prev_plan.get("destination_point"), prev_plan.get("alight_point"), prev_plan.get("board_point"), ): if _valid_point(candidate): source_point = deepcopy(candidate) break if not source_point: return None source_label = source_point.get("label") or _normalize_label(prev_preset.get("label"), "이전 카드") return { "preset_id": prev_preset.get("id"), "preset_label": _normalize_label(prev_preset.get("label"), f"카드 {idx}"), "source_point": source_point, "source_label": source_label, } def _get_mode_bucket(root, preset_id, mode_name): preset = _get_preset(root, preset_id) return preset["modes"][_normalize_mode_name(mode_name)] def resolve_active_selection(root, preset_id=None, mode_name=None): normalized = normalize_commute_state(root) resolved_preset = _get_preset(normalized, preset_id or normalized.get("activePresetId")) resolved_mode = _normalize_mode_name(mode_name or normalized.get("activeMode")) return normalized, resolved_preset["id"], resolved_mode def get_mode_snapshot(root, preset_id=None, mode_name=None): normalized, resolved_preset, resolved_mode = resolve_active_selection(root, preset_id, mode_name) bucket = _get_mode_bucket(normalized, resolved_preset, resolved_mode) return { "presetId": resolved_preset, "mode": resolved_mode, "config": deepcopy(bucket.get("config") or {}), "last_plan": deepcopy(bucket.get("last_plan")), "runtime_state": deepcopy(bucket.get("runtime_state") or default_runtime_state()), } def validate_current_position(current_position): if not _valid_point(current_position): raise ValueError("현재 위치 좌표(lat,lng)가 필요합니다.") return { "lat": _parse_float(current_position.get("lat")), "lng": _parse_float(current_position.get("lng")), } def build_plan(root, preset_id, mode_name, current_position): root, preset_id, mode_name = resolve_active_selection(root, preset_id, mode_name) current_position = validate_current_position(current_position) bucket = _get_mode_bucket(root, preset_id, mode_name) config = deepcopy(bucket["config"]) transport_type = _normalize_transport_type(config.get("transport_type"), "bus") preset_index = _get_preset_index(root, preset_id) transfer_context = _get_previous_card_context(root, preset_id, mode_name) if preset_index > 0 else None if preset_index > 0: config["bus"]["origin_mode"] = "gps" config["subway"]["origin_mode"] = "gps" api_keys = get_api_keys(root) candidates = [] errors = [] if transport_type == "bus": try: bus_plan = _build_bus_plan(config.get("bus") or {}, current_position, api_keys=api_keys, transfer_context=transfer_context) if bus_plan: candidates.append(bus_plan) except Exception as exc: errors.append(str(exc)) else: try: subway_plan = _build_subway_plan(config.get("subway") or {}, current_position, api_keys=api_keys, transfer_context=transfer_context) if subway_plan: candidates.append(subway_plan) except Exception as exc: errors.append(str(exc)) if not candidates: raise ValueError(errors[0] if errors else "출발 가능한 계획을 만들지 못했습니다.") candidates.sort(key=lambda item: item.get("arrival_time") or "") selected = candidates[0] runtime_state = default_runtime_state() runtime_state["journey_phase"] = "started" runtime_state["start_triggered_at"] = _now_iso() bucket["last_plan"] = selected bucket["runtime_state"] = runtime_state root["activePresetId"] = preset_id root["activeMode"] = mode_name return selected, root def select_candidate_trip(root, preset_id, mode_name, trip_index, force_select=False): root, preset_id, mode_name = resolve_active_selection(root, preset_id, mode_name) bucket = _get_mode_bucket(root, preset_id, mode_name) last_plan = deepcopy(bucket.get("last_plan")) if not last_plan: raise ValueError("먼저 출발을 눌러 실행계획을 생성해야 합니다.") candidates = last_plan.get("candidate_trips") or [] if not candidates: raise ValueError("선택 가능한 차량 후보가 없습니다.") trip_index = int(trip_index) if trip_index < 0 or trip_index >= len(candidates): raise ValueError("잘못된 차량 선택입니다.") selected_trip = candidates[trip_index] if not force_select and not selected_trip.get("selectable"): raise ValueError("현재 기준으로 탑승할 수 없는 차량입니다.") last_plan = _apply_selected_trip(last_plan, trip_index) bucket["last_plan"] = last_plan return last_plan, root def _project_on_segment(current_position, start_point, end_point): lat_scale = 110540.0 lng_scale = max(1.0, 111320.0 * math.cos(math.radians(current_position["lat"]))) sx = (start_point["lng"] - current_position["lng"]) * lng_scale sy = (start_point["lat"] - current_position["lat"]) * lat_scale ex = (end_point["lng"] - current_position["lng"]) * lng_scale ey = (end_point["lat"] - current_position["lat"]) * lat_scale dx = ex - sx dy = ey - sy denom = dx * dx + dy * dy if denom <= 0: return None t = max(0.0, min(1.0, -((sx * dx) + (sy * dy)) / denom)) proj_x = sx + dx * t proj_y = sy + dy * t return { "ratio": t, "distance_m": round(math.hypot(proj_x, proj_y), 1), "lat": current_position["lat"] + (proj_y / lat_scale), "lng": current_position["lng"] + (proj_x / lng_scale), } def _segment_kind(start_point, end_point): transit_types = {"bus_stop", "subway_station"} if start_point.get("type") in transit_types and end_point.get("type") in transit_types: return "차량구간" return "도보구간" def _project_to_segments(segment_points, current_position): if not (_valid_point(current_position) and isinstance(segment_points, list) and segment_points): return None candidates = [] for idx, point in enumerate(segment_points): if not _valid_point(point): continue distance = _haversine_m(current_position["lat"], current_position["lng"], point["lat"], point["lng"]) candidates.append( { "index": idx, "point_type": _point_kind(point), "label": point.get("label") or f"지점 {idx + 1}", "distance_m": round(distance, 1), "lat": point["lat"], "lng": point["lng"], "source": "point", } ) for idx in range(len(segment_points) - 1): start_point = segment_points[idx] end_point = segment_points[idx + 1] if not (_valid_point(start_point) and _valid_point(end_point)): continue projection = _project_on_segment(current_position, start_point, end_point) if not projection: continue candidates.append( { "index": idx + (1 if projection["ratio"] >= 0.5 else 0), "point_type": _segment_kind(start_point, end_point), "label": f"{start_point.get('label') or '시작'} → {end_point.get('label') or '도착'}", "distance_m": projection["distance_m"], "lat": projection["lat"], "lng": projection["lng"], "segment_start_label": start_point.get("label") or "", "segment_end_label": end_point.get("label") or "", "source": "segment", } ) if not candidates: return None point_candidates = [item for item in candidates if item["source"] == "point" and item["distance_m"] <= 120] if point_candidates: return min(point_candidates, key=lambda item: item["distance_m"]) return min(candidates, key=lambda item: item["distance_m"]) def _subway_board_segment(last_plan): journey_segments = last_plan.get("journey_segments") or [] if not journey_segments: return None preferred_key = _extract_subway_line_key( (last_plan.get("boardable_trip") or {}).get("line_key"), (last_plan.get("summary") or {}).get("subway_line_key"), (last_plan.get("vehicle_marker") or {}).get("line_key"), ) for segment in journey_segments: if segment.get("kind") == "subway" and preferred_key and _extract_subway_line_key(segment.get("line_key"), segment.get("route_label")) == preferred_key: return segment for segment in journey_segments: if segment.get("kind") == "subway": return segment return None def _segment_station_index(segment, label): normalized = _normalized_name(label) if not normalized: return None for idx, point in enumerate(segment.get("points") or []): if _normalized_name(point.get("label")) == normalized: return idx return None def _subway_train_position_ref(last_plan, live_positions): selected_trip = last_plan.get("boardable_trip") or {} selected_train = str(selected_trip.get("train_no") or "") if not selected_train: return None, None, False board_segment = _subway_board_segment(last_plan) if not board_segment: return None, None, False board_index = _segment_station_index(board_segment, board_segment.get("board_label")) matched_position = next((item for item in live_positions if str(item.get("train_no") or "") == selected_train), None) if not matched_position: return None, None, False station_index = _segment_station_index(board_segment, matched_position.get("station_label")) if station_index is None: return matched_position, None, False point = (board_segment.get("points") or [])[station_index] point_meta = point.get("meta") or {} position_ref = { "index": point_meta.get("global_index"), "point_type": _point_kind(point), "label": point.get("label") or matched_position.get("station_label") or "", "distance_m": 0.0, "lat": point.get("lat"), "lng": point.get("lng"), "source": "train_position", } has_passed_board = board_index is not None and station_index > board_index return matched_position, position_ref, has_passed_board def refresh_live(root, preset_id, mode_name, current_position, confirm_boarding=False): root, preset_id, mode_name = resolve_active_selection(root, preset_id, mode_name) current_position = validate_current_position(current_position) bucket = _get_mode_bucket(root, preset_id, mode_name) last_plan = bucket.get("last_plan") api_keys = get_api_keys(root) if not last_plan: raise ValueError("먼저 출발을 눌러 실행계획을 생성해야 합니다.") runtime = deepcopy(bucket.get("runtime_state") or default_runtime_state()) runtime["last_known_position"] = _point("현재 위치", current_position.get("lat"), current_position.get("lng"), "gps") runtime["last_known_position_updated_at"] = _now_iso() runtime["nearest_segment_ref"] = _project_to_segments(last_plan.get("segment_points") or [], current_position) if last_plan.get("transport_type") == "bus": live_candidates = fetch_bus_arrivals(last_plan.get("board_point") or {}, api_keys=api_keys) live_candidates = _filter_bus_trips(live_candidates, [{"busRouteNm": (last_plan.get("summary") or {}).get("route_label"), "busRouteId": (last_plan.get("boardable_trip") or {}).get("route_id")}]) live_positions = [] else: board_point = last_plan.get("board_point") or {} alight_point = last_plan.get("alight_point") or {} live_candidates = fetch_subway_arrivals( board_point.get("label") or "", "", alight_point.get("label") or (last_plan.get("vehicle_marker") or {}).get("destination") or "", api_keys=api_keys, ) live_positions = fetch_subway_positions( (last_plan.get("summary") or {}).get("subway_line_key") or (last_plan.get("vehicle_marker") or {}).get("line_key") or (last_plan.get("boardable_trip") or {}).get("line_key"), api_keys=api_keys, ) if last_plan.get("journey_segments"): last_plan["journey_segments"] = _attach_subway_segment_candidates(last_plan.get("journey_segments") or [], api_keys=api_keys) bucket["last_plan"] = last_plan runtime["live_candidate_trips"] = live_candidates[:8] runtime["live_boardable_trip"] = deepcopy(last_plan.get("boardable_trip")) if last_plan.get("boardable_trip") else (live_candidates[0] if live_candidates else None) runtime["live_train_position"] = None runtime["live_vehicle_marker"] = { "type": last_plan.get("transport_type"), "label": (runtime["live_boardable_trip"] or {}).get("route_name") or last_plan.get("summary", {}).get("route_label"), "eta_text": (runtime["live_boardable_trip"] or {}).get("eta_text") or "", "direction": (runtime["live_boardable_trip"] or {}).get("direction") or "", "destination": (runtime["live_boardable_trip"] or {}).get("destination") or "", "line_key": _extract_subway_line_key( (runtime["live_boardable_trip"] or {}).get("line_key"), (last_plan.get("summary") or {}).get("subway_line_key"), (runtime["live_boardable_trip"] or {}).get("route_name"), (last_plan.get("vehicle_marker") or {}).get("line_key"), ), } if live_candidates or last_plan.get("summary") else None runtime["live_position_marker"] = runtime["nearest_segment_ref"] if confirm_boarding: runtime["journey_phase"] = "boarded" runtime["boarding_confirmed_at"] = _now_iso() elif last_plan.get("transport_type") == "subway": matched_train, position_ref, has_passed_board = _subway_train_position_ref(last_plan, live_positions) if matched_train: runtime["live_train_position"] = matched_train if runtime.get("live_vehicle_marker") is None: runtime["live_vehicle_marker"] = {"type": "subway"} runtime["live_vehicle_marker"]["station_label"] = matched_train.get("station_label") or "" runtime["live_vehicle_marker"]["train_no"] = matched_train.get("train_no") or "" runtime["live_vehicle_marker"]["status_code"] = matched_train.get("status_code") or "" if position_ref: runtime["live_position_marker"] = position_ref if runtime.get("journey_phase") == "started" and has_passed_board: runtime["journey_phase"] = "boarded" runtime["boarding_confirmed_at"] = _now_iso() elif runtime.get("journey_phase") == "idle": runtime["journey_phase"] = "started" bucket["runtime_state"] = runtime return runtime, root def reset_journey_state(root, preset_id, mode_name): root, preset_id, mode_name = resolve_active_selection(root, preset_id, mode_name) bucket = _get_mode_bucket(root, preset_id, mode_name) runtime = default_runtime_state() runtime["journey_phase"] = "started" if bucket.get("last_plan") else "idle" runtime["start_triggered_at"] = (bucket.get("runtime_state") or {}).get("start_triggered_at") bucket["runtime_state"] = runtime return runtime, root