Add departure card mapping

custom_components/hasl3/rrapi/client.py

@@ -1,7 +1,13 @@
 import logging
 from datetime import datetime, tzinfo, UTC
 from typing import Any, cast
-from .model import StopLookupResponse, LocationSearchType
+from .model import (
+    ListOfArrivals,
+    ListOfDepartures,
+    StopLookupResponse,
+    LocationSearchType,
+    TransportCategory,
+)
 
 import aiohttp
 import isodate
@@ -157,87 +163,107 @@ async def find_trip(self, origin: str, destination: str) -> list[dict[str, Any]]
 
         return result
 
+    transportMap = {
+        "BLT": "BUS",
+        "BXB": "BUS",
+        "BAX": "BUS",
+        "BRE": "BUS",
+        "BBL": "BUS",
+        "ULT": "METRO",
+        "JAX": "TRAIN",
+        "JEX": "TRAIN",
+        "JIC": "TRAIN",
+        "JLT": "TRAIN",
+        "JPT": "TRAIN",
+        "JST": "TRAIN",
+        "JRE": "TRAIN",
+        "SLT": "TRAM",
+        "FLT": "FERRY",
+        "FUT": "FERRY",
+    }
+
     async def get_departures(
         self, location_id: str, now: datetime
     ) -> list[dict[str, Any]]:
         """Get departures from a specific location."""
-        url = f"{self._base_url}/departureBoard?format=json&id={location_id}&accessId={self._api_key}"
-        data = await self._get_json(url)
+        data = await self._get_json(
+            url=f"{self._base_url}/departureBoard",
+            params={
+                "format": "json",
+                "id": location_id,
+                "accessId": self._api_key,
+            },
+        )
 
+        data = cast(ListOfDepartures, data)
         # transform data
         departures = []
         for departure in data["Departure"]:
-            date, time = departure["date"], departure["time"]
-
-            if "rtDate" in departure and "rtTime" in departure:
-                diff_date, diff_time = departure["rtDate"], departure["rtTime"]
-            else:
-                diff_date, diff_time = date, time
-
-            adjustedDateTime = now.replace(tzinfo=None)
-            diff = (
-                datetime.strptime(f"{diff_date} {diff_time}", "%Y-%m-%d %H:%M:%S")
-                - adjustedDateTime
-            )
-            diff = round(diff.total_seconds() / 60)
-
-            expected = datetime.strptime(
-                f"{diff_date} {diff_time}", "%Y-%m-%d %H:%M:%S"
-            ).replace(tzinfo=self._timezone)
+            (date, time) = departure["date"], departure["time"]
+            scheduledTime = datetime.strptime(f"{date} {time}", "%Y-%m-%d %H:%M:%S").replace(tzinfo=self._timezone)
+            (adjustedDate, adjustedTime) =  departure.get("rtDate", date), departure.get("rtTime", time)
+            expectedTime = datetime.strptime(f"{adjustedDate} {adjustedTime}", "%Y-%m-%d %H:%M:%S").replace(tzinfo=self._timezone)
 
             departures.append(
                 {
-                    "line": departure["ProductAtStop"]["displayNumber"],
-                    "direction": departure["directionFlag"],
-                    "departure": datetime.strptime(
-                        f"{date} {time}", "%Y-%m-%d %H:%M:%S"
-                    ),
                     "destination": departure["direction"],
-                    "time": diff,
-                    "operator": departure["ProductAtStop"]["operator"],
-                    "expected": expected,
-                    "type": departure["ProductAtStop"]["catOut"],
+                    "direction": "",
+                    "direction_code": departure.get("directionFlag", 0),
+                    "state": "EXPECTED",
+                    "display": departure["name"],
+                    "stop_point": {"name": departure["stop"], "designation": ""},
+                    "line": {
+                        "id": departure["ProductAtStop"].get("lineId", 0),
+                        "designation": departure["ProductAtStop"].get("displayNumber"),
+                        "transport_mode": self.transportMap.get(
+                            departure["ProductAtStop"].get("catOut", TransportCategory.BLT)
+                        ),
+                        "group_of_lines": "",
+                    },
+                    "scheduled": scheduledTime.isoformat(),
+                    "expected": expectedTime.isoformat(),
                 }
             )
 
         return departures
 
     async def get_arrivals(self, location_id: str, now: datetime):
-        url = f"{self._base_url}/arrivalBoard?format=json&id={location_id}&accessId={self._api_key}"
-        data = await self._get_json(url)
+        data = await self._get_json(
+            url=f"{self._base_url}/arrivalBoard",
+            params={
+                "format": "json",
+                "id": location_id,
+                "accessId": self._api_key,
+            },
+        )
 
+        data = cast(ListOfArrivals, data)
         # transform data
         arrivals = []
         for arrival in data["Arrival"]:
-            date, time = arrival["date"], arrival["time"]
-
-            if "rtDate" in arrival and "rtTime" in arrival:
-                diff_date, diff_time = arrival["rtDate"], arrival["rtTime"]
-            else:
-                diff_date, diff_time = date, time
-
-            adjustedDateTime = now.replace(tzinfo=None)
-            diff = (
-                datetime.strptime(f"{diff_date} {diff_time}", "%Y-%m-%d %H:%M:%S")
-                - adjustedDateTime
-            )
-            diff = round(diff.total_seconds() / 60)
-
-            expected = datetime.strptime(
-                f"{diff_date} {diff_time}", "%Y-%m-%d %H:%M:%S"
-            ).replace(tzinfo=self._timezone)
+            (date, time) = arrival["date"], arrival["time"]
+            scheduledTime = datetime.strptime(f"{date} {time}", "%Y-%m-%d %H:%M").replace(tzinfo=self._timezone)
+            (adjustedDate, adjustedTime) = arrival.get("rtDate", time), arrival.get("rtTime", time)
+            expectedTime = datetime.strptime(f"{adjustedDate} {adjustedTime}", "%Y-%m-%d %H:%M").replace(tzinfo=self._timezone)
 
             arrivals.append(
                 {
-                    "line": arrival["ProductAtStop"]["displayNumber"],
-                    "arrival": datetime.strptime(
-                        f"{date} {time}", "%Y-%m-%d %H:%M:%S"
-                    ),
-                    "origin": arrival["origin"],
-                    "time": diff,
-                    "operator": arrival["ProductAtStop"]["operator"],
-                    "expected": expected,
-                    "type": arrival["ProductAtStop"]["catOut"],
+                    "destination": arrival["direction"],
+                    "direction": "",
+                    "direction_code": arrival.get("directionFlag", 0),
+                    "state": "EXPECTED",
+                    "display": arrival["name"],
+                    "stop_point": {"name": arrival["stop"], "designation": ""},
+                    "line": {
+                        "id": arrival["ProductAtStop"].get("lineId", 0),
+                        "designation": arrival["ProductAtStop"].get("displayNumber"),
+                        "transport_mode": self.transportMap.get(
+                            arrival["ProductAtStop"].get("catOut", TransportCategory.ULT)
+                        ),
+                        "group_of_lines": "",
+                    },
+                    "scheduled": scheduledTime.isoformat(),
+                    "expected": expectedTime.isoformat(),
                 }
             )
 

custom_components/hasl3/rrapi/model.py

@@ -1,5 +1,5 @@
 from typing import TypedDict, NotRequired
-from enum import Enum
+from enum import Enum, IntEnum
 
 
 class LocationSearchType(Enum):
@@ -12,6 +12,56 @@ class LocationSearchType(Enum):
     ADDRESS_OR_POI = "AP"  # Search for addresses and POIs
 
 
+class BoardLanguage(Enum):
+    SV = "sv"  # Swedish
+    EN = "en"  # English
+    DA = "da"  # Danish
+    NO = "no"  # Norwegian
+    DE = "de"  # German
+    FR = "fr"  # French
+    IT = "it"  # Italian
+    NL = "nl"  # Dutch
+    TR = "tr"  # Turkish
+    PL = "pl"  # Polish
+    ES = "es"  # Spanish
+    HU = "hu"  # Hungarian
+
+
+class LocationType(Enum):
+    ST = "ST"  # Stop or station
+    ADR = "ADR"  # Address
+    POI = "POI"  # Point of interest
+    CRD = "CRD"  # Coordinate
+    MCP = "MCP"  # Mode change point
+    HL = "HL"  # Hailing point
+
+
+class TransportCategory(Enum):
+    BLT = "BLT"  # Regional bus (lanstrafik), e.g. SL, UL, Skanetrafiken
+    BXB = "BXB"  # Express bus
+    BAX = "BAX"  # Airport express bus
+    BRE = "BRE"  # Regional bus other than lanstrafik
+    BBL = "BBL"  # Train replacement bus
+    ULT = "ULT"  # Metro
+    JAX = "JAX"  # Airport express train
+    JEX = "JEX"  # Express train
+    JIC = "JIC"  # InterCity train
+    JLT = "JLT"  # Local train
+    JPT = "JPT"  # PagaTag
+    JST = "JST"  # High-speed train
+    JRE = "JRE"  # Regional train
+    SLT = "SLT"  # Tram
+    FLT = "FLT"  # Local ferry
+    FUT = "FUT"  # International ferry
+
+
+class JourneyStatus(Enum):
+    PLANNED = "P"  # Planned
+    REPLACEMENT = "R"  # Replacement
+    ADDITIONAL = "A"  # Additional
+    SPECIAL = "S"  # Special
+
+
 class StopLookupEntry(TypedDict):
     id: str
     extId: NotRequired[str]
@@ -28,3 +78,95 @@ class StopLocationWrapper(TypedDict):
 
 class StopLookupResponse(TypedDict):
     stopLocationOrCoordLocation: list[StopLocationWrapper]
+
+
+class JourneyDetailRef(TypedDict):
+    ref: str
+
+
+class Stop(TypedDict):
+    name: str
+    id: str
+    extId: str
+    lon: float
+    lat: float
+    routeIdx: int
+    arrTime: NotRequired[str]
+    arrDate: NotRequired[str]
+    depTime: NotRequired[str]
+    depDate: NotRequired[str]
+
+
+class Product(TypedDict):
+    name: str
+    cls: str
+    internalName: NotRequired[str]
+    num: NotRequired[str]
+    displayNumber: NotRequired[str]
+    line: NotRequired[str]
+    lineId: NotRequired[str]
+    catCode: NotRequired[str]
+    catOut: NotRequired[TransportCategory]
+    catIn: NotRequired[TransportCategory]
+    catOutS: NotRequired[TransportCategory]
+    catOutL: NotRequired[str]
+    operatorCode: NotRequired[str]
+    operator: NotRequired[str]
+
+
+class DepartureStops(TypedDict):
+    stop: list[Stop]
+
+
+class DepartureBoardEntry(TypedDict):
+    Stops: DepartureStops
+    ProductAtStop: Product
+    Product: list[Product]
+    name: str
+    type: LocationType
+    stop: str
+    stopid: str
+    stopExtId: str
+    time: str # Scheduled departure/arrival time, formatted as HH:MM:SS
+    date: str # Scheduled departure/arrival date, formatted as YYYY-MM-DD
+    rtTime: NotRequired[str] # Realtime departure/arrival time, formatted as HH:MM:SS
+    rtDate: NotRequired[str] # Realtime departure/arrival date, formatted as YYYY-MM-DD
+    direction: str
+    transportNumber: str
+    transportCategory: TransportCategory
+    reachable: NotRequired[bool]
+    JourneyStatus: NotRequired[JourneyStatus]
+    JourneyDetailRef: NotRequired[JourneyDetailRef]
+
+
+class ArrivalBoardEntry(TypedDict):
+    Stops: list[Stop]
+    ProductAtStop: Product
+    Product: list[Product]
+    name: str
+    type: LocationType
+    stop: str
+    stopid: str
+    stopExtId: str
+    time: str
+    date: str
+    direction: str
+    transportNumber: str
+    transportCategory: TransportCategory
+    JourneyStatus: NotRequired[JourneyStatus]
+    JourneyDetailRef: NotRequired[JourneyDetailRef]
+
+
+class ListOfDepartures(TypedDict):
+    Departure: list[DepartureBoardEntry]
+    requestId: NotRequired[str]
+
+
+class ListOfArrivals(TypedDict):
+    Arrival: list[ArrivalBoardEntry]
+    requestId: NotRequired[str]
+
+
+class ResrobotException(TypedDict):
+    errorCode: str
+    errorText: str