HEX
Server: Apache/2.4.63 (Unix)
System: Linux Synopilou92 4.4.302+ #72806 SMP Mon Jul 21 23:16:00 CST 2025 x86_64
User: pilou92 (1026)
PHP: 8.0.30
Disabled: NONE
Upload Files
File: /volume1/docker/homeassistant2025/custom_components/rtetempo/calendar.py
"""RTE Tempo Calendar."""
from __future__ import annotations

import asyncio
import datetime
import logging

from homeassistant.components.calendar import CalendarEntity, CalendarEvent
from homeassistant.config_entries import ConfigEntry
from homeassistant.core import HomeAssistant
from homeassistant.helpers.device_registry import DeviceEntryType
from homeassistant.helpers.entity import DeviceInfo
from homeassistant.helpers.entity_platform import AddEntitiesCallback

from .api_worker import APIWorker, TempoDay
from .const import (
    API_ATTRIBUTION,
    API_REQ_TIMEOUT,
    API_VALUE_BLUE,
    API_VALUE_RED,
    API_VALUE_WHITE,
    DEVICE_MANUFACTURER,
    DEVICE_MODEL,
    DEVICE_NAME,
    DOMAIN,
    FRANCE_TZ,
    SENSOR_COLOR_BLUE_EMOJI,
    SENSOR_COLOR_BLUE_NAME,
    SENSOR_COLOR_RED_EMOJI,
    SENSOR_COLOR_RED_NAME,
    SENSOR_COLOR_UNKNOWN_EMOJI,
    SENSOR_COLOR_WHITE_EMOJI,
    SENSOR_COLOR_WHITE_NAME,
)

_LOGGER = logging.getLogger(__name__)


# config flow setup
async def async_setup_entry(
    hass: HomeAssistant,
    config_entry: ConfigEntry,
    async_add_entities: AddEntitiesCallback,
) -> None:
    """Set up entry."""
    _LOGGER.debug("%s: setting up calendar plateform", config_entry.title)
    # Retrieve the API Worker object
    try:
        api_worker = hass.data[DOMAIN][config_entry.entry_id]
    except KeyError:
        _LOGGER.error(
            "%s: can not calendar: failed to get the API worker object",
            config_entry.title,
        )
        return
    # Wait request timeout to let API worker get first batch of data before initializing calendar
    await asyncio.sleep(API_REQ_TIMEOUT)
    # Init sensors
    async_add_entities(
        [TempoCalendar(api_worker, config_entry.entry_id)],
        True,
    )


class TempoCalendar(CalendarEntity):
    """Create a Home Assistant calendar returning tempo days."""

    # Generic entity properties
    _attr_has_entity_name = True
    _attr_attribution = API_ATTRIBUTION

    def __init__(self, api_worker: APIWorker, config_id) -> None:
        """Initialize the calendar."""
        # Generic entity properties
        self._attr_name = "Calendrier"
        self._attr_unique_id = f"{DOMAIN}_{config_id}_calendar"
        # TempoCalendar properties
        self._api_worker = api_worker
        self._config_id = config_id
        super().__init__()

    async def async_get_events(
        self,
        hass: HomeAssistant,
        start_date: datetime.datetime,
        end_date: datetime.datetime,
    ) -> list[CalendarEvent]:
        """Return calendar events within a datetime range."""
        tempo_days = self._api_worker.get_calendar_days()
        events: list[CalendarEvent] = []
        if self._api_worker.adjusted_days:
            # we are dealing with datetimes
            for tempo_day in tempo_days:
                if tempo_day.Start >= start_date and tempo_day.End <= end_date:
                    events.append(forge_calendar_event(tempo_day))
                elif tempo_day.Start < start_date < tempo_day.End < end_date:
                    events.append(forge_calendar_event(tempo_day))
                elif start_date < tempo_day.Start < end_date < tempo_day.End:
                    events.append(forge_calendar_event(tempo_day))
        else:
            # we are dealing with dates (all day events)
            for tempo_day in tempo_days:
                if (
                    tempo_day.Start >= start_date.date()
                    and tempo_day.End <= end_date.date()
                ):
                    events.append(forge_calendar_event(tempo_day))
                elif (
                    tempo_day.Start
                    <= start_date.date()
                    <= tempo_day.End
                    <= end_date.date()
                ):
                    events.append(forge_calendar_event(tempo_day))
                elif (
                    start_date.date()
                    <= tempo_day.Start
                    <= end_date.date()
                    <= tempo_day.End
                ):
                    events.append(forge_calendar_event(tempo_day))
        _LOGGER.debug(
            "Returning %d events (on %d available) for range %s <> %s",
            len(events),
            len(tempo_days),
            start_date,
            end_date,
        )
        return events

    @property
    def device_info(self) -> DeviceInfo:
        """Return the device info."""
        return DeviceInfo(
            entry_type=DeviceEntryType.SERVICE,
            identifiers={(DOMAIN, self._config_id)},
            name=DEVICE_NAME,
            manufacturer=DEVICE_MANUFACTURER,
            model=DEVICE_MODEL,
        )

    @property
    def event(self) -> CalendarEvent | None:
        """Return the current active event if any."""
        localized_now = datetime.datetime.now(FRANCE_TZ)
        if self._api_worker.adjusted_days:
            # we are dealing with datetimes
            for tempo_day in self._api_worker.get_calendar_days():
                if tempo_day.Start <= localized_now < tempo_day.End:
                    return forge_calendar_event(tempo_day)
        else:
            # we are dealing with dates (all day events)
            for tempo_day in self._api_worker.get_calendar_days():
                if tempo_day.Start <= localized_now.date() < tempo_day.End:
                    return forge_calendar_event(tempo_day)
        return None


def forge_calendar_event(tempo_day: TempoDay):
    """Forge a Home Assistant Calendar Event from a Tempo day."""
    return CalendarEvent(
        start=tempo_day.Start,
        end=tempo_day.End,
        summary=get_value_emoji(tempo_day.Value),
        description=forge_calendar_event_description(tempo_day),
        location="France",
        uid=f"{DOMAIN}_{tempo_day.Start.year}_{tempo_day.Start.month}_{tempo_day.Start.day}",
    )


def get_value_emoji(value) -> str:
    """Get corresponding emoji for tempo value."""
    if value == API_VALUE_RED:
        return SENSOR_COLOR_RED_EMOJI
    if value == API_VALUE_WHITE:
        return SENSOR_COLOR_WHITE_EMOJI
    if value == API_VALUE_BLUE:
        return SENSOR_COLOR_BLUE_EMOJI
    return SENSOR_COLOR_UNKNOWN_EMOJI


def forge_calendar_event_description(tempo_day: TempoDay) -> str:
    """Forge a calendar event summary from a tempo day value."""
    if tempo_day.Value == API_VALUE_RED:
        return f"Jour Tempo {SENSOR_COLOR_RED_NAME}"
    if tempo_day.Value == API_VALUE_WHITE:
        return f"Jour Tempo {SENSOR_COLOR_WHITE_NAME}"
    if tempo_day.Value == API_VALUE_BLUE:
        return f"Jour Tempo {SENSOR_COLOR_BLUE_NAME}"
    return f"Jour Tempo inconnu ({tempo_day.Value})"