145 lines
4.3 KiB
Python
145 lines
4.3 KiB
Python
|
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)
|