sensor.py

Go to the documentation of this file.

"""Support for showing the date and the time."""
 
from __future__ import annotations
 
from collections.abc import Callable, Mapping
from datetime import datetime, timedelta
import logging
from typing import Any
 
import voluptuous as vol
 
from homeassistant.components.sensor import (
    ENTITY_ID_FORMAT,
    PLATFORM_SCHEMA as SENSOR_PLATFORM_SCHEMA,
    SensorEntity,
)
from homeassistant.config_entries import ConfigEntry
from homeassistant.const import CONF_DISPLAY_OPTIONS, EVENT_CORE_CONFIG_UPDATE
from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.entity_platform import AddEntitiesCallback
from homeassistant.helpers.event import async_track_point_in_utc_time
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
import homeassistant.util.dt as dt_util
 
from .const import OPTION_TYPES
 
_LOGGER = logging.getLogger(__name__)
 
TIME_STR_FORMAT = "%H:%M"
 
 
PLATFORM_SCHEMA = SENSOR_PLATFORM_SCHEMA.extend(
    {
        vol.Optional(CONF_DISPLAY_OPTIONS, default=["time"]): vol.All(
            cv.ensure_list, [vol.In(OPTION_TYPES)]
        )
    }
)
 
 
async def async_setup_platform(
    hass: HomeAssistant,
    config: ConfigType,
    async_add_entities: AddEntitiesCallback,
    discovery_info: DiscoveryInfoType | None = None,
) -> None:
    """Set up the Time and Date sensor."""
    if hass.config.time_zone is None:
        _LOGGER.error("Timezone is not set in Home Assistant configuration")  # type: ignore[unreachable]
        return
 
    async_add_entities(
        [TimeDateSensor(variable) for variable in config[CONF_DISPLAY_OPTIONS]]
    )
 
 
async def async_setup_entry(
    hass: HomeAssistant, entry: ConfigEntry, async_add_entities: AddEntitiesCallback
) -> None:
    """Set up the Time & Date sensor."""
 
    async_add_entities(
        [TimeDateSensor(entry.options[CONF_DISPLAY_OPTIONS], entry.entry_id)]
    )
 
 
class TimeDateSensor(SensorEntity):
    """Implementation of a Time and Date sensor."""
 
    _attr_should_poll = False
    _attr_has_entity_name = True
    _state: str | None = None
    unsub: CALLBACK_TYPE | None = None
 
    def __init__(self, option_type: str, entry_id: str | None = None) -> None:
        """Initialize the sensor."""
        self._attr_translation_key_attr_translation_key = option_type
        self.typetype = option_type
        self.entity_identity_identity_id = ENTITY_ID_FORMAT.format(option_type)
        self._attr_unique_id_attr_unique_id = option_type if entry_id else None
 
        self._update_internal_state_update_internal_state(dt_util.utcnow())
 
    @property
    def native_value(self) -> str | None:
        """Return the state of the sensor."""
        return self._state_state
 
    @property
    def icon(self) -> str:
        """Icon to use in the frontend, if any."""
        if "date" in self.typetype and "time" in self.typetype:
            return "mdi:calendar-clock"
        if "date" in self.typetype:
            return "mdi:calendar"
        return "mdi:clock"
 
    @callback
    def async_start_preview(
        self,
        preview_callback: Callable[[str, Mapping[str, Any]], None],
    ) -> CALLBACK_TYPE:
        """Render a preview."""
 
        @callback
        def point_in_time_listener(time_date: datetime | None) -> None:
            """Update preview."""
 
            now = dt_util.utcnow()
            self._update_internal_state_update_internal_state(now)
            self.unsubunsub = async_track_point_in_utc_time(
                self.hasshass, point_in_time_listener, self.get_next_intervalget_next_interval(now)
            )
            calculated_state = self._async_calculate_state_async_calculate_state()
            preview_callback(calculated_state.state, calculated_state.attributes)
 
        @callback
        def async_stop_preview() -> None:
            """Stop preview."""
            if self.unsubunsub:
                self.unsubunsub()
                self.unsubunsub = None
 
        point_in_time_listener(None)
        return async_stop_preview
 
    async def async_added_to_hass(self) -> None:
        """Set up first update."""
 
        async def async_update_config(event: Event) -> None:
            """Handle core config update."""
            self._update_state_and_setup_listener_update_state_and_setup_listener()
            self.async_write_ha_stateasync_write_ha_state()
 
        self.async_on_removeasync_on_remove(
            self.hasshass.bus.async_listen(EVENT_CORE_CONFIG_UPDATE, async_update_config)
        )
        self._update_state_and_setup_listener_update_state_and_setup_listener()
 
    async def async_will_remove_from_hass(self) -> None:
        """Cancel next update."""
        if self.unsubunsub:
            self.unsubunsub()
            self.unsubunsub = None
 
    def get_next_interval(self, time_date: datetime) -> datetime:
        """Compute next time an update should occur."""
        if self.typetype == "date":
            tomorrow = dt_util.as_local(time_date) + timedelta(days=1)
            return dt_util.start_of_local_day(tomorrow)
 
        timestamp = dt_util.as_timestamp(time_date)
        interval = 60
 
        delta = interval - (timestamp % interval)
        next_interval = time_date + timedelta(seconds=delta)
        _LOGGER.debug("%s + %s -> %s (%s)", time_date, delta, next_interval, self.typetype)
 
        return next_interval
 
    def _update_internal_state(self, time_date: datetime) -> None:
        time = dt_util.as_local(time_date).strftime(TIME_STR_FORMAT)
        time_utc = time_date.strftime(TIME_STR_FORMAT)
        date = dt_util.as_local(time_date).date().isoformat()
        date_utc = time_date.date().isoformat()
 
        if self.typetype == "time":
            self._state_state = time
        elif self.typetype == "date":
            self._state_state = date
        elif self.typetype == "date_time":
            self._state_state = f"{date}, {time}"
        elif self.typetype == "date_time_utc":
            self._state_state = f"{date_utc}, {time_utc}"
        elif self.typetype == "time_date":
            self._state_state = f"{time}, {date}"
        elif self.typetype == "time_utc":
            self._state_state = time_utc
        elif self.typetype == "date_time_iso":
            self._state_state = dt_util.parse_datetime(
                f"{date} {time}", raise_on_error=True
            ).isoformat()
 
    def _update_state_and_setup_listener(self) -> None:
        """Update state and setup listener for next interval."""
        now = dt_util.utcnow()
        self._update_internal_state_update_internal_state(now)
        self.unsubunsub = async_track_point_in_utc_time(
            self.hasshass, self.point_in_time_listenerpoint_in_time_listener, self.get_next_intervalget_next_interval(now)
        )
 
    @callback
    def point_in_time_listener(self, time_date: datetime) -> None:
        """Get the latest data and update state."""
        self._update_state_and_setup_listener_update_state_and_setup_listener()
        self.async_write_ha_stateasync_write_ha_state()