158 lines
4.8 KiB
Python
158 lines
4.8 KiB
Python
"""Calendar platform for La Marzocco espresso machines."""
|
|
|
|
from collections.abc import Iterator
|
|
from datetime import datetime, timedelta
|
|
|
|
from pylamarzocco.const import WeekDay
|
|
|
|
from homeassistant.components.calendar import CalendarEntity, CalendarEvent
|
|
from homeassistant.core import HomeAssistant
|
|
from homeassistant.helpers.entity_platform import AddConfigEntryEntitiesCallback
|
|
from homeassistant.util import dt as dt_util
|
|
|
|
from .coordinator import LaMarzoccoConfigEntry, LaMarzoccoUpdateCoordinator
|
|
from .entity import LaMarzoccoBaseEntity
|
|
|
|
# Coordinator is used to centralize the data updates
|
|
PARALLEL_UPDATES = 0
|
|
|
|
CALENDAR_KEY = "auto_on_off_schedule"
|
|
|
|
WEEKDAY_TO_ENUM = {
|
|
0: WeekDay.MONDAY,
|
|
1: WeekDay.TUESDAY,
|
|
2: WeekDay.WEDNESDAY,
|
|
3: WeekDay.THURSDAY,
|
|
4: WeekDay.FRIDAY,
|
|
5: WeekDay.SATURDAY,
|
|
6: WeekDay.SUNDAY,
|
|
}
|
|
|
|
|
|
async def async_setup_entry(
|
|
hass: HomeAssistant,
|
|
entry: LaMarzoccoConfigEntry,
|
|
async_add_entities: AddConfigEntryEntitiesCallback,
|
|
) -> None:
|
|
"""Set up switch entities and services."""
|
|
|
|
coordinator = entry.runtime_data.schedule_coordinator
|
|
|
|
async_add_entities(
|
|
LaMarzoccoCalendarEntity(coordinator, CALENDAR_KEY, schedule.identifier)
|
|
for schedule in coordinator.device.schedule.smart_wake_up_sleep.schedules
|
|
if schedule.identifier
|
|
)
|
|
|
|
|
|
class LaMarzoccoCalendarEntity(LaMarzoccoBaseEntity, CalendarEntity):
|
|
"""Class representing a La Marzocco calendar."""
|
|
|
|
_attr_translation_key = CALENDAR_KEY
|
|
|
|
def __init__(
|
|
self,
|
|
coordinator: LaMarzoccoUpdateCoordinator,
|
|
key: str,
|
|
identifier: str,
|
|
) -> None:
|
|
"""Set up calendar."""
|
|
super().__init__(coordinator, f"{key}_{identifier}")
|
|
self._identifier = identifier
|
|
self._attr_translation_placeholders = {"id": identifier}
|
|
|
|
@property
|
|
def event(self) -> CalendarEvent | None:
|
|
"""Return the next upcoming event."""
|
|
now = dt_util.now()
|
|
|
|
events = self._get_events(
|
|
start_date=now,
|
|
end_date=now + timedelta(days=7), # only need to check a week ahead
|
|
)
|
|
return next(iter(events), None)
|
|
|
|
async def async_get_events(
|
|
self,
|
|
hass: HomeAssistant,
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
) -> list[CalendarEvent]:
|
|
"""Return calendar events within a datetime range."""
|
|
|
|
return self._get_events(
|
|
start_date=start_date,
|
|
end_date=end_date,
|
|
)
|
|
|
|
def _get_events(
|
|
self,
|
|
start_date: datetime,
|
|
end_date: datetime,
|
|
) -> list[CalendarEvent]:
|
|
"""Get calendar events within a datetime range."""
|
|
|
|
events: list[CalendarEvent] = []
|
|
for date in self._get_date_range(start_date, end_date):
|
|
if scheduled := self._async_get_calendar_event(date):
|
|
if scheduled.end < start_date:
|
|
continue
|
|
if scheduled.start > end_date:
|
|
continue
|
|
events.append(scheduled)
|
|
return events
|
|
|
|
def _get_date_range(
|
|
self, start_date: datetime, end_date: datetime
|
|
) -> Iterator[datetime]:
|
|
current_date = start_date
|
|
while current_date.date() < end_date.date():
|
|
yield current_date
|
|
current_date += timedelta(days=1)
|
|
yield end_date
|
|
|
|
def _async_get_calendar_event(self, date: datetime) -> CalendarEvent | None:
|
|
"""Return calendar event for a given weekday."""
|
|
|
|
schedule_entry = (
|
|
self.coordinator.device.schedule.smart_wake_up_sleep.schedules_dict[
|
|
self._identifier
|
|
]
|
|
)
|
|
# check first if auto/on off is turned on in general
|
|
if not schedule_entry.enabled:
|
|
return None
|
|
|
|
# parse the schedule for the day
|
|
|
|
if WEEKDAY_TO_ENUM[date.weekday()] not in schedule_entry.days:
|
|
return None
|
|
|
|
hour_on = schedule_entry.on_time_minutes // 60
|
|
minute_on = schedule_entry.on_time_minutes % 60
|
|
hour_off = schedule_entry.off_time_minutes // 60
|
|
minute_off = schedule_entry.off_time_minutes % 60
|
|
|
|
day_offset = 0
|
|
if hour_off == 24:
|
|
# if the machine is scheduled to turn off at midnight, we need to
|
|
# set the end date to the next day
|
|
day_offset = 1
|
|
hour_off = 0
|
|
|
|
end_date = date.replace(
|
|
hour=int(hour_off),
|
|
minute=int(minute_off),
|
|
)
|
|
end_date += timedelta(days=day_offset)
|
|
|
|
return CalendarEvent(
|
|
start=date.replace(
|
|
hour=int(hour_on),
|
|
minute=int(minute_on),
|
|
),
|
|
end=end_date,
|
|
summary=f"Machine {self.coordinator.config_entry.title} on",
|
|
description="Machine is scheduled to turn on at the start time and off at the end time",
|
|
)
|