Last active
October 4, 2018 14:51
-
-
Save OttoWinter/b2f050e5fb976771eb05ce00831ff8e4 to your computer and use it in GitHub Desktop.
Cron Next Matcher
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from datetime import datetime, timedelta | |
import pytz | |
tz = pytz.timezone('Europe/Vienna') | |
def _lower_bound(cmp, arr): | |
# Return the first value in arr greater or equal to cmp | |
# Return None if no such value exists | |
# If arr were sorted, this could of course be done in log(n) | |
return next((x for x in arr if x >= cmp), None) | |
# The algorithm: | |
# * Try finding the next second within the minute to match | |
# * If not exist, reset second to first matching second and increase | |
# minute by one | |
# * Try finding the next minute within the hour to match | |
# * If not exist, reset seconds/minutes to first matching values | |
# and increase hour by one | |
# * Same for hour | |
def _next_time(orig_dt): | |
# Seconds for which to trigger, "*" will be [0..59], "/5" will be [0,5,...,55] | |
# "42,43" will be [42,43] | |
# Parsing logic can be found here: https://github.com/OttoWinter/esphomeyaml/blob/master/esphomeyaml/components/time/__init__.py#L122-L248 | |
ALLOWED_SECONDS = [0] | |
# List of minutes for which to trigger | |
ALLOWED_MINUTES = [30] | |
# List of hours for which to trigger | |
ALLOWED_HOURS = [2] | |
# ... | |
# Next time must at least be one second apart | |
dt = orig_dt + timedelta(seconds=1) | |
# Match next second in allowed seconds | |
next_second = _lower_bound(dt.second, ALLOWED_SECONDS) | |
if next_second is None: | |
# If roll-over, set second to first valid one and | |
# increase minute by one because of roll-over | |
next_second = ALLOWED_SECONDS[0] | |
dt += timedelta(minutes=1) | |
next_minute = _lower_bound(dt.minute, ALLOWED_MINUTES) | |
if next_minute != dt.minute: | |
# Not in same minute, reset seconds to first value | |
next_second = ALLOWED_SECONDS[0] | |
if next_minute is None: | |
# If roll-over, set minute to first valid value and | |
# increase hour by one because of roll-over | |
next_minute = ALLOWED_MINUTES[0] | |
dt += timedelta(hours=1) | |
next_hour = _lower_bound(dt.hour, ALLOWED_HOURS) | |
if next_hour != dt.hour: | |
# Not in same hour, reset seconds,minutes to first values | |
next_second = ALLOWED_SECONDS[0] | |
next_minute = ALLOWED_MINUTES[0] | |
if next_hour is None: | |
next_hour = ALLOWED_HOURS[0] | |
dt += timedelta(days=1) | |
# Day of week, day of month are a bit more complicated to | |
# match (but possible), but fortunately Home Assistant doesn't support that | |
# Replace dt attributes by our chosen time values. | |
# Also make the datetime object "naive" by removing tzinfo | |
# so that we can call localize. | |
dt = dt.replace( | |
hour=next_hour, | |
minute=next_minute, | |
second=next_second, | |
tzinfo=None | |
) | |
# Now transform the naive dt back to a localized dt | |
try: | |
dt = tz.localize(dt, is_dst=None) | |
except pytz.AmbiguousTimeError: | |
# If the new time is in an ambiguous phase (math: non-injective) | |
# we need to choose whether to use the one with DST on/off | |
# We "hack" a little by just using the DST on/off from the day before, which should be fine | |
# because DSTs always seem to last at least a month: https://www.timeanddate.com/time/dst/2018.html | |
use_dst = bool((orig_dt - timedelta(days=1)).dst()) | |
dt = tz.localize(dt, is_dst=use_dst) | |
except pytz.NonExistentTimeError: | |
# We're attempting to return a time that will never happen in real life. | |
# For example if the clock jumps from 2:00:00 to 3:00:00 and we're matching on 2:30:00 | |
return _next_time(dt) | |
return dt | |
begin = datetime(year=2018, month=10, day=28, hour=2, minute=5) | |
dt = tz.localize(begin, is_dst=False) | |
local = _next_time(dt) | |
utc = local.astimezone(pytz.UTC) | |
wait = utc - dt.astimezone(pytz.UTC) | |
print(f"Begin is at: {dt}") | |
print(f"Next time is: local={local}, as utc={utc}") | |
print(f"Time to wait: {wait}") | |
# Begin is at: 2018-10-28 02:05:00+01:00 | |
# Next time is: local=2018-10-28 02:30:00+01:00, as utc=2018-10-28 01:30:00+00:00 | |
# Time to wait: 0:25:00 | |
begin = datetime(year=2018, month=3, day=25, hour=1, minute=50) | |
dt = tz.localize(begin, is_dst=True) | |
local = _next_time(dt) | |
utc = local.astimezone(pytz.UTC) | |
wait = utc - dt.astimezone(pytz.UTC) | |
print(f"Begin is at: {dt}, as_utc={dt.astimezone(pytz.UTC)}") | |
print(f"Next time is: local={local}, as utc={utc}") | |
print(f"Time to wait: {wait}") | |
print(utc.astimezone(tz)) | |
# Begin is at: 2018-03-25 01:50:00+01:00, as_utc=2018-03-25 00:50:00+00:00 | |
# Next time is: local=2018-03-26 02:30:00+02:00, as utc=2018-03-26 00:30:00+00:00 | |
# Time to wait: 23:40:00 | |
# 2018-03-26 02:30:00+02:00 |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment