From 233f40dd41b2968d0ef72939a890bc613711fbc6 Mon Sep 17 00:00:00 2001 From: "google-labs-jules[bot]" <161369871+google-labs-jules[bot]@users.noreply.github.com> Date: Tue, 13 Jan 2026 23:29:30 +0000 Subject: [PATCH] Vendor idfm-api library into custom component - Copy the `idfm_api` source code from the custom fork (branch `feature/expose-request-error...`) directly into `custom_components/idfm/idfm_api/`. - Remove the external `git+https://` dependency from `manifest.json`. - Update all internal imports in the component (e.g., `from idfm_api import ...`) to relative imports (`from .idfm_api import ...`) to use the local copy. - This resolves dependency installation issues and ensures the component has access to the required `RequestError` class. - Update version to 2.3.2. --- API_INFO.md | 2 +- custom_components/idfm/__init__.py | 2 +- custom_components/idfm/api_wrapper.py | 4 +- custom_components/idfm/config_flow.py | 4 +- custom_components/idfm/entity.py | 2 +- custom_components/idfm/idfm_api/__init__.py | 311 ++++++++++++++++++ .../idfm/idfm_api/attribution.py | 11 + custom_components/idfm/idfm_api/dataset.py | 149 +++++++++ custom_components/idfm/idfm_api/models.py | 293 +++++++++++++++++ custom_components/idfm/idfm_api/utils.py | 36 ++ custom_components/idfm/manifest.json | 4 +- 11 files changed, 808 insertions(+), 10 deletions(-) create mode 100644 custom_components/idfm/idfm_api/__init__.py create mode 100644 custom_components/idfm/idfm_api/attribution.py create mode 100644 custom_components/idfm/idfm_api/dataset.py create mode 100644 custom_components/idfm/idfm_api/models.py create mode 100644 custom_components/idfm/idfm_api/utils.py diff --git a/API_INFO.md b/API_INFO.md index c6f6ecb..9e0a5d4 100644 --- a/API_INFO.md +++ b/API_INFO.md @@ -7,7 +7,7 @@ Cette intégration utilise la librairie `idfm-api` pour communiquer avec les ser Ces appels nécessitent une clé d'API (token) configurée dans l'intégration. ### 1. Temps Réel (Stop Monitoring) -Récupère les prochains passages pour un arrêt donné. +Récupère les prochains passages pour un arrêt donné.. - **URL**: `https://prim.iledefrance-mobilites.fr/marketplace/stop-monitoring` - **Méthode**: `GET` - **Paramètres**: diff --git a/custom_components/idfm/__init__.py b/custom_components/idfm/__init__.py index 2a1972d..d897044 100644 --- a/custom_components/idfm/__init__.py +++ b/custom_components/idfm/__init__.py @@ -11,7 +11,7 @@ from homeassistant.exceptions import ConfigEntryNotReady from homeassistant.helpers.aiohttp_client import async_get_clientsession from homeassistant.helpers.update_coordinator import DataUpdateCoordinator, UpdateFailed -from idfm_api.models import TransportType +from .idfm_api.models import TransportType from .api_wrapper import MultiKeyIDFMApi from .const import ( diff --git a/custom_components/idfm/api_wrapper.py b/custom_components/idfm/api_wrapper.py index 0900acc..287e05f 100644 --- a/custom_components/idfm/api_wrapper.py +++ b/custom_components/idfm/api_wrapper.py @@ -2,8 +2,8 @@ import logging from typing import List, Optional -from idfm_api import IDFMApi, RequestError -from idfm_api.models import TrafficData, InfoData, ReportData, LineData, StopData, TransportType +from .idfm_api import IDFMApi, RequestError +from .idfm_api.models import TrafficData, InfoData, ReportData, LineData, StopData, TransportType _LOGGER = logging.getLogger(__name__) diff --git a/custom_components/idfm/config_flow.py b/custom_components/idfm/config_flow.py index e698e64..148aa8f 100644 --- a/custom_components/idfm/config_flow.py +++ b/custom_components/idfm/config_flow.py @@ -4,8 +4,8 @@ import voluptuous as vol from aiohttp import ClientSession from homeassistant import config_entries -from idfm_api.dataset import Dataset -from idfm_api.models import TransportType +from .idfm_api.dataset import Dataset +from .idfm_api.models import TransportType from .api_wrapper import MultiKeyIDFMApi diff --git a/custom_components/idfm/entity.py b/custom_components/idfm/entity.py index 622815f..d856a44 100644 --- a/custom_components/idfm/entity.py +++ b/custom_components/idfm/entity.py @@ -1,6 +1,6 @@ """IDFMEntity class""" from homeassistant.helpers.update_coordinator import CoordinatorEntity -from idfm_api.attribution import ( +from .idfm_api.attribution import ( IDFM_API_LICENCE, IDFM_API_LICENCE_LINK, IDFM_API_LINK, diff --git a/custom_components/idfm/idfm_api/__init__.py b/custom_components/idfm/idfm_api/__init__.py new file mode 100644 index 0000000..076ff29 --- /dev/null +++ b/custom_components/idfm/idfm_api/__init__.py @@ -0,0 +1,311 @@ +import asyncio +import logging +from typing import List, Optional + +import aiohttp +import async_timeout + +from .dataset import Dataset +from .models import ( + InfoData, + LineData, + ReportData, + StopData, + TrafficData, + TransportType, +) + +TIMEOUT = 60 +_LOGGER: logging.Logger = logging.getLogger(__package__) + + +class IDFMApi: + def __init__( + self, session: aiohttp.ClientSession, apikey: str, timeout: int = TIMEOUT + ) -> None: + self._session = session + self._apikey = apikey + self._timeout = timeout + + async def __request(self, url): + """ + API request helper for PRIM + Args: + url: the url to request + Returns: + A json object + Raises: + UnknownIdentifierException + """ + try: + async with async_timeout.timeout(self._timeout): + response = await self._session.get( + url, + headers={ + "apiKey": self._apikey, + "Content-Type": "application/json", + "Accept-encoding": "gzip, deflate", + }, + ) + if response.status != 200: + try: + err = (await response.json())["Siri"]["ServiceDelivery"][ + "StopMonitoringDelivery" + ][0]["ErrorCondition"]["ErrorInformation"]["ErrorText"] + if ( + err == "Le couple MonitoringRef/LineRef n'existe pas" + or err + == "La requête contient des identifiants qui sont inconnus" + ): + raise UnknownIdentifierException() + except KeyError: + pass + _LOGGER.warn( + "Error while fetching information from %s - %s", + url, + response._body, + ) + raise RequestError(response.status, response._body) + resp = (await response.json())["Siri"]["ServiceDelivery"] + if "GeneralMessageDelivery" in resp: + resp = resp["GeneralMessageDelivery"][0] + elif "StopMonitoringDelivery" in resp: + resp = resp["StopMonitoringDelivery"][0] + + if resp["Status"] == "false": + _LOGGER.warn( + "Error while fetching information from %s - %s", + url, + response._body, + ) + return None + + return resp + + except asyncio.TimeoutError as exception: + _LOGGER.error( + "Timeout error fetching information from %s - %s", + url, + exception, + ) + + async def __navitia_request(self, url): + """ + API request helper for navitia + Args: + url: the url to request + Returns: + A json object + Raises: + UnknownIdentifierException + """ + try: + async with async_timeout.timeout(self._timeout): + response = await self._session.get( + url, + headers={ + "apiKey": self._apikey, + "Content-Type": "application/json", + "Accept-encoding": "gzip, deflate", + }, + ) + if response.status != 200: + _LOGGER.warn( + "Error while fetching information from %s - %s", + url, + response._body, + ) + raise RequestError(response.status, response._body) + + return await response.json() + + except asyncio.TimeoutError as exception: + _LOGGER.error( + "Timeout error fetching information from %s - %s", + url, + exception, + ) + return None + + async def get_stops(self, line_id: str) -> List[StopData]: + """ + Return a list of stop areas corresponding to the specified line + Args: + line_id: A string indicating id of a line + Returns: + A list of StopData objects + """ + ret = [] + data = await Dataset.get_stops(self._session) + if line_id in data: + for i in data[line_id]: + ret.append(StopData.from_json(i)) + return ret + + async def get_traffic( + self, + stop_id: str, + destination_name: Optional[str] = None, + direction_name: Optional[str] = None, + line_id: Optional[str] = None, + ) -> List[TrafficData]: + """ + Returns the next schedules in a line for a specified depart area to an optional destination + + Args: + stop_id: A string indicating the id of the depart stop area + destination_name: A string indicating the final destination (I.E. the station name returned by get_directions), the schedules for all the available destinations are returned if not specified + direction_name: A boolean indicating the direction of a train, ignored if not specified + line_id: A string indicating id of a line (if not specified, all schedules for this stop/direction will be returned regardless of the line) + Returns: + A list of TrafficData objects + """ + + # for backward compatibility where only the stoppoint id is specified + if stop_id[0:4] != "STIF": + stop_id = f"STIF:StopPoint:Q:{stop_id.split(':')[-1]}:" + + line = f"&LineRef=STIF:Line::{line_id}:" if line_id is not None else "" + request = f"https://prim.iledefrance-mobilites.fr/marketplace/stop-monitoring?MonitoringRef={stop_id}" + try: + response = await self.__request(request + line) + except UnknownIdentifierException: + # if the MonitoringRef/LineRef couple does not exists, fallback to use only the MonitoringRef + _LOGGER.debug( + "unknown MonitoringRef/LineRef couple, falling back to only MonitoringRef" + ) + response = await self.__request(request) + + ret = [] + for i in response["MonitoredStopVisit"]: + d = TrafficData.from_json(i) + if ( + d + and (direction_name is None or d.direction == direction_name) + and (destination_name is None or d.destination_name == destination_name) + ): + ret.append(d) + return sorted(ret) + + async def get_destinations( + self, + stop_id: str, + direction_name: Optional[str] = None, + line_id: Optional[str] = None, + ) -> List[str]: + """ + Returns the available destinations for a specified line + + Args: + stop_id: A string indicating the id of the depart stop area + direction_name: The direction of a train + line_id: A string indicating id of a line (if not specified, all destinations for this stop will be returned regardless of the line) + Returns: + A list of string representing the stations names + """ + ret = set() + for i in await self.get_traffic( + stop_id, direction_name=direction_name, line_id=line_id + ): + ret.add(i.destination_name) + return list(ret) + + async def get_directions( + self, stop_id: str, line_id: Optional[str] = None + ) -> List[str]: + """ + Returns the available directions for a specified line + + Args: + stop_id: A string indicating the id of the depart stop area + line_id: A string indicating id of a line (if not specified, all directions for this stop will be returned regardless of the line) + Returns: + A list of string representing the stations names + """ + ret = set() + for i in await self.get_traffic(stop_id, line_id=line_id): + ret.add(i.direction) + return list(ret) + + async def get_infos(self, line_id: str) -> List[InfoData]: + """ + Returns the traffic informations (usually the current/planned perturbations) for the specified line + + Warning: DEPRECATED in favor of get_line_reports + + Args: + line_id: A string indicating the id of a line + Returns: + A list of InfoData objects, the list is empty if no perturbations are registered + """ + ret = [] + data = await self.__request( + f"https://prim.iledefrance-mobilites.fr/marketplace/general-message?LineRef=STIF:Line::{line_id}:" + ) + if data: + for i in data["InfoMessage"]: + ret.append(InfoData.from_json(i)) + return ret + + async def get_line_reports( + self, line_id: str, exclude_elevator: bool = True + ) -> List[ReportData]: + """ + Return the traffic informations (usually the current/planned perturbations) for the specified line + + Args: + line_id: A string indicating the id of a line + exclude_elevator: if the elevator failures perturbations should be ignored + Returns: + A list of InfoData objects, the list is empty if no perturbations are registered + """ + ret = [] + data = await self.__navitia_request( + f"https://prim.iledefrance-mobilites.fr/marketplace/v2/navitia/lines%2Fline%3AIDFM%3A{line_id}/line_reports" + ) + if data: + for i in data["disruptions"]: + if ( + not exclude_elevator + or "tags" not in i + or "Ascenseur" not in i["tags"] + ): + ret.append(ReportData.from_json(i)) + return ret + + async def get_lines( + self, transport: Optional[TransportType] = None + ) -> List[LineData]: + """ + Returns the available lines by transport type + + Args: + transport: the transport type, all of them are returned if this is omitted + Returns: + A list of LineData objects + """ + ret = [] + data = await Dataset.get_lines(self._session) + if transport.value in data: + for name, id in data[transport.value].items(): + ret.append(LineData(name=name, id=id, type=transport)) + return ret + + +class UnknownIdentifierException(Exception): + """ + Exception raised when the identifier (MonitoringRef/LineRef) is unknown + """ + + pass + + +class RequestError(Exception): + """ + Exception raised when the API returns an error status code + """ + + def __init__(self, code, body): + self.code = code + self.body = body + super().__init__(f"Error {code} while fetching information - {body}") diff --git a/custom_components/idfm/idfm_api/attribution.py b/custom_components/idfm/idfm_api/attribution.py new file mode 100644 index 0000000..723182b --- /dev/null +++ b/custom_components/idfm/idfm_api/attribution.py @@ -0,0 +1,11 @@ +IDFM_DB_LICENCE = "Licence ODbL Version Française" +IDFM_DB_LICENCE_LINK = "http://vvlibri.org/fr/licence/odbl-10/legalcode/unofficial" +IDFM_DB_SOURCES = { + "Arrêts et lignes associées": "https://data.iledefrance-mobilites.fr/explore/dataset/arrets-lignes", + "Référentiel des lignes de transport en commun d'île-de-France": "https://data.iledefrance-mobilites.fr/explore/dataset/referentiel-des-lignes", + "Référentiel des arrêts : Relations": "https://data.iledefrance-mobilites.fr/explore/dataset/relations", + "Référentiel des arrêts : Zones de correspondance": "https://data.iledefrance-mobilites.fr/explore/dataset/zones-de-correspondance" +} +IDFM_API_LICENCE = "Licence Mobilité" +IDFM_API_LICENCE_LINK = "https://cloud.fabmob.io/s/eYWWJBdM3fQiFNm" +IDFM_API_LINK = "https://prim.iledefrance-mobilites.fr/fr/donnees-dynamiques/idfm-ivtr-requete_unitaire" \ No newline at end of file diff --git a/custom_components/idfm/idfm_api/dataset.py b/custom_components/idfm/idfm_api/dataset.py new file mode 100644 index 0000000..49f9b76 --- /dev/null +++ b/custom_components/idfm/idfm_api/dataset.py @@ -0,0 +1,149 @@ +import aiohttp +import logging + +LINES = "https://data.iledefrance-mobilites.fr/explore/dataset/referentiel-des-lignes/download/?format=json&timezone=Europe/Berlin&lang=fr" +STOP_AND_LINES = "https://data.iledefrance-mobilites.fr/explore/dataset/arrets-lignes/download/?format=json&timezone=Europe/Berlin&lang=fr" +STOP_RELATIONS = "https://data.iledefrance-mobilites.fr/explore/dataset/relations/download/?format=json&timezone=Europe/Berlin&lang=fr" +EXCHANGE_AREAS = "https://data.iledefrance-mobilites.fr/api/explore/v2.1/catalog/datasets/zones-de-correspondance/exports/json?lang=fr&timezone=Europe/Berlin" + +_LOGGER: logging.Logger = logging.getLogger(__package__) + + +class Dataset: + """ + Class used to generate the lines and stops listings + + The data is fetched only once per execution, the result is then cached as this process takes multiple seconds + + To find a list of all the stops we need to use multiple datasets: + LINES + -> get line ID + -> STOP_AND_LINES -> get corresponding stops areas for trains (ZdAId) OR get corresponding stops points for other modes (ArRId) + -> STOP_RELATIONS -> map the ArRid to ZdAId AND map the ZdAId to the exchange area (ZdCId) + -> EXCHANGE_AREAS -> get the exchange area data + + So the process looks like this: LineID -> ZdAId -> ZdCId OR LineID -> ArId -> ZdAId -> ZdCId + """ + + lines = None + stops = None + + @staticmethod + async def get_lines(session: aiohttp.ClientSession) -> dict[str, list[dict]]: + """ + Fetch the latest data from IDFM (if needed) and returns the available lines + + Args: + session: aiohttp session + Returns: + dict[str,list[dict]]: a map of the TransportType to a list of lines (Name:ID) + """ + if Dataset.lines is None: + await Dataset.fetch_data(session) + return Dataset.lines + + @staticmethod + async def get_stops(session: aiohttp.ClientSession) -> dict[str, list[dict]]: + """ + Fetch the latest data from IDFM (if needed) and returns the available stops + + Args: + session: aiohttp session + Returns: + dict[str, list[dict]]: a map of the line id to a list of stops + """ + if Dataset.stops is None: + await Dataset.fetch_data(session) + return Dataset.stops + + @staticmethod + async def fetch_data(session: aiohttp.ClientSession): + """ + Fetch and process the data from IDFM datasets + + Args: + session: the aiohttp session + """ + _LOGGER.debug("fetching idfm datasets") + lines = {} + line_ids = [] + for l in await (await session.get(LINES)).json(): + mode = l["fields"]["transportmode"] + if mode not in lines: + lines[mode] = {} + + name = l["fields"]["name_line"] + if mode == "bus" and "operatorname" in l["fields"]: + name += " / " + l["fields"]["operatorname"] + + lines[mode][name] = l["fields"]["id_line"] + line_ids.append(l["fields"]["id_line"]) + + arid_to_zdaid = {} + zdaid_to_zdcid = {} + for i in await (await session.get(STOP_RELATIONS)).json(): + try: + arid_to_zdaid[i["fields"]["arrid"]] = i["fields"]["zdaid"] + except KeyError: + pass + try: + zdaid_to_zdcid[i["fields"]["zdaid"]] = i["fields"]["zdcid"] + except KeyError: + pass + + zdc = {} + for i in await (await session.get(EXCHANGE_AREAS)).json(): + zdc[i["zdcid"]] = i + + # map line to stops + line_to_stops = {} + stop_ids = {} + for i in await (await session.get(STOP_AND_LINES)).json(): + id = i["fields"]["id"].split(":")[1] + if id not in line_to_stops: + line_to_stops[id] = [] + stop_ids[id] = [] + + if id in line_ids: + stop_id = i["fields"]["stop_id"] + if stop_id.find("monomodalStopPlace") == -1: + try: + stop_id = arid_to_zdaid[stop_id.split(":")[-1]] + except KeyError: + pass + else: + stop_id = stop_id[24:] + + # try to find the corresponding Exchange Area ID (ZdCId) + zdcid = zdaid_to_zdcid.get(stop_id) + + if stop_id not in stop_ids[id]: + line_to_stops[id].append( + { + "exchange_area_id": None + if zdcid is None + else "STIF:StopArea:SP:" + zdcid + ":", + "exchange_area_name": None + if zdcid is None + else zdc[zdcid]["zdcname"], + "stop_id": "STIF:StopPoint:Q:" + stop_id + ":", + "name": i["fields"]["stop_name"], + "city": i["fields"]["nom_commune"], + "zipCode": i["fields"]["code_insee"], + "x": i["fields"]["stop_lat"], + "y": i["fields"]["stop_lon"], + } + ) + stop_ids[id].append(stop_id) + + # remove lines with no associated stops + filtered_lines = {} + for mode, data in lines.items(): + for name, value in data.items(): + if value in line_to_stops: + if mode not in filtered_lines: + filtered_lines[mode] = {} + filtered_lines[mode][name] = value + + Dataset.lines = filtered_lines + Dataset.stops = line_to_stops diff --git a/custom_components/idfm/idfm_api/models.py b/custom_components/idfm/idfm_api/models.py new file mode 100644 index 0000000..50e2057 --- /dev/null +++ b/custom_components/idfm/idfm_api/models.py @@ -0,0 +1,293 @@ +from dataclasses import dataclass +from datetime import datetime, timezone +from enum import Enum, unique +from functools import total_ordering +from zoneinfo import ZoneInfo + +from .utils import strip_html + + +@unique +class TransportType(str, Enum): + """ + Represents the type of transport + """ + + METRO = "metro" + TRAM = "tram" + TRAIN = "rail" + BUS = "bus" + + +@unique +class TransportStatus(str, Enum): + """ + Represents the status of a transport + """ + + ON_TIME = "onTime" + MISSED = "missed" + ARRIVED = "arrived" + NOT_EXPECTED = "notExpected" + DELAYED = "delayed" + EARLY = "early" + CANCELLED = "cancelled" + NO_REPORT = "noReport" + UNKNOWN = "unknown" + + +@dataclass(frozen=True) +class LineData: + """ + Represents a line of a transport + """ + + name: str + id: str + type: TransportType + + +@dataclass(frozen=True) +class StopData: + """ + Represents a stop area of a line + """ + + name: str + stop_id: str + x: float + y: float + zip_code: str + city: str + exchange_area_id: str + exchange_area_name: str + + @staticmethod + def from_json(data: dict): + return StopData( + name=data.get("name"), + stop_id=data.get("stop_id"), + x=data.get("x"), + y=data.get("y"), + zip_code=data.get("zipCode"), + city=data.get("city"), + exchange_area_id=data.get("exchange_area_id"), + exchange_area_name=data.get("exchange_area_name"), + ) + + +@dataclass(frozen=True) +class InfoData: + """ + Represents a traffic information fragment + """ + + id: str + name: str + message: str + start_time: datetime + end_time: datetime + severity: int + type: str + + @staticmethod + def from_json(data: dict): + name = "" + message = "" + if "Message" in data["Content"]: + for i in data["Content"]["Message"]: + if "MessageType" in i: + if i["MessageType"] == "TEXT_ONLY": + message = i["MessageText"]["value"] + if i["MessageType"] == "SHORT_MESSAGE": + name = i["MessageText"]["value"] + + return InfoData( + name=name, + id=data.get("id"), + message=message, + start_time=datetime.strptime( + data.get("RecordedAtTime"), "%Y-%m-%dT%H:%M:%S.%fZ" + ).replace(tzinfo=timezone.utc), + end_time=datetime.strptime( + data.get("ValidUntilTime"), "%Y-%m-%dT%H:%M:%S.%fZ" + ).replace(tzinfo=timezone.utc), + type=data["InfoChannelRef"]["value"], + severity=data.get("InfoMessageVersion"), + ) + + +@dataclass(frozen=True) +class ReportData: + """ + Represents a traffic information fragment (navitia version) + """ + + id: str + name: str + message: str + periods: list[(datetime, datetime)] + severity: int + effect: str + category: str + cause: str + type: str + + @staticmethod + def from_json(data: dict): + name = "" + message = "" + if "messages" in data: + for i in data["messages"]: + if i["channel"]["name"] == "titre": + name = i["text"] + elif i["channel"]["name"] == "moteur": + message = strip_html(i["text"]) + + periods = [] + for i in data["application_periods"]: + periods.append( + ( + datetime.strptime(i["begin"], "%Y%m%dT%H%M%S").replace( + tzinfo=ZoneInfo("Europe/Paris") + ), + datetime.strptime(i["end"], "%Y%m%dT%H%M%S").replace( + tzinfo=ZoneInfo("Europe/Paris") + ), + ) + ) + + return ReportData( + name=name, + id=data.get("id"), + message=message, + periods=periods, + category=data.get("category"), + cause=data.get("cause"), + severity=data["severity"]["priority"], + effect=data["severity"]["effect"], + type=data["severity"]["name"], + ) + + +@dataclass(frozen=True) +@total_ordering +class TrafficData: + """ + Represents a schedule for a specific path + """ + + line_id: str + note: str + destination_name: str + destination_id: str + direction: str + schedule: datetime + retarted: bool + at_stop: bool + platform: str + status: str + + @staticmethod + def from_json(data: dict): + try: + dir = data["MonitoredVehicleJourney"]["DirectionName"][0]["value"] + except (KeyError, IndexError): + dir = data["MonitoredVehicleJourney"]["DestinationName"][0]["value"] + + try: + note = data["MonitoredVehicleJourney"]["JourneyNote"][0]["value"] + except (KeyError, IndexError): + note = "" + + sch = None + if "ExpectedArrivalTime" in data["MonitoredVehicleJourney"]["MonitoredCall"]: + sch = datetime.strptime( + data["MonitoredVehicleJourney"]["MonitoredCall"]["ExpectedArrivalTime"], + "%Y-%m-%dT%H:%M:%S.%fZ", + ).replace(tzinfo=timezone.utc) + elif ( + "ExpectedDepartureTime" in data["MonitoredVehicleJourney"]["MonitoredCall"] + ): + sch = datetime.strptime( + data["MonitoredVehicleJourney"]["MonitoredCall"][ + "ExpectedDepartureTime" + ], + "%Y-%m-%dT%H:%M:%S.%fZ", + ).replace(tzinfo=timezone.utc) + else: + return None + + try: + atstop = data["MonitoredVehicleJourney"]["MonitoredCall"]["VehicleAtStop"] + except KeyError: + atstop = None + + try: + plat = data["MonitoredVehicleJourney"]["MonitoredCall"][ + "ArrivalPlatformName" + ]["value"] + except KeyError: + plat = "" + + if ( + "ArrivalStatus" in data["MonitoredVehicleJourney"]["MonitoredCall"] + and data["MonitoredVehicleJourney"]["MonitoredCall"]["ArrivalStatus"] != "" + ): + status = TransportStatus( + data["MonitoredVehicleJourney"]["MonitoredCall"]["ArrivalStatus"] + ) + elif ( + "DepartureStatus" in data["MonitoredVehicleJourney"]["MonitoredCall"] + and data["MonitoredVehicleJourney"]["MonitoredCall"]["DepartureStatus"] + != "" + ): + status = TransportStatus( + data["MonitoredVehicleJourney"]["MonitoredCall"]["DepartureStatus"] + ) + else: + status = TransportStatus.UNKNOWN + + return TrafficData( + line_id=data["MonitoredVehicleJourney"]["LineRef"]["value"], + note=note, + destination_name=data["MonitoredVehicleJourney"]["DestinationName"][0][ + "value" + ], + destination_id=data["MonitoredVehicleJourney"]["DestinationRef"]["value"], + direction=dir, + schedule=sch, + retarted=status + not in [ + TransportStatus.ON_TIME, + TransportStatus.ARRIVED, + TransportStatus.UNKNOWN, + ], + at_stop=atstop, + platform=plat, + status=status, + ) + + def __eq__(self, other): + if type(other) is TrafficData: + return ( + self.schedule == other.schedule + and self.line_id == other.line_id + and self.destination_id == other.destination_id + ) + else: + return False + + def __lt__(self, other): + if type(other) is datetime: + return self.schedule < other + elif type(other) is TrafficData: + return ( + (self.schedule is None or other.schedule is None) + or self.schedule < other.schedule + ) or ( + (self.destination_name is None or other.destination_name is None) + or self.destination_name < other.destination_name + ) + else: + return NotImplemented diff --git a/custom_components/idfm/idfm_api/utils.py b/custom_components/idfm/idfm_api/utils.py new file mode 100644 index 0000000..fc4de9c --- /dev/null +++ b/custom_components/idfm/idfm_api/utils.py @@ -0,0 +1,36 @@ +from io import StringIO +from html.parser import HTMLParser + +# from https://stackoverflow.com/questions/753052/strip-html-from-strings-in-python + + +class MLStripper(HTMLParser): + """ + Class used to remove HTML tags from a string + """ + + def __init__(self): + super().__init__() + self.reset() + self.strict = False + self.convert_charrefs = True + self.text = StringIO() + + def handle_data(self, d): + self.text.write(d) + + def get_data(self): + return self.text.getvalue() + + +def strip_html(html): + """ + Removes HTML tags from the specified string + Args: + html: the string that contains the HTML tags to remove + Returns: + The specified string without the HTML tags + """ + s = MLStripper() + s.feed(html) + return s.get_data() diff --git a/custom_components/idfm/manifest.json b/custom_components/idfm/manifest.json index 36f7644..6049dae 100644 --- a/custom_components/idfm/manifest.json +++ b/custom_components/idfm/manifest.json @@ -7,8 +7,6 @@ "documentation": "https://github.com/azman0101/idfm", "iot_class": "cloud_polling", "issue_tracker": "https://github.com/azman0101/idfm/issues", - "requirements": [ - "git+https://github.com/azman0101/idfm-api.git@feature/expose-request-error-13065518971479384433#idfm-api" - ], + "requirements": [], "version": "2.3.2" }