import logging
from typing import Dict, Iterator
from datetime import datetime, timedelta, time, timezone
_LOGGER = logging.getLogger(__name__)
[docs]
class Scheduler:
def __init__(self, data: Dict, tzinfo: timezone) -> None:
self._timeSlots = []
self._timezone = tzinfo
self._today = datetime.now(self._timezone)
# Monday = 1, Sunday = 7
self._weekday: int = self._today.weekday() + 1
self._hour: int = self._today.hour
self._parseData(data)
[docs]
def getCurrentScreen(self) -> tuple[datetime, str, datetime]:
# Check if the next screen in the list is old
now = datetime.now(self._timezone)
# Make sure the list is up to date
nextTs, _ = self.getNextScreen()
# The last one is the current one
timestamp, name = self._timeSlots[-1]
timestamp = timestamp - timedelta(days=7)
return timestamp, name, nextTs
[docs]
def getNextScreen(self, skip: int = 0) -> tuple[datetime, str]:
now = datetime.now(self._timezone)
while True:
# Get the next screen
timestamp, name = self._timeSlots[0]
if timestamp >= now:
# Found suitable one
break
# Too old, re-add it to the end, one week later
timestamp, name = self._timeSlots.pop(0)
self._timeSlots.append((timestamp + timedelta(days=7), name))
return self._timeSlots[skip]
def _parseData(self, data: Dict) -> None:
for schedule in data:
for slot in self._parseSchedule(schedule):
self._timeSlots.append((slot, schedule.get("screen")))
self._timeSlots.sort(key=lambda el: el[0])
def _parseSchedule(self, schedule: Dict) -> Iterator[datetime]:
hour: int = schedule.get("hour", 0)
hours: list[int] = schedule.get("hours", [])
for day in schedule.get("days", []):
if not hours:
yield self._parseDay(day, hour)
continue
for h in hours:
yield self._parseDay(day, h)
def _parseDay(self, day: int, hour: 0) -> datetime:
dayDiff = day - self._weekday
if dayDiff == 0:
# Day is today. Check if the hour is in the past
if hour <= self._hour:
dayDiff = 7
if dayDiff < 0:
dayDiff += 7
date = self._today.date() + timedelta(days=dayDiff)
return datetime.combine(date, time(hour=hour), tzinfo=self._timezone)