Create costs.py

services/costs.py

@@ -0,0 +1,106 @@
+from __future__ import annotations
+from typing import Optional, Union, Dict, Any
+import re
+
+Number = Union[int, float]
+
+def _parse_fare_hint(fare_hint: Any) -> Optional[float]:
+    """
+    Navitia等からの運賃ヒントをなるべく数値(JPY)にして返す。
+    - dict: {"total": {"value": 420, "currency": "JPY"}} などを優先
+            {"amount": 420, "currency": "JPY"} にも対応
+    - str : "¥420", "JPY 420", "420 JPY" 等から数値を抽出
+    - それ以外: None
+    """
+    # dictパターン
+    if isinstance(fare_hint, dict):
+        # Navitia: fare.total.value / fare.total.currency
+        total = fare_hint.get("total") if isinstance(fare_hint.get("total"), dict) else None
+        if total and isinstance(total.get("value"), (int, float)):
+            cur = str(total.get("currency") or "JPY").upper()
+            if cur in ("JPY", "YEN", "¥", "JPY "):
+                return float(total["value"])
+            else:
+                # 通貨が別の時はここでは換算せず None（過小評価を避ける）
+                return None
+        # amount/currency 形式
+        amt = fare_hint.get("amount")
+        cur = str(fare_hint.get("currency") or "JPY").upper()
+        if isinstance(amt, (int, float)) and cur in ("JPY", "YEN", "¥"):
+            return float(amt)
+        # 別のキーでも value を持っていればそれを採用
+        for k in ("value", "price", "fare"):
+            v = fare_hint.get(k)
+            if isinstance(v, (int, float)):
+                return float(v)
+
+    # 文字列パターン
+    if isinstance(fare_hint, str):
+        m = re.search(r"([\d,]+(?:\.\d+)?)", fare_hint)
+        if m:
+            try:
+                return float(m.group(1).replace(",", ""))
+            except Exception:
+                pass
+
+    return None
+
+
+def estimate_leg_cost(
+    mode: str,
+    distance_km: Number,
+    duration_min: Number,
+    fare_hint: Optional[Union[dict, str, Number]] = None,
+) -> float:
+    """
+    区間の推定交通費（JPY）を返す簡易関数。
+    - mode: "Walk" | "Transit" | "Drive/Taxi"
+    - fare_hint: Navitia等の運賃情報がある場合はそれを優先
+    - 返り値: 四捨五入して10円単位
+    """
+    # ヒントがあれば最優先（数値化できればそのまま採用）
+    if isinstance(fare_hint, (int, float)):
+        return float(round(fare_hint / 10.0) * 10)
+    hint = _parse_fare_hint(fare_hint)
+    if hint is not None:
+        return float(round(hint / 10.0) * 10)
+
+    d = max(0.0, float(distance_km))
+    t = max(0.0, float(duration_min))
+    mode = (mode or "Walk").strip()
+
+    # 徒歩は無料
+    if mode.lower().startswith("walk"):
+        return 0.0
+
+    # 公共交通：ざっくり距離比例（東京近郊の初乗り含むイメージ）
+    if mode.lower().startswith("transit"):
+        base = 180.0            # 初乗りイメージ
+        per_km = 22.0           # 距離係数
+        est = base + per_km * d
+        # 長距離で上限（空港連絡等は別だが、概算としてのキャップ）
+        est = min(est, 1200.0)
+        return float(round(est / 10.0) * 10)
+
+    # タクシー/自動車：ざっくり距離＋低速時の時間加算
+    # 東京の実勢に近い概算：初乗り500円 + 420円/km + 渋滞待機 45円/分（低速時のみ）
+    if mode.lower().startswith("drive"):
+        flag_fall = 500.0
+        per_km = 420.0
+        est = flag_fall + per_km * d
+
+        # 低速（12km/h未満）なら待機料金を加算（概算）
+        speed_kmh = (d / (t / 60.0)) if t > 0 else 999.0
+        if speed_kmh < 12.0:
+            wait_fee_per_min = 45.0
+            # 12km/h で同距離を走るのに要する時間との差を「渋滞/待機」とみなす
+            ideal_min = (d / 12.0) * 60.0
+            extra_min = max(0.0, t - ideal_min)
+            est += wait_fee_per_min * extra_min
+
+        # 短距離の過小評価を避ける最低料金
+        est = max(est, 680.0)
+        return float(round(est / 10.0) * 10)
+
+    # 未知のモードは0円にしておく（上流で弾かれる想定）
+    return 0.0