sensor.py

Go to the documentation of this file.

"""Support for Dublin RTPI information from data.dublinked.ie.
 
For more info on the API see :
https://data.gov.ie/dataset/real-time-passenger-information-rtpi-for-dublin-bus-bus-eireann-luas-and-irish-rail/resource/4b9f2c4f-6bf5-4958-a43a-f12dab04cf61
"""
 
from __future__ import annotations
 
from contextlib import suppress
from datetime import datetime, timedelta
from http import HTTPStatus
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_NAME, 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
import homeassistant.util.dt as dt_util
 
_RESOURCE = "https://data.dublinked.ie/cgi-bin/rtpi/realtimebusinformation"
 
ATTR_STOP_ID = "Stop ID"
ATTR_ROUTE = "Route"
ATTR_DUE_IN = "Due in"
ATTR_DUE_AT = "Due at"
ATTR_NEXT_UP = "Later Bus"
 
CONF_STOP_ID = "stopid"
CONF_ROUTE = "route"
 
DEFAULT_NAME = "Next Bus"
 
 
SCAN_INTERVAL = timedelta(minutes=1)
TIME_STR_FORMAT = "%H:%M"
 
PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_STOP_ID): cv.string,
        vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
        vol.Optional(CONF_ROUTE, default=""): cv.string,
    }
)
 
 
def due_in_minutes(timestamp):
    """Get the time in minutes from a timestamp.
 
    The timestamp should be in the format day/month/year hour/minute/second
    """
    diff = datetime.strptime(timestamp, "%d/%m/%Y %H:%M:%S") - dt_util.now().replace(
        tzinfo=None
    )
 
    return str(int(diff.total_seconds() / 60))
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the Dublin public transport sensor."""
    name = config[CONF_NAME]
    stop = config[CONF_STOP_ID]
    route = config[CONF_ROUTE]
 
    data = PublicTransportData(stop, route)
    add_entities([DublinPublicTransportSensor(data, stop, route, name)], True)
 
 
class DublinPublicTransportSensor(SensorEntity):
    """Implementation of an Dublin public transport sensor."""
 
    _attr_attribution = "Data provided by data.dublinked.ie"
    _attr_icon = "mdi:bus"
 
    def __init__(self, data, stop, route, name):
        """Initialize the sensor."""
        self.datadata = data
        self._name_name = name
        self._stop_stop = stop
        self._route_route = route
        self._times_times = 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
 
    @property
    def extra_state_attributes(self) -> dict[str, Any] | None:
        """Return the state attributes."""
        if self._times_times is not None:
            next_up = "None"
            if len(self._times_times) > 1:
                next_up = f"{self._times[1][ATTR_ROUTE]} in "
                next_up += self._times_times[1][ATTR_DUE_IN]
 
            return {
                ATTR_DUE_IN: self._times_times[0][ATTR_DUE_IN],
                ATTR_DUE_AT: self._times_times[0][ATTR_DUE_AT],
                ATTR_STOP_ID: self._stop_stop,
                ATTR_ROUTE: self._times_times[0][ATTR_ROUTE],
                ATTR_NEXT_UP: next_up,
            }
        return None
 
    @property
    def native_unit_of_measurement(self):
        """Return the unit this state is expressed in."""
        return UnitOfTime.MINUTES
 
    def update(self) -> None:
        """Get the latest data from opendata.ch and update the states."""
        self.datadata.update()
        self._times_times = self.datadata.info
        with suppress(TypeError):
            self._state_state = self._times_times[0][ATTR_DUE_IN]
 
 
class PublicTransportData:
    """The Class for handling the data retrieval."""
 
    def __init__(self, stop, route):
        """Initialize the data object."""
        self.stopstop = stop
        self.routeroute = route
        self.infoinfo = [{ATTR_DUE_AT: "n/a", ATTR_ROUTE: self.routeroute, ATTR_DUE_IN: "n/a"}]
 
    def update(self):
        """Get the latest data from opendata.ch."""
        params = {}
        params["stopid"] = self.stopstop
 
        if self.routeroute:
            params["routeid"] = self.routeroute
 
        params["maxresults"] = 2
        params["format"] = "json"
 
        response = requests.get(_RESOURCE, params, timeout=10)
 
        if response.status_code != HTTPStatus.OK:
            self.infoinfo = [
                {ATTR_DUE_AT: "n/a", ATTR_ROUTE: self.routeroute, ATTR_DUE_IN: "n/a"}
            ]
            return
 
        result = response.json()
 
        if str(result["errorcode"]) != "0":
            self.infoinfo = [
                {ATTR_DUE_AT: "n/a", ATTR_ROUTE: self.routeroute, ATTR_DUE_IN: "n/a"}
            ]
            return
 
        self.infoinfo = []
        for item in result["results"]:
            due_at = item.get("departuredatetime")
            route = item.get("route")
            if due_at is not None and route is not None:
                bus_data = {
                    ATTR_DUE_AT: due_at,
                    ATTR_ROUTE: route,
                    ATTR_DUE_IN: due_in_minutes(due_at),
                }
                self.infoinfo.append(bus_data)
 
        if not self.infoinfo:
            self.infoinfo = [
                {ATTR_DUE_AT: "n/a", ATTR_ROUTE: self.routeroute, ATTR_DUE_IN: "n/a"}
            ]