import fcntl import json from datetime import datetime, timedelta from pathlib import Path MINS = "minutes" class Span: def __init__(self, start, end): self.start = min(start, end) self.end = max(start, end) def __eq__(self, other): return (self.start, self.end) == (other.start, other.end) def __lt__(self, other): return (self.start, self.end) < (other.start, other.end) class MinuteSpan(Span): @classmethod def from_args(cls, **kwargs): start = kwargs["start"].split(":") start = int(start[0]) * 60 + int(start[1]) end = kwargs["end"].split(":") end = int(end[0]) * 60 + int(end[1]) if end == 0: # To 00:00 should mean until midnight (24:00) end = 24 * 60 return cls(start, end) def to_dict(self): return { "start": f"{self.start // 60}:{self.start % 60:02}", "end": f"{self.end // 60}:{self.end % 60:02}", } class DateTimeSpan(Span): def __init__(self, start, end, action): super().__init__(start, end) self.action = action def __eq__(self, other): return (self.start, self.end, self.action) == (other.start, other.end, other.action) @classmethod def from_args(cls, **kwargs): return cls( datetime.fromisoformat(kwargs["start"]), datetime.fromisoformat(kwargs["end"]), kwargs["action"], ) @classmethod def from_minutespan(cls, date, minutespan): start = minutespan.start end = minutespan.end add_days = 0 if end == 24 * 60: # Ending midnight needs to be set as 0:00 of the next day end = 0 add_days = 1 return cls( date.replace(hour=start // 60, minute=start % 60, second=0, microsecond=0), date.replace(hour=end // 60, minute=end % 60, second=0, microsecond=0) + timedelta(days=add_days), True, ) def to_dict(self): return { "start": self.start.isoformat(timespec=MINS), "end": self.end.isoformat(timespec=MINS), "action": self.action, } class Status: def __init__(self, action, until=None, then_until=None): self.action = action self.until = until self.then_until = then_until @classmethod def from_args(cls, **kwargs): until = kwargs.get("until") then_until = kwargs.get("then_until") return cls( kwargs["action"], datetime.fromisoformat(until) if until else None, datetime.fromisoformat(then_until) if then_until else None, ) def to_dict(self): return { "action": self.action, "until": self.until.isoformat(timespec=MINS) if self.until else None, "then_until": self.then_until.isoformat(timespec=MINS) if self.then_until else None, } class DataFile: def __init__(self, path): self._path = Path(path) self._file = None self.status = None self.schedules = None self.overrides = None def __enter__(self): self._file = self._path.open("r+", encoding="utf-8") fcntl.flock(self._file, fcntl.LOCK_EX) data = json.load(self._file) self.status = Status.from_args(**data.get("status", {"action": False})) self.schedules = { int(day): [MinuteSpan.from_args(**span) for span in schedule] for day, schedule in data.get("schedules", {day: [] for day in range(7)}).items() } overrides = [DateTimeSpan.from_args(**override) for override in data.get("overrides", [])] # Remove expired overrides now = datetime.now() self.overrides = [o for o in overrides if o.end > now] return self def __exit__(self, exc_type, exc_value, traceback): self._file.close() def to_dict(self): return { "status": self.status.to_dict(), "schedules": { day: [span.to_dict() for span in schedule] for day, schedule in self.schedules.items() }, "overrides": [override.to_dict() for override in self.overrides], } def save(self): self._file.seek(0) self._file.truncate(0) json.dump(self.to_dict(), self._file)