sensor.py

Go to the documentation of this file.

"""Support for currencylayer.com exchange rates service."""
 
from __future__ import annotations
 
from datetime import timedelta
import logging
 
import requests
import voluptuous as vol
 
from homeassistant.components.sensor import (
    PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
    SensorEntity,
)
from homeassistant.const import CONF_API_KEY, CONF_BASE, CONF_NAME, CONF_QUOTE
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
 
_LOGGER = logging.getLogger(__name__)
_RESOURCE = "http://apilayer.net/api/live"
 
DEFAULT_BASE = "USD"
DEFAULT_NAME = "CurrencyLayer Sensor"
 
 
SCAN_INTERVAL = timedelta(hours=4)
 
PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
    {
        vol.Required(CONF_API_KEY): cv.string,
        vol.Required(CONF_QUOTE): vol.All(cv.ensure_list, [cv.string]),
        vol.Optional(CONF_BASE, default=DEFAULT_BASE): 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 Currencylayer sensor."""
    base = config[CONF_BASE]
    api_key = config[CONF_API_KEY]
    parameters = {"source": base, "access_key": api_key, "format": 1}
 
    rest = CurrencylayerData(_RESOURCE, parameters)
 
    response = requests.get(_RESOURCE, params=parameters, timeout=10)
    if "error" in response.json():
        return
    add_entities(
        (CurrencylayerSensor(rest, base, variable) for variable in config[CONF_QUOTE]),
        True,
    )
 
 
class CurrencylayerSensor(SensorEntity):
    """Implementing the Currencylayer sensor."""
 
    _attr_attribution = "Data provided by currencylayer.com"
    _attr_icon = "mdi:currency"
 
    def __init__(self, rest, base, quote):
        """Initialize the sensor."""
        self.restrest = rest
        self._quote_quote = quote
        self._base_base = base
        self._state_state = None
 
    @property
    def native_unit_of_measurement(self):
        """Return the unit of measurement of this entity, if any."""
        return self._quote_quote
 
    @property
    def name(self):
        """Return the name of the sensor."""
        return self._base_base
 
    @property
    def native_value(self):
        """Return the state of the sensor."""
        return self._state_state
 
    def update(self) -> None:
        """Update current date."""
        self.restrest.update()
        if (value := self.restrest.data) is not None:
            self._state_state = round(value[f"{self._base}{self._quote}"], 4)
 
 
class CurrencylayerData:
    """Get data from Currencylayer.org."""
 
    def __init__(self, resource, parameters):
        """Initialize the data object."""
        self._resource_resource = resource
        self._parameters_parameters = parameters
        self.datadata = None
 
    def update(self):
        """Get the latest data from Currencylayer."""
        try:
            result = requests.get(self._resource_resource, params=self._parameters_parameters, timeout=10)
            if "error" in result.json():
                raise ValueError(result.json()["error"]["info"])  # noqa: TRY301
            self.datadata = result.json()["quotes"]
            _LOGGER.debug("Currencylayer data updated: %s", result.json()["timestamp"])
        except ValueError as err:
            _LOGGER.error("Check Currencylayer API %s", err.args)
            self.datadata = None