add new timegen: cron

larigira/tests/test_time_cron.py

@@ -0,0 +1,28 @@
+import datetime
+
+import pytest
+
+from larigira.timegen_cron import CronAlarm
+
+
+@pytest.fixture
+def a_time():
+    return datetime.datetime(2019, 8, 6, 10, 42, 0)
+
+
+def CA(fmt):
+    return CronAlarm(dict(cron_format=fmt))
+
+
+def test_valid_cron_format():
+    CA("* * * * *")
+
+
+def test_valid_cron_format_six():
+    CA("* * * * * *")
+
+
+def test_never_equal(a_time):
+    c = CA("* * * * *")
+    nt = c.next_ring(a_time)
+    assert nt.minute != a_time.minute

larigira/timegen_cron.py

@@ -0,0 +1,27 @@
+import logging
+from datetime import datetime
+
+from croniter import croniter
+
+from .timegen_every import Alarm
+
+log = logging.getLogger("time-cron")
+
+
+class CronAlarm(Alarm):
+    def __init__(self, obj):
+        super().__init__()
+
+        self.cron_format = obj["cron_format"]
+        if not croniter.is_valid(self.cron_format):
+            raise ValueError("Invalid cron_format: %s" % self.cron_format)
+
+    def next_ring(self, current_time=None):
+        if current_time is None:
+            current_time = datetime.now()
+
+        return croniter(self.cron_format, current_time).get_next(datetime)
+
+    def has_ring(self, current_time=None):
+        # cron specification has no possibility of being over
+        return True

setup.py

@@ -47,6 +47,7 @@ setup(name='larigira',
           'Flask-WTF',
           'flask==0.11',
           'pytimeparse',
+          'croniter==0.3.29',
           'tinydb'
       ],
       tests_require=['pytest-timeout==1.0', 'py>=1.4.29', 'pytest==3.0', ],
@@ -73,6 +74,7 @@ setup(name='larigira',
           'larigira.timegenerators': [
               'frequency = larigira.timegen_every:FrequencyAlarm',
               'single = larigira.timegen_every:SingleAlarm',
+              'cron = larigira.timegen_cron:CronAlarm',
           ],
           'larigira.timeform_create': [
               'single = larigira.timeform_base:SingleAlarmForm',