Cron Next Matcher

cron-match.py

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