From 4edbbfd275148bd0b3aa03666a19ca6e7951d5d5 Mon Sep 17 00:00:00 2001 From: Hipstercat Date: Mon, 3 Jan 2022 20:56:40 +0100 Subject: [PATCH] init --- .gitignore | 3 ++ __init__.py | 30 ++++++++++++++++++ config_flow.py | 73 ++++++++++++++++++++++++++++++++++++++++++++ const.py | 3 ++ manifest.json | 15 +++++++++ sensor.py | 41 +++++++++++++++++++++++++ sncf/__init__.py | 3 ++ sncf/sncf.py | 73 ++++++++++++++++++++++++++++++++++++++++++++ strings.json | 19 ++++++++++++ translations/en.json | 19 ++++++++++++ 10 files changed, 279 insertions(+) create mode 100644 .gitignore create mode 100644 __init__.py create mode 100644 config_flow.py create mode 100644 const.py create mode 100644 manifest.json create mode 100644 sensor.py create mode 100644 sncf/__init__.py create mode 100644 sncf/sncf.py create mode 100644 strings.json create mode 100644 translations/en.json diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..8c75de4 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +.idea/ +__pycache__/ +*.pyc \ No newline at end of file diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..67905af --- /dev/null +++ b/__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 diff --git a/config_flow.py b/config_flow.py new file mode 100644 index 0000000..45fd302 --- /dev/null +++ b/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.""" diff --git a/const.py b/const.py new file mode 100644 index 0000000..ee78166 --- /dev/null +++ b/const.py @@ -0,0 +1,3 @@ +"""Constants for the SNCF integration.""" + +DOMAIN = "sncf" diff --git a/manifest.json b/manifest.json new file mode 100644 index 0000000..8e2c74c --- /dev/null +++ b/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" +} \ No newline at end of file diff --git a/sensor.py b/sensor.py new file mode 100644 index 0000000..21ba3bb --- /dev/null +++ b/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() diff --git a/sncf/__init__.py b/sncf/__init__.py new file mode 100644 index 0000000..c8ac611 --- /dev/null +++ b/sncf/__init__.py @@ -0,0 +1,3 @@ +from .sncf import SNCF, SNCFJourney + +__all__ = ['SNCF', 'SNCFJourney'] \ No newline at end of file diff --git a/sncf/sncf.py b/sncf/sncf.py new file mode 100644 index 0000000..e1f965a --- /dev/null +++ b/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)) diff --git a/strings.json b/strings.json new file mode 100644 index 0000000..c42ab07 --- /dev/null +++ b/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%]" + } + } +} \ No newline at end of file diff --git a/translations/en.json b/translations/en.json new file mode 100644 index 0000000..b787542 --- /dev/null +++ b/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" + } + } + } + } +} \ No newline at end of file