from __future__ import print_function
import logging
log = logging.getLogger("time-every")
from datetime import datetime, timedelta
from pytimeparse.timeparse import timeparse
[docs]def getdate(val):
if type(val) is int:
return datetime.fromtimestamp(val)
return val
[docs]class Alarm(object):
def __init__(self):
pass
[docs] def next_ring(self, current_time=None):
"""if current_time is None, it is now()
returns the next time it will ring; or None if it will not anymore
"""
raise NotImplementedError()
[docs] def has_ring(self, time=None):
"""returns True IFF the alarm will ring exactly at ``time``"""
raise NotImplementedError()
[docs] def all_rings(self, current_time=None):
"""
all future rings
this, of course, is an iterator (they could be infinite)
"""
ring = self.next_ring(current_time)
while ring is not None:
yield ring
ring = self.next_ring(ring)
[docs]class SingleAlarm(Alarm):
"""
rings a single time
"""
description = "Only once, at a specified date and time"
def __init__(self, obj):
super().__init__()
self.dt = getdate(obj["timestamp"])
[docs] def next_ring(self, current_time=None):
"""if current_time is None, it is now()"""
if current_time is None:
current_time = datetime.now()
if current_time >= self.dt:
return None
return self.dt
[docs] def has_ring(self, current_time=None):
if current_time is None:
current_time = datetime.now()
return current_time == self.dt
[docs]class FrequencyAlarm(Alarm):
"""
rings on {t | exists a k integer >= 0 s.t. t = start+k*t, start<t<end}
"""
description = "Events at a specified frequency. Example: every 30minutes"
def __init__(self, obj):
self.start = getdate(obj["start"])
try:
self.interval = int(obj["interval"])
except ValueError:
self.interval = timeparse(obj["interval"])
assert type(self.interval) is int
self.end = getdate(obj["end"]) if "end" in obj else None
self.weekdays = [int(x) for x in obj["weekdays"]] if "weekdays" in obj else None
if self.weekdays is not None:
for weekday in self.weekdays:
if not 1 <= weekday <= 7:
raise ValueError("Not a valid weekday: {}".format(weekday))
[docs] def next_ring(self, current_time=None):
"""if current_time is None, it is now()"""
if current_time is None:
current_time = datetime.now()
if self.end is not None and current_time > self.end:
return None
if current_time < self.start:
return self.start
if self.end is not None:
assert self.start <= current_time <= self.end
else:
assert self.start <= current_time
# this "infinite" loop is required by the weekday exclusion: in
# fact, it is necessary to retry until a valid event/weekday is
# found. a "while True" might have been more elegant (and maybe
# fast), but this gives a clear upper bound to the cycle.
for _ in range(max(60 * 60 * 24 * 7 // self.interval, 1)):
n_interval = (
(current_time - self.start).total_seconds() // self.interval
) + 1
ring = self.start + timedelta(seconds=self.interval * n_interval)
if ring == current_time:
ring += timedelta(seconds=self.interval)
if self.end is not None and ring > self.end:
return None
if self.weekdays is not None and ring.isoweekday() not in self.weekdays:
current_time = ring
continue
return ring
log.warning(
"Can't find a valid time for event %s; " "something went wrong", str(self)
)
return None
[docs] def has_ring(self, current_time=None):
if current_time is None:
current_time = datetime.now()
if not self.start >= current_time >= self.end:
return False
n_interval = (current_time - self.start).total_seconds() // self.interval
expected_time = self.start + timedelta(seconds=self.interval * n_interval)
return expected_time == current_time
def __str__(self):
return "FrequencyAlarm(every %ds)" % self.interval