Add support for simple trip query with optional destination ID

README.md

@@ -43,7 +43,9 @@ sensor:
         stop_id: 33000112 # actual Stop ID for the API
       - name: "Altmarkt" # you can add more that one stop to track
         stop_id: 33000004
-
+      - name: "Hauptbahnhof --> Nürnberger Platz" # Only show Departures in that direction
+        stop_id: 33000028
+        destination_id: 33000132 # you can add a destination ID (using the same approach as for the stop ID) to show only usable departures for that direction/trip 
         # Optional parameter with value in minutes that hide transport closer than N minutes
         # walking_time: 5
 ```

custom_components/dresden_transport/const.py

@@ -4,6 +4,7 @@
 DOMAIN = "dresden_transport"
 SCAN_INTERVAL = timedelta(seconds=90)
 API_ENDPOINT = "https://webapi.vvo-online.de/dm"
+API_ENDPOINT_TRIP = "https://webapi.vvo-online.de/tr/trips?format=json"
 API_MAX_RESULTS = 10
 
 DEFAULT_ICON = "mdi:clock"
@@ -13,6 +14,7 @@
 CONF_DEPARTURES_STOP_ID = "stop_id"
 CONF_DEPARTURES_WALKING_TIME = "walking_time"
 CONF_DEPARTURES_DIRECTION = "direction"
+CONF_DEPARTURES_DESTINATION_ID = "destination_id"
 CONF_TYPE_SUBURBAN = "suburban"
 CONF_TYPE_SUBWAY = "subway"
 CONF_TYPE_TRAM = "Tram"

custom_components/dresden_transport/departure.py

@@ -38,7 +38,34 @@ def from_dict(cls, source):
             bg_color=source.get("line", {}).get("color", {}).get("bg"),
             fallback_color=line_visuals.get("color"),
         )
-
+    @classmethod
+    def from_dict_dir(cls, source):
+        line_type = source.get("MotChain", [])[0].get("Type")
+        line_visuals = TRANSPORT_TYPE_VISUALS.get(line_type) or {}
+        time_str = source.get("PartialRoutes",[])[0].get("RegularStops",[])[0].get("DepartureTime")
+        line_name = source.get("MotChain", [])[0].get("Name")
+        line_type = source.get("MotChain", [])[0].get("Type")
+        direction = source.get("MotChain", [])[0].get("Direction")
+        platform = source.get("PartialRoutes", [])[0].get("RegularStops", [])[0].get("Platform", {}).get("Name")
+        bg_color = source.get("line", {}).get("color", {}).get("bg")
+        fallback_color = line_visuals.get("color")
+
+        res = re.search(r'\d+', time_str) 
+        time = int(int(res.group()) / 1000)
+        gap = round((datetime.fromtimestamp(time) - datetime.now()).total_seconds() / 60)
+        return cls(
+            line_name=line_name,
+            line_type=line_type,
+            gap=gap,
+            timestamp=time,
+            time=datetime.fromtimestamp(time).strftime("%H:%M"),
+            direction=direction,
+            platform=platform,
+            icon=line_visuals.get("icon") or DEFAULT_ICON,
+            bg_color=bg_color,
+            fallback_color=fallback_color,
+        )
+
     def to_dict(self):
         return {
             "line_name": self.line_name,

custom_components/dresden_transport/sensor.py

@@ -17,10 +17,12 @@
     DOMAIN,  # noqa
     SCAN_INTERVAL,  # noqa
     API_ENDPOINT,
+    API_ENDPOINT_TRIP,
     API_MAX_RESULTS,
     CONF_DEPARTURES,
     CONF_DEPARTURES_DIRECTION,
     CONF_DEPARTURES_STOP_ID,
+    CONF_DEPARTURES_DESTINATION_ID,
     CONF_DEPARTURES_WALKING_TIME,
     CONF_TYPE_BUS,
     CONF_TYPE_EXPRESS,
@@ -53,6 +55,7 @@
                 vol.Required(CONF_DEPARTURES_NAME): str,
                 vol.Required(CONF_DEPARTURES_STOP_ID): int,
                 vol.Optional(CONF_DEPARTURES_DIRECTION): str,
+                vol.Optional(CONF_DEPARTURES_DESTINATION_ID): int,
                 vol.Optional(CONF_DEPARTURES_WALKING_TIME, default=1): int,
                 **TRANSPORT_TYPES_SCHEMA,
             }
@@ -81,6 +84,7 @@ def __init__(self, hass: HomeAssistant, config: dict) -> None:
         self.stop_id: int = config[CONF_DEPARTURES_STOP_ID]
         self.sensor_name: str | None = config.get(CONF_DEPARTURES_NAME)
         self.direction: str | None = config.get(CONF_DEPARTURES_DIRECTION)
+        self.destination_id: int | None = config.get(CONF_DEPARTURES_DESTINATION_ID)
         self.walking_time: int = config.get(CONF_DEPARTURES_WALKING_TIME) or 1
         # we add +1 minute anyway to delete the "just gone" transport
 
@@ -116,17 +120,59 @@ def update(self):
         self.departures = self.fetch_departures()
 
     def fetch_departures(self) -> Optional[list[Departure]]:
+        if self.destination_id == None:
+            try:
+                response = requests.get(
+                    url=f"{API_ENDPOINT}",
+                    params={
+                        "time": (
+                            datetime.now() + timedelta(minutes=self.walking_time)
+                        ).isoformat(),
+                        "format": "json",
+                        "limit": API_MAX_RESULTS,
+                        "stopID": self.stop_id,
+                        "isarrival": False,
+                        "shorttermchanges": True,
+                        "mentzonly": False,
+                    },
+                    timeout=30,
+                )
+                response.raise_for_status()
+            except requests.exceptions.HTTPError as ex:
+                _LOGGER.warning(f"API error: {ex}")
+                return []
+            except requests.exceptions.Timeout as ex:
+                _LOGGER.warning(f"API timeout: {ex}")
+                return []
+
+            _LOGGER.debug(f"OK: departures for {self.stop_id}: {response.text}")
+
+            # parse JSON response
+            try:
+                departures = response.json().get('Departures')
+            except requests.exceptions.InvalidJSONError as ex:
+                _LOGGER.error(f"API invalid JSON: {ex}")
+                return []
+
+            # convert api data into objects
+            unsorted = [Departure.from_dict(departure) for departure in departures]
+            return sorted(unsorted, key=lambda d: d.timestamp)
+        else:
+            return self.fetch_departures_directional()
+
+    def fetch_departures_directional(self) -> Optional[list[Departure]]:
         try:
             response = requests.get(
-                url=f"{API_ENDPOINT}",
+                url=f"{API_ENDPOINT_TRIP}",
                 params={
                     "time": (
                         datetime.now() + timedelta(minutes=self.walking_time)
                     ).isoformat(),
                     "format": "json",
                     "limit": API_MAX_RESULTS,
-                    "stopID": self.stop_id,
-                    "isarrival": False,
+                    "origin": self.stop_id,
+                    "destination": self.destination_id,
+                    "isarrivaltime": False,
                     "shorttermchanges": True,
                     "mentzonly": False,
                 },
@@ -144,15 +190,16 @@ def fetch_departures(self) -> Optional[list[Departure]]:
 
         # parse JSON response
         try:
-            departures = response.json().get('Departures')
+            routes = response.json().get('Routes')
         except requests.exceptions.InvalidJSONError as ex:
             _LOGGER.error(f"API invalid JSON: {ex}")
             return []
 
         # convert api data into objects
-        unsorted = [Departure.from_dict(departure) for departure in departures]
-        return sorted(unsorted, key=lambda d: d.timestamp)
-
+        unsorted_departures = [Departure.from_dict_dir(route) for route in routes]
+        #unsorted = [Departure.from_dict(departure) for departure in departures]
+        return sorted(unsorted_departures, key=lambda d: d.timestamp)
+
     def next_departure(self):
         if self.departures and isinstance(self.departures, list):
             return self.departures[0]