sensor.py

Go to the documentation of this file.

"""Support for UK public transport data provided by transportapi.com."""
 
from __future__ import annotations
 
from datetime import datetime, timedelta
from http import HTTPStatus
import logging
import re
from typing import Any
 
import requests
import voluptuous as vol
 
from homeassistant.components.sensor import (
    PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
    SensorEntity,
)
from homeassistant.const import CONF_MODE, UnitOfTime
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
from homeassistant.util import Throttle
import homeassistant.util.dt as dt_util
 
_LOGGER = logging.getLogger(__name__)
 
ATTR_ATCOCODE = "atcocode"
ATTR_LOCALITY = "locality"
ATTR_STOP_NAME = "stop_name"
ATTR_REQUEST_TIME = "request_time"
ATTR_NEXT_BUSES = "next_buses"
ATTR_STATION_CODE = "station_code"
ATTR_CALLING_AT = "calling_at"
ATTR_NEXT_TRAINS = "next_trains"
 
CONF_API_APP_KEY = "app_key"
CONF_API_APP_ID = "app_id"
CONF_QUERIES = "queries"
CONF_ORIGIN = "origin"
CONF_DESTINATION = "destination"
 
_QUERY_SCHEME = vol.Schema(
    {
        vol.Required(CONF_MODE): vol.All(cv.ensure_list, [vol.In(["bus", "train"])]),
        vol.Required(CONF_ORIGIN): cv.string,
        vol.Required(CONF_DESTINATION): cv.string,
    }
)
 
PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_API_APP_ID): cv.string,
        vol.Required(CONF_API_APP_KEY): cv.string,
        vol.Required(CONF_QUERIES): [_QUERY_SCHEME],
    }
)
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Get the uk_transport sensor."""
    sensors: list[UkTransportSensor] = []
    number_sensors = len(queries := config[CONF_QUERIES])
    interval = timedelta(seconds=87 * number_sensors)
 
    api_app_id = config[CONF_API_APP_ID]
    api_app_key = config[CONF_API_APP_KEY]
 
    for query in queries:
        if "bus" in query.get(CONF_MODE):
            stop_atcocode = query.get(CONF_ORIGIN)
            bus_direction = query.get(CONF_DESTINATION)
            sensors.append(
                UkTransportLiveBusTimeSensor(
                    api_app_id,
                    api_app_key,
                    stop_atcocode,
                    bus_direction,
                    interval,
                )
            )
 
        elif "train" in query.get(CONF_MODE):
            station_code = query.get(CONF_ORIGIN)
            calling_at = query.get(CONF_DESTINATION)
            sensors.append(
                UkTransportLiveTrainTimeSensor(
                    api_app_id,
                    api_app_key,
                    station_code,
                    calling_at,
                    interval,
                )
            )
 
    add_entities(sensors, True)
 
 
class UkTransportSensor(SensorEntity):
    """Sensor that reads the UK transport web API.
 
    transportapi.com provides comprehensive transport data for UK train, tube
    and bus travel across the UK via simple JSON API. Subclasses of this
    base class can be used to access specific types of information.
    """
 
    TRANSPORT_API_URL_BASE = "https://transportapi.com/v3/uk/"
    _attr_icon = "mdi:train"
    _attr_native_unit_of_measurement = UnitOfTime.MINUTES
 
    def __init__(self, name, api_app_id, api_app_key, url):
        """Initialize the sensor."""
        self._data_data = {}
        self._api_app_id_api_app_id = api_app_id
        self._api_app_key_api_app_key = api_app_key
        self._url_url = self.TRANSPORT_API_URL_BASETRANSPORT_API_URL_BASE + url
        self._name_name = name
        self._state_state = None
 
    @property
    def name(self):
        """Return the name of the sensor."""
        return self._name_name
 
    @property
    def native_value(self):
        """Return the state of the sensor."""
        return self._state_state
 
    def _do_api_request(self, params):
        """Perform an API request."""
        request_params = dict(
            {"app_id": self._api_app_id_api_app_id, "app_key": self._api_app_key_api_app_key}, **params
        )
 
        response = requests.get(self._url_url, params=request_params, timeout=10)
        if response.status_code != HTTPStatus.OK:
            _LOGGER.warning("Invalid response from API")
        elif "error" in response.json():
            if "exceeded" in response.json()["error"]:
                self._state_state = "Usage limits exceeded"
            if "invalid" in response.json()["error"]:
                self._state_state = "Credentials invalid"
        else:
            self._data_data = response.json()
 
 
class UkTransportLiveBusTimeSensor(UkTransportSensor):
    """Live bus time sensor from UK transportapi.com."""
 
    _attr_icon = "mdi:bus"
 
    def __init__(self, api_app_id, api_app_key, stop_atcocode, bus_direction, interval):
        """Construct a live bus time sensor."""
        self._stop_atcocode_stop_atcocode = stop_atcocode
        self._bus_direction_bus_direction = bus_direction
        self._next_buses_next_buses = []
        self._destination_re_destination_re = re.compile(f"{bus_direction}", re.IGNORECASE)
 
        sensor_name = f"Next bus to {bus_direction}"
        stop_url = f"bus/stop/{stop_atcocode}/live.json"
 
        UkTransportSensor.__init__(self, sensor_name, api_app_id, api_app_key, stop_url)
        self.updateupdate = Throttle(interval)(self._update_update)
 
    def _update(self):
        """Get the latest live departure data for the specified stop."""
        params = {"group": "route", "nextbuses": "no"}
 
        self._do_api_request_do_api_request(params)
 
        if self._data_data != {}:
            self._next_buses_next_buses = []
 
            for route, departures in self._data_data["departures"].items():
                for departure in departures:
                    if self._destination_re_destination_re.search(departure["direction"]):
                        self._next_buses_next_buses.append(
                            {
                                "route": route,
                                "direction": departure["direction"],
                                "scheduled": departure["aimed_departure_time"],
                                "estimated": departure["best_departure_estimate"],
                            }
                        )
 
            if self._next_buses_next_buses:
                self._state_state_state = min(
                    _delta_mins(bus["scheduled"]) for bus in self._next_buses_next_buses
                )
            else:
                self._state_state_state = None
 
    @property
    def extra_state_attributes(self) -> dict[str, Any] | None:
        """Return other details about the sensor state."""
        if self._data_data is not None:
            attrs = {ATTR_NEXT_BUSES: self._next_buses_next_buses}
            for key in (
                ATTR_ATCOCODE,
                ATTR_LOCALITY,
                ATTR_STOP_NAME,
                ATTR_REQUEST_TIME,
            ):
                attrs[key] = self._data_data.get(key)
            return attrs
        return None
 
 
class UkTransportLiveTrainTimeSensor(UkTransportSensor):
    """Live train time sensor from UK transportapi.com."""
 
    _attr_icon = "mdi:train"
 
    def __init__(self, api_app_id, api_app_key, station_code, calling_at, interval):
        """Construct a live bus time sensor."""
        self._station_code_station_code = station_code
        self._calling_at_calling_at = calling_at
        self._next_trains_next_trains = []
 
        sensor_name = f"Next train to {calling_at}"
        query_url = f"train/station/{station_code}/live.json"
 
        UkTransportSensor.__init__(
            self, sensor_name, api_app_id, api_app_key, query_url
        )
        self.updateupdate = Throttle(interval)(self._update_update)
 
    def _update(self):
        """Get the latest live departure data for the specified stop."""
        params = {
            "darwin": "false",
            "calling_at": self._calling_at_calling_at,
            "train_status": "passenger",
        }
 
        self._do_api_request_do_api_request(params)
        self._next_trains_next_trains = []
 
        if self._data_data != {}:
            if self._data_data["departures"]["all"] == []:
                self._state_state_state = "No departures"
            else:
                for departure in self._data_data["departures"]["all"]:
                    self._next_trains_next_trains.append(
                        {
                            "origin_name": departure["origin_name"],
                            "destination_name": departure["destination_name"],
                            "status": departure["status"],
                            "scheduled": departure["aimed_departure_time"],
                            "estimated": departure["expected_departure_time"],
                            "platform": departure["platform"],
                            "operator_name": departure["operator_name"],
                        }
                    )
 
                if self._next_trains_next_trains:
                    self._state_state_state = min(
                        _delta_mins(train["scheduled"]) for train in self._next_trains_next_trains
                    )
                else:
                    self._state_state_state = None
 
    @property
    def extra_state_attributes(self) -> dict[str, Any] | None:
        """Return other details about the sensor state."""
        if self._data_data is not None:
            attrs = {
                ATTR_STATION_CODE: self._station_code_station_code,
                ATTR_CALLING_AT: self._calling_at_calling_at,
            }
            if self._next_trains_next_trains:
                attrs[ATTR_NEXT_TRAINS] = self._next_trains_next_trains
            return attrs
        return None
 
 
def _delta_mins(hhmm_time_str):
    """Calculate time delta in minutes to a time in hh:mm format."""
    now = dt_util.now()
    hhmm_time = datetime.strptime(hhmm_time_str, "%H:%M")
 
    hhmm_datetime = now.replace(hour=hhmm_time.hour, minute=hhmm_time.minute)
 
    if hhmm_datetime < now:
        hhmm_datetime += timedelta(days=1)
 
    return (hhmm_datetime - now).total_seconds() // 60