init

.gitignore

@@ -0,0 +1,3 @@
+.idea/
+__pycache__/
+*.pyc

__init__.py

@@ -0,0 +1,30 @@
+"""The SNCF integration."""
+from __future__ import annotations
+
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.core import HomeAssistant
+
+from .const import DOMAIN
+
+# TODO List the platforms that you want to support.
+# For your initial PR, limit it to 1 platform.
+PLATFORMS: list[str] = ["light"]
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up SNCF from a config entry."""
+    # TODO Store an API object for your platforms to access
+    # hass.data[DOMAIN][entry.entry_id] = MyApi(...)
+
+    hass.config_entries.async_setup_platforms(entry, PLATFORMS)
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Unload a config entry."""
+    unload_ok = await hass.config_entries.async_unload_platforms(entry, PLATFORMS)
+    if unload_ok:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

config_flow.py

@@ -0,0 +1,73 @@
+"""Config flow for SNCF integration."""
+from __future__ import annotations
+
+import logging
+from typing import Any
+import voluptuous as vol
+from homeassistant import config_entries
+from homeassistant.core import HomeAssistant
+from homeassistant.data_entry_flow import FlowResult
+from homeassistant.exceptions import HomeAssistantError
+from sncf import SNCF
+from .const import DOMAIN
+
+_LOGGER = logging.getLogger(__name__)
+
+STEP_USER_DATA_SCHEMA = vol.Schema(
+    {
+        vol.Required("token"): str,
+        vol.Required("source_stop_point"): str,
+        vol.Required("dest_stop_point"): str,
+        vol.Required("max_transfers"): int,
+        vol.Required("max_duration_secs"): int,
+    }
+)
+
+
+async def validate_input(hass: HomeAssistant, data: dict[str, Any]) -> dict[str, Any]:
+    """Validate the user input allows us to connect.
+    Data has the keys from STEP_USER_DATA_SCHEMA with values provided by the user.
+    """
+    sncf = SNCF(data["token"])
+    token_is_valid = await hass.async_add_executor_job(sncf.test_api_key)
+
+    if not token_is_valid:
+        raise InvalidAuth
+
+    # Return info that you want to store in the config entry.
+    return {"title": "SNCF"}
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for SNCF."""
+
+    VERSION = 1
+
+    async def async_step_user(
+        self, user_input: dict[str, Any] | None = None
+    ) -> FlowResult:
+        """Handle the initial step."""
+        if user_input is None:
+            return self.async_show_form(
+                step_id="user", data_schema=STEP_USER_DATA_SCHEMA
+            )
+
+        errors = {}
+
+        try:
+            info = await validate_input(self.hass, user_input)
+        except InvalidAuth:
+            errors["base"] = "invalid_auth"
+        except Exception:  # pylint: disable=broad-except
+            _LOGGER.exception("Unexpected exception")
+            errors["base"] = "unknown"
+        else:
+            return self.async_create_entry(title=info["title"], data=user_input)
+
+        return self.async_show_form(
+            step_id="user", data_schema=STEP_USER_DATA_SCHEMA, errors=errors
+        )
+
+
+class InvalidAuth(HomeAssistantError):
+    """Error to indicate there is invalid auth."""

const.py

@@ -0,0 +1,3 @@
+"""Constants for the SNCF integration."""
+
+DOMAIN = "sncf"

manifest.json

@@ -0,0 +1,15 @@
+{
+  "domain": "sncf",
+  "name": "SNCF",
+  "config_flow": true,
+  "documentation": "https://www.home-assistant.io/integrations/sncf",
+  "requirements": [],
+  "ssdp": [],
+  "zeroconf": [],
+  "homekit": {},
+  "dependencies": [],
+  "codeowners": [
+    "amazed@git.hipstercat.fr"
+  ],
+  "iot_class": "cloud_polling"
+}

sensor.py

@@ -0,0 +1,41 @@
+import datetime
+from homeassistant.components.switch import SensorEntity
+from homeassistant.core import HomeAssistant
+from homeassistant.const import TIME_MINUTES
+from homeassistant.helpers.entity_platform import AddEntitiesCallback
+from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
+from sncf import SNCF, SNCFJourney
+
+
+def setup_platform(
+    hass: HomeAssistant,
+    config: ConfigType,
+    add_entities: AddEntitiesCallback,
+    discovery_info: DiscoveryInfoType | None = None
+) -> None:
+    """Set up the sensor platform."""
+    add_entities([SNCFNextJourneys(config)])
+
+
+class SNCFNextJourneys(SensorEntity):
+    def __init__(self, config: ConfigType):
+        self._config: ConfigType = config
+        self._next_schedules: list[SNCFJourney] = []
+        self._sncf = SNCF(self._config["token"])
+        self._sncf.set_source_stop_point(self._config["source_stop_point"])
+        self._sncf.set_dest_stop_point(self._config["dest_stop_point"])
+        self._sncf.max_transfers = self._config["max_transfers"]
+        self._sncf.max_duration_secs = self._config["max_duration_secs"]
+
+    @property
+    def unit_of_measurement(self) -> str:
+        """Return the unit of measurement."""
+        return TIME_MINUTES
+
+    @property
+    def native_value(self):
+        """Return the state of the sensor."""
+        return SNCF.api_date_to_datetime(self._next_schedules[0]["sections"][0]['departure_date_time'])
+
+    def update(self):
+        self._next_schedules = self._sncf.get_next_journeys()

sncf/__init__.py

@@ -0,0 +1,3 @@
+from .sncf import SNCF, SNCFJourney
+
+__all__ = ['SNCF', 'SNCFJourney']

sncf/sncf.py

@@ -0,0 +1,73 @@
+import datetime
+import requests
+
+
+class SNCFJourney(dict):
+    def __repr__(self):
+        sections = []
+        for section in self["sections"]:
+            nice_dep_date = SNCF.api_date_to_datetime(section['departure_date_time']).strftime("%H:%M")
+            nice_arr_date = SNCF.api_date_to_datetime(section['arrival_date_time']).strftime("%H:%M")
+            sections.append(f"{nice_dep_date} {section['from']['name']} - {section['to']['name']} {nice_arr_date}")
+        return "\n".join(sections)
+
+
+class SNCF:
+    BASE_URL = "https://api.sncf.com/v1/coverage/sncf"
+
+    def __init__(self, token: str):
+        self.token = token
+        self.source_stop_point: str = ""
+        self.dest_stop_point: str = ""
+        self.allowed_lines: list[str] = []
+        self.max_transfers: int = 0
+        self.max_duration_secs: int = 1800
+        self.prefered_date_format = "%H:%M"
+
+    def set_source_stop_point(self, stop_point: str):
+        self.source_stop_point = stop_point
+
+    def set_dest_stop_point(self, stop_point: str):
+        self.dest_stop_point = stop_point
+
+    def set_allowed_lines(self, allowed_lines: list[str]):
+        self.allowed_lines = allowed_lines
+
+    def get_next_journeys(self, count: int = 1):
+        valid_journeys = []
+        api_response = self.api_request(f"/journeys?from={self.source_stop_point}&to={self.dest_stop_point}&datetime={self.date_now()}&count={count}")
+        journeys = api_response["journeys"]
+        for journey in journeys:
+            if journey["nb_transfers"] <= self.max_transfers and journey["duration"] <= self.max_duration_secs:
+                valid_journeys.append(SNCFJourney(journey))
+        return valid_journeys
+
+    def api_request(self, url):
+        req = requests.get(SNCF.BASE_URL + url, headers={"Authorization": self.token})
+        req.raise_for_status()
+        return req.json()
+
+    @staticmethod
+    def date_now() -> str:
+        return datetime.datetime.now().strftime("%Y%m%dT%H%M%S")
+
+    @staticmethod
+    def api_date_to_datetime(api_date: str) -> datetime:
+        return datetime.datetime.strptime(api_date, "%Y%m%dT%H%M%S")
+
+    def test_api_key(self) -> bool:
+        try:
+            self.api_request("")
+            return True
+        except requests.HTTPError:
+            return False
+
+
+if __name__ == '__main__':
+    sncf = SNCF("5722269b-2e58-49ee-986b-21741438d5ff")
+    sncf.set_source_stop_point("stop_point:SNCF:87214056:Train")
+    sncf.set_dest_stop_point("stop_point:SNCF:87212027:Train")
+    next_journeys = sncf.get_next_journeys(count=5)
+    for next_journey in next_journeys:
+        print(next_journey)
+    #print(json.dumps(next_schedules, indent=4))

strings.json

@@ -0,0 +1,19 @@
+{
+  "config": {
+    "step": {
+      "user": {
+        "data": {
+          "token": "[%key:common::config_flow::data::token%]",
+        }
+      }
+    },
+    "error": {
+      "cannot_connect": "[%key:common::config_flow::error::cannot_connect%]",
+      "invalid_auth": "[%key:common::config_flow::error::invalid_auth%]",
+      "unknown": "[%key:common::config_flow::error::unknown%]"
+    },
+    "abort": {
+      "already_configured": "[%key:common::config_flow::abort::already_configured_device%]"
+    }
+  }
+}

translations/en.json

@@ -0,0 +1,19 @@
+{
+    "config": {
+        "abort": {
+            "already_configured": "Device is already configured"
+        },
+        "error": {
+            "cannot_connect": "Failed to connect",
+            "invalid_auth": "Invalid authentication",
+            "unknown": "Unexpected error"
+        },
+        "step": {
+            "user": {
+                "data": {
+                    "token": "SNCF API Token"
+                }
+            }
+        }
+    }
+}