127 lines
4.0 KiB
Python
127 lines
4.0 KiB
Python
"""Entity representing a Jewish Calendar sensor."""
|
|
|
|
from abc import abstractmethod
|
|
from dataclasses import dataclass
|
|
import datetime as dt
|
|
import logging
|
|
|
|
from hdate import HDateInfo, Location, Zmanim
|
|
from hdate.translator import Language, set_language
|
|
|
|
from homeassistant.config_entries import ConfigEntry
|
|
from homeassistant.core import CALLBACK_TYPE, callback
|
|
from homeassistant.helpers import event
|
|
from homeassistant.helpers.device_registry import DeviceEntryType, DeviceInfo
|
|
from homeassistant.helpers.entity import Entity, EntityDescription
|
|
from homeassistant.util import dt as dt_util
|
|
|
|
from .const import DOMAIN
|
|
|
|
_LOGGER = logging.getLogger(__name__)
|
|
|
|
type JewishCalendarConfigEntry = ConfigEntry[JewishCalendarData]
|
|
|
|
|
|
@dataclass
|
|
class JewishCalendarDataResults:
|
|
"""Jewish Calendar results dataclass."""
|
|
|
|
dateinfo: HDateInfo
|
|
zmanim: Zmanim
|
|
|
|
|
|
@dataclass
|
|
class JewishCalendarData:
|
|
"""Jewish Calendar runtime dataclass."""
|
|
|
|
language: Language
|
|
diaspora: bool
|
|
location: Location
|
|
candle_lighting_offset: int
|
|
havdalah_offset: int
|
|
results: JewishCalendarDataResults | None = None
|
|
|
|
|
|
class JewishCalendarEntity(Entity):
|
|
"""An HA implementation for Jewish Calendar entity."""
|
|
|
|
_attr_has_entity_name = True
|
|
_attr_should_poll = False
|
|
_update_unsub: CALLBACK_TYPE | None = None
|
|
|
|
def __init__(
|
|
self,
|
|
config_entry: JewishCalendarConfigEntry,
|
|
description: EntityDescription,
|
|
) -> None:
|
|
"""Initialize a Jewish Calendar entity."""
|
|
self.entity_description = description
|
|
self._attr_unique_id = f"{config_entry.entry_id}-{description.key}"
|
|
self._attr_device_info = DeviceInfo(
|
|
entry_type=DeviceEntryType.SERVICE,
|
|
identifiers={(DOMAIN, config_entry.entry_id)},
|
|
)
|
|
self.data = config_entry.runtime_data
|
|
set_language(self.data.language)
|
|
|
|
def make_zmanim(self, date: dt.date) -> Zmanim:
|
|
"""Create a Zmanim object."""
|
|
return Zmanim(
|
|
date=date,
|
|
location=self.data.location,
|
|
candle_lighting_offset=self.data.candle_lighting_offset,
|
|
havdalah_offset=self.data.havdalah_offset,
|
|
)
|
|
|
|
async def async_added_to_hass(self) -> None:
|
|
"""Call when entity is added to hass."""
|
|
await super().async_added_to_hass()
|
|
self._schedule_update()
|
|
|
|
async def async_will_remove_from_hass(self) -> None:
|
|
"""Run when entity will be removed from hass."""
|
|
if self._update_unsub:
|
|
self._update_unsub()
|
|
self._update_unsub = None
|
|
return await super().async_will_remove_from_hass()
|
|
|
|
@abstractmethod
|
|
def _update_times(self, zmanim: Zmanim) -> list[dt.datetime | None]:
|
|
"""Return a list of times to update the sensor."""
|
|
|
|
def _schedule_update(self) -> None:
|
|
"""Schedule the next update of the sensor."""
|
|
now = dt_util.now()
|
|
zmanim = self.make_zmanim(now.date())
|
|
update = dt_util.start_of_local_day() + dt.timedelta(days=1)
|
|
|
|
for update_time in self._update_times(zmanim):
|
|
if update_time is not None and now < update_time < update:
|
|
update = update_time
|
|
|
|
if self._update_unsub:
|
|
self._update_unsub()
|
|
self._update_unsub = event.async_track_point_in_time(
|
|
self.hass, self._update, update
|
|
)
|
|
|
|
@callback
|
|
def _update(self, now: dt.datetime | None = None) -> None:
|
|
"""Update the sensor data."""
|
|
self._update_unsub = None
|
|
self._schedule_update()
|
|
self.create_results(now)
|
|
self.async_write_ha_state()
|
|
|
|
def create_results(self, now: dt.datetime | None = None) -> None:
|
|
"""Create the results for the sensor."""
|
|
if now is None:
|
|
now = dt_util.now()
|
|
|
|
_LOGGER.debug("Now: %s Location: %r", now, self.data.location)
|
|
|
|
today = now.date()
|
|
zmanim = self.make_zmanim(today)
|
|
dateinfo = HDateInfo(today, diaspora=self.data.diaspora)
|
|
self.data.results = JewishCalendarDataResults(dateinfo, zmanim)
|