hat.util.cron

  1import datetime
  2import typing
  3
  4
  5class AllSubExpr(typing.NamedTuple):
  6    pass
  7
  8
  9class ValueSubExpr(typing.NamedTuple):
 10    value: int
 11
 12
 13class RangeSubExpr(typing.NamedTuple):
 14    from_: ValueSubExpr
 15    to: ValueSubExpr
 16
 17
 18class ListSubExpr(typing.NamedTuple):
 19    subexprs: list[ValueSubExpr]
 20
 21
 22SubExpr: typing.TypeAlias = (AllSubExpr |
 23                             ValueSubExpr |
 24                             RangeSubExpr |
 25                             ListSubExpr)
 26
 27
 28class Expr(typing.NamedTuple):
 29    minute: SubExpr
 30    hour: SubExpr
 31    day: SubExpr
 32    month: SubExpr
 33    day_of_week: SubExpr
 34
 35
 36def parse(expr_str: str) -> Expr:
 37    return Expr(*(_parse_subexpr(i) for i in expr_str.split(' ')))
 38
 39
 40def next(expr: Expr,
 41         t: datetime.datetime
 42         ) -> datetime.datetime:
 43    t = t.replace(second=0, microsecond=0)
 44
 45    while True:
 46        t = t + datetime.timedelta(minutes=1)
 47
 48        if match(expr, t):
 49            return t
 50
 51
 52def match(expr: Expr,
 53          t: datetime.datetime
 54          ) -> bool:
 55    if t.second or t.microsecond:
 56        return False
 57
 58    if not _match_subexpr(expr.minute, t.minute):
 59        return False
 60
 61    if not _match_subexpr(expr.hour, t.hour):
 62        return False
 63
 64    if not _match_subexpr(expr.day, t.day):
 65        return False
 66
 67    if not _match_subexpr(expr.month, t.month):
 68        return False
 69
 70    if not _match_subexpr(expr.day_of_week, t.isoweekday() % 7):
 71        return False
 72
 73    return True
 74
 75
 76def _parse_subexpr(subexpr_str):
 77    if subexpr_str == '*':
 78        return AllSubExpr()
 79
 80    if '-' in subexpr_str:
 81        from_str, to_str = subexpr_str.split('-')
 82        return RangeSubExpr(int(from_str), int(to_str))
 83
 84    if ',' in subexpr_str:
 85        return ListSubExpr([int(i) for i in subexpr_str.split(',')])
 86
 87    return ValueSubExpr(int(subexpr_str))
 88
 89
 90def _match_subexpr(subexpr, value):
 91    if isinstance(subexpr, AllSubExpr):
 92        return True
 93
 94    if isinstance(subexpr, ValueSubExpr):
 95        return value == subexpr.value
 96
 97    if isinstance(subexpr, RangeSubExpr):
 98        return subexpr.from_ <= value <= subexpr.to
 99
100    if isinstance(subexpr, ListSubExpr):
101        return value in subexpr.subexprs
102
103    raise ValueError('unsupported subexpression')