sensor.py

Go to the documentation of this file.

"""Support for OASA Telematics from telematics.oasa.gr."""
 
from __future__ import annotations
 
from datetime import timedelta
import logging
from operator import itemgetter
 
import oasatelematics
import voluptuous as vol
 
from homeassistant.components.sensor import (
    PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
    SensorDeviceClass,
    SensorEntity,
)
from homeassistant.const import CONF_NAME
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 dt as dt_util
 
_LOGGER = logging.getLogger(__name__)
 
ATTR_STOP_ID = "stop_id"
ATTR_STOP_NAME = "stop_name"
ATTR_ROUTE_ID = "route_id"
ATTR_ROUTE_NAME = "route_name"
ATTR_NEXT_ARRIVAL = "next_arrival"
ATTR_SECOND_NEXT_ARRIVAL = "second_next_arrival"
ATTR_NEXT_DEPARTURE = "next_departure"
 
CONF_STOP_ID = "stop_id"
CONF_ROUTE_ID = "route_id"
 
DEFAULT_NAME = "OASA Telematics"
 
 
SCAN_INTERVAL = timedelta(seconds=60)
 
PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_STOP_ID): cv.string,
        vol.Required(CONF_ROUTE_ID): cv.string,
        vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
    }
)
 
 
def setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the OASA Telematics sensor."""
    name = config[CONF_NAME]
    stop_id = config[CONF_STOP_ID]
    route_id = config.get(CONF_ROUTE_ID)
 
    data = OASATelematicsData(stop_id, route_id)
 
    add_entities([OASATelematicsSensor(data, stop_id, route_id, name)], True)
 
 
class OASATelematicsSensor(SensorEntity):
    """Implementation of the OASA Telematics sensor."""
 
    _attr_attribution = "Data retrieved from telematics.oasa.gr"
    _attr_icon = "mdi:bus"
 
    def __init__(self, data, stop_id, route_id, name):
        """Initialize the sensor."""
        self.datadata = data
        self._name_name = name
        self._stop_id_stop_id = stop_id
        self._route_id_route_id = route_id
        self._name_data_name_data = self._times_times = self._state_state = None
 
    @property
    def name(self):
        """Return the name of the sensor."""
        return self._name_name
 
    @property
    def device_class(self):
        """Return the class of this sensor."""
        return SensorDeviceClass.TIMESTAMP
 
    @property
    def native_value(self):
        """Return the state of the sensor."""
        return self._state_state
 
    @property
    def extra_state_attributes(self):
        """Return the state attributes."""
        params = {}
        if self._times_times is not None:
            next_arrival_data = self._times_times[0]
            if ATTR_NEXT_ARRIVAL in next_arrival_data:
                next_arrival = next_arrival_data[ATTR_NEXT_ARRIVAL]
                params.update({ATTR_NEXT_ARRIVAL: next_arrival.isoformat()})
            if len(self._times_times) > 1:
                second_next_arrival_time = self._times_times[1][ATTR_NEXT_ARRIVAL]
                if second_next_arrival_time is not None:
                    second_arrival = second_next_arrival_time
                    params.update(
                        {ATTR_SECOND_NEXT_ARRIVAL: second_arrival.isoformat()}
                    )
            params.update(
                {
                    ATTR_ROUTE_ID: self._times_times[0][ATTR_ROUTE_ID],
                    ATTR_STOP_ID: self._stop_id_stop_id,
                }
            )
        params.update(
            {
                ATTR_ROUTE_NAME: self._name_data_name_data[ATTR_ROUTE_NAME],
                ATTR_STOP_NAME: self._name_data_name_data[ATTR_STOP_NAME],
            }
        )
        return {k: v for k, v in params.items() if v}
 
    def update(self) -> None:
        """Get the latest data from OASA API and update the states."""
        self.datadata.update()
        self._times_times = self.datadata.info
        self._name_data_name_data = self.datadata.name_data
        next_arrival_data = self._times_times[0]
        if ATTR_NEXT_ARRIVAL in next_arrival_data:
            self._state_state = next_arrival_data[ATTR_NEXT_ARRIVAL]
 
 
class OASATelematicsData:
    """The class for handling data retrieval."""
 
    def __init__(self, stop_id, route_id):
        """Initialize the data object."""
        self.stop_idstop_id = stop_id
        self.route_idroute_id = route_id
        self.infoinfo = self.empty_resultempty_result()
        self.oasa_apioasa_api = oasatelematics
        self.name_dataname_data = {
            ATTR_ROUTE_NAME: self.get_route_nameget_route_name(),
            ATTR_STOP_NAME: self.get_stop_nameget_stop_name(),
        }
 
    def empty_result(self):
        """Object returned when no arrivals are found."""
        return [{ATTR_ROUTE_ID: self.route_idroute_id}]
 
    def get_route_name(self):
        """Get the route name from the API."""
        try:
            route = self.oasa_apioasa_api.getRouteName(self.route_idroute_id)
            if route:
                return route[0].get("route_departure_eng")
        except TypeError:
            _LOGGER.error("Cannot get route name from OASA API")
        return None
 
    def get_stop_name(self):
        """Get the stop name from the API."""
        try:
            name_data = self.oasa_apioasa_api.getStopNameAndXY(self.stop_idstop_id)
            if name_data:
                return name_data[0].get("stop_descr_matrix_eng")
        except TypeError:
            _LOGGER.error("Cannot get stop name from OASA API")
        return None
 
    def update(self):
        """Get the latest arrival data from telematics.oasa.gr API."""
        self.infoinfo = []
 
        results = self.oasa_apioasa_api.getStopArrivals(self.stop_idstop_id)
 
        if not results:
            self.infoinfo = self.empty_resultempty_result()
            return
 
        # Parse results
        results = [r for r in results if r.get("route_code") in self.route_idroute_id]
        current_time = dt_util.utcnow()
 
        for result in results:
            if (btime2 := result.get("btime2")) is not None:
                arrival_min = int(btime2)
                timestamp = current_time + timedelta(minutes=arrival_min)
                arrival_data = {
                    ATTR_NEXT_ARRIVAL: timestamp,
                    ATTR_ROUTE_ID: self.route_idroute_id,
                }
                self.infoinfo.append(arrival_data)
 
        if not self.infoinfo:
            _LOGGER.debug("No arrivals with given parameters")
            self.infoinfo = self.empty_resultempty_result()
            return
 
        # Sort the data by time
        sort = sorted(self.infoinfo, key=itemgetter(ATTR_NEXT_ARRIVAL))
        self.infoinfo = sort