Merge pull request #5 from bremor/bremor-patch-3

Add files via upload

custom_components/public_transport_victoria/PublicTransportVictoria/public_transport_victoria.py

@@ -0,0 +1,143 @@
+"""Public Transport Victoria API connector."""
+import aiohttp
+import asyncio
+import datetime
+import hmac
+import logging
+from hashlib import sha1
+
+from homeassistant.util import Throttle
+
+BASE_URL = "https://timetableapi.ptv.vic.gov.au"
+DEPARTURES_PATH = "/v3/departures/route_type/{}/stop/{}/route/{}?direction_id={}&max_results={}"
+DIRECTIONS_PATH = "/v3/directions/route/{}"
+MIN_TIME_BETWEEN_UPDATES = datetime.timedelta(minutes=1)
+MAX_RESULTS = 5
+ROUTE_TYPES_PATH = "/v3/route_types"
+ROUTES_PATH = "/v3/routes?route_types={}"
+STOPS_PATH = "/v3/stops/route/{}/route_type/{}"
+
+_LOGGER = logging.getLogger(__name__)
+
+class Connector:
+    """mydlink cloud connector."""
+
+    manufacturer = "Demonstration Corp"
+
+    def __init__(self, hass, id, api_key, route_type=None, route=None,
+                 direction=None, stop=None):
+        """Init Public Transport Victoria connector."""
+        self.hass = hass
+        self.id = id
+        self.api_key = api_key
+        self.route_type = route_type
+        self.route = route
+        self.direction = direction
+        self.stop = stop
+
+    async def _init(self):
+        """Async Init Public Transport Victoria connector."""
+        self.departures_path = DEPARTURES_PATH.format(
+            self.route_type, self.stop, self.route, self.direction, MAX_RESULTS
+        )
+        await self.async_update()
+
+    async def async_route_types(self):
+        """Test credentials for Public Transport Victoria API."""
+        url = build_URL(self.id, self.api_key, ROUTE_TYPES_PATH)
+
+        async with aiohttp.ClientSession() as session:
+            response = await session.get(url)
+
+        if response is not None and response.status == 200:
+            response = await response.json()
+            route_types = {}
+            for r in response["route_types"]:
+                route_types[r["route_type"]] = r["route_type_name"]
+
+            return route_types
+
+    async def async_routes(self, route_type):
+        """Test credentials for Public Transport Victoria API."""
+        url = build_URL(self.id, self.api_key, ROUTES_PATH.format(route_type))
+
+        async with aiohttp.ClientSession() as session:
+            response = await session.get(url)
+
+        if response is not None and response.status == 200:
+            response = await response.json()
+            routes = {}
+            for r in response["routes"]:
+                routes[r["route_id"]] = r["route_name"]
+
+            self.route_type = route_type
+
+            return routes
+
+    async def async_directions(self, route):
+        """Test credentials for Public Transport Victoria API."""
+        url = build_URL(self.id, self.api_key, DIRECTIONS_PATH.format(route))
+
+        async with aiohttp.ClientSession() as session:
+            response = await session.get(url)
+
+        if response is not None and response.status == 200:
+            response = await response.json()
+            directions = {}
+            for r in response["directions"]:
+                directions[r["direction_id"]] = r["direction_name"]
+
+            self.route = route
+
+            return directions
+
+    async def async_stops(self, route):
+        """Test credentials for Public Transport Victoria API."""
+        url = build_URL(self.id, self.api_key, STOPS_PATH.format(route, self.route_type))
+
+        async with aiohttp.ClientSession() as session:
+            response = await session.get(url)
+
+        if response is not None and response.status == 200:
+            response = await response.json()
+            stops = {}
+            for r in response["stops"]:
+                stops[r["stop_id"]] = r["stop_name"]
+
+            self.route = route
+
+            return stops
+
+    @Throttle(MIN_TIME_BETWEEN_UPDATES)
+    async def async_update(self):
+        """Update the departure information."""
+        url = build_URL(self.id, self.api_key, self.departures_path)
+
+        async with aiohttp.ClientSession() as session:
+            response = await session.get(url)
+
+        if response is not None and response.status == 200:
+            response = await response.json()
+            self.departures = []
+            for r in response["departures"]:
+                if r["estimated_departure_utc"] is not None:
+                    r["departure"] = convert_utc_to_local(r["estimated_departure_utc"])
+                else:
+                    r["departure"] = convert_utc_to_local(r["scheduled_departure_utc"])
+                self.departures.append(r)
+
+        for departure in self.departures:
+            _LOGGER.debug(departure)
+
+def build_URL(id, api_key, request):
+    request = request + ('&' if ('?' in request) else '?')
+    raw = request + 'devid={}'.format(id)
+    hashed = hmac.new(api_key.encode('utf-8'), raw.encode('utf-8'), sha1)
+    signature = hashed.hexdigest()
+    return BASE_URL + raw + '&signature={}'.format(signature)
+
+def convert_utc_to_local(utc):
+    d = datetime.datetime.strptime(utc, '%Y-%m-%dT%H:%M:%SZ')
+    d = d.replace(tzinfo=datetime.timezone.utc)
+    d = d.astimezone()
+    return d.strftime('%I:%M %p')

custom_components/public_transport_victoria/__init__.py

@@ -1 +1,64 @@
+"""Public Transport Victoria integration."""
+import asyncio
 
+from homeassistant.config_entries import ConfigEntry
+from homeassistant.const import CONF_API_KEY, CONF_ID
+from homeassistant.core import HomeAssistant
+
+from .const import (
+    CONF_DIRECTION, CONF_DIRECTION_NAME, CONF_ROUTE, CONF_ROUTE_NAME,
+    CONF_ROUTE_TYPE, CONF_ROUTE_TYPE_NAME, CONF_STOP, CONF_STOP_NAME, DOMAIN
+)
+from .PublicTransportVictoria.public_transport_victoria import Connector
+
+PLATFORMS = ["sensor"]
+
+
+async def async_setup(hass: HomeAssistant, config: dict):
+    """Set up the MyDlink component."""
+    # Ensure our name space for storing objects is a known type. A dict is
+    # common/preferred as it allows a separate instance of your class for each
+    # instance that has been created in the UI.
+    hass.data.setdefault(DOMAIN, {})
+
+    return True
+
+
+async def async_setup_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
+    """Set up MyDlink from a config entry."""
+    connector = Connector(hass,
+                          entry.data[CONF_ID],
+                          entry.data[CONF_API_KEY],
+                          entry.data[CONF_ROUTE_TYPE],
+                          entry.data[CONF_ROUTE],
+                          entry.data[CONF_DIRECTION],
+                          entry.data[CONF_STOP],
+    )
+    await connector._init()
+
+    hass.data[DOMAIN][entry.entry_id] = connector
+
+    # This creates each HA object for each platform your device requires.
+    # It's done by calling the `async_setup_entry` function in each platform module.
+    for component in PLATFORMS:
+        hass.async_create_task(
+            hass.config_entries.async_forward_entry_setup(entry, component)
+        )
+
+    return True
+
+
+async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry):
+    """Unload a config entry."""
+    unload_ok = all(
+        await asyncio.gather(
+            *[
+                hass.config_entries.async_forward_entry_unload(entry, component)
+                for component in PLATFORMS
+            ]
+        )
+    )
+    if unload_ok:
+        hass.data[DOMAIN].pop(entry.entry_id)
+
+    return unload_ok

custom_components/public_transport_victoria/config_flow.py

@@ -0,0 +1,176 @@
+"""Config flow for Public Transport Victoria integration."""
+import hashlib
+import logging
+
+import voluptuous as vol
+
+from homeassistant import config_entries, core, exceptions
+from homeassistant.const import CONF_API_KEY, CONF_ID
+
+from .const import (
+    CONF_DIRECTION, CONF_DIRECTION_NAME, CONF_ROUTE, CONF_ROUTE_NAME,
+    CONF_ROUTE_TYPE, CONF_ROUTE_TYPE_NAME, CONF_STOP, CONF_STOP_NAME, DOMAIN
+)
+from .PublicTransportVictoria.public_transport_victoria import Connector
+
+_LOGGER = logging.getLogger(__name__)
+
+
+class ConfigFlow(config_entries.ConfigFlow, domain=DOMAIN):
+    """Handle a config flow for Public Transport Victoria."""
+
+    VERSION = 1
+
+    CONNECTION_CLASS = config_entries.CONN_CLASS_CLOUD_POLL
+
+    async def async_step_user(self, user_input=None):
+        """Handle the initial step."""
+        data_schema = vol.Schema({
+            vol.Required(CONF_ID, default="3000906"): str,
+            vol.Required(CONF_API_KEY, default="542fbf84-930b-467c-844d-21d74f15c38a"): str,
+        })
+
+        errors = {}
+        if user_input is not None:
+            try:
+                self.connector = Connector(self.hass, user_input[CONF_ID], user_input[CONF_API_KEY])
+                self.route_types = await self.connector.async_route_types()
+
+                if not self.route_types:
+                    raise CannotConnect
+
+                self.data = user_input
+
+                return await self.async_step_route_types()
+
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+
+        # If there is no user input or there were errors, show the form again.
+        return self.async_show_form(
+            step_id="user", data_schema=data_schema, errors=errors
+        )
+
+    async def async_step_route_types(self, user_input=None):
+        """Handle the route types step."""
+        data_schema = vol.Schema({
+            vol.Required(CONF_ROUTE_TYPE, default="0"): vol.In(self.route_types),
+        })
+
+        errors = {}
+        if user_input is not None:
+            try:
+                self.routes = await self.connector.async_routes(
+                    user_input[CONF_ROUTE_TYPE]
+                )
+
+                self.data[CONF_ROUTE_TYPE] = user_input[CONF_ROUTE_TYPE]
+                self.data[CONF_ROUTE_TYPE_NAME] = self.route_types[user_input[CONF_ROUTE_TYPE]]
+
+                return await self.async_step_routes()
+
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+
+        # If there is no user input or there were errors, show the form again.
+        return self.async_show_form(
+            step_id="route_types", data_schema=data_schema, errors=errors
+        )
+
+    async def async_step_routes(self, user_input=None):
+        """Handle the route types step."""
+        data_schema = vol.Schema({
+            vol.Required(CONF_ROUTE, default=next(iter(self.routes))): vol.In(self.routes),
+        })
+
+        errors = {}
+        if user_input is not None:
+            try:
+                self.directions = await self.connector.async_directions(
+                    user_input[CONF_ROUTE]
+                )
+
+                self.data[CONF_ROUTE] = user_input[CONF_ROUTE]
+                self.data[CONF_ROUTE_NAME] = self.routes[user_input[CONF_ROUTE]]
+
+                return await self.async_step_directions()
+
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+
+        # If there is no user input or there were errors, show the form again.
+        return self.async_show_form(
+            step_id="routes", data_schema=data_schema, errors=errors
+        )
+
+    async def async_step_directions(self, user_input=None):
+        """Handle the direction types step."""
+        data_schema = vol.Schema({
+            vol.Required(CONF_DIRECTION, default=next(iter(self.directions))): vol.In(self.directions),
+        })
+
+        errors = {}
+        if user_input is not None:
+            try:
+                self.stops = await self.connector.async_stops(
+                    self.data[CONF_ROUTE]
+                )
+
+                self.data[CONF_DIRECTION] = user_input[CONF_DIRECTION]
+                self.data[CONF_DIRECTION_NAME] = self.directions[user_input[CONF_DIRECTION]]
+
+                return await self.async_step_stops()
+
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+
+        # If there is no user input or there were errors, show the form again.
+        return self.async_show_form(
+            step_id="directions", data_schema=data_schema, errors=errors
+        )
+
+    async def async_step_stops(self, user_input=None):
+        """Handle the stops types step."""
+        data_schema = vol.Schema({
+            vol.Required(CONF_STOP, default=next(iter(self.stops))): vol.In(self.stops),
+        })
+
+        errors = {}
+        if user_input is not None:
+            try:
+                self.data[CONF_STOP] = user_input[CONF_STOP]
+                self.data[CONF_STOP_NAME] = self.stops[user_input[CONF_STOP]]
+
+                title = "{} line to {} from {}".format(
+                    self.data[CONF_ROUTE_NAME],
+                    self.data[CONF_DIRECTION_NAME],
+                    self.data[CONF_STOP_NAME]
+                )
+
+                return self.async_create_entry(title=title, data=self.data)
+
+            except CannotConnect:
+                errors["base"] = "cannot_connect"
+            except Exception:
+                _LOGGER.exception("Unexpected exception")
+                errors["base"] = "unknown"
+
+        # If there is no user input or there were errors, show the form again.
+        return self.async_show_form(
+            step_id="stops", data_schema=data_schema, errors=errors
+        )
+
+class CannotConnect(exceptions.HomeAssistantError):
+    """Error to indicate we cannot connect."""

custom_components/public_transport_victoria/const.py

@@ -0,0 +1,11 @@
+"""Constants for the Public Transport Victoria integration."""
+
+CONF_DIRECTION = "direction"
+CONF_DIRECTION_NAME = "direction_name"
+CONF_ROUTE = "route"
+CONF_ROUTE_NAME = "route_name"
+CONF_ROUTE_TYPE = "route_type"
+CONF_ROUTE_TYPE_NAME = "route_type_name"
+CONF_STOP = "stop"
+CONF_STOP_NAME = "stop_name"
+DOMAIN = "public_transport_victoria"