Merge branch '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/timeform_cron.py

@@ -0,0 +1,39 @@
+import logging
+
+from flask_wtf import Form
+from wtforms import StringField, validators, SubmitField, ValidationError
+from croniter import croniter
+
+log = logging.getLogger(__name__)
+
+
+class CronAlarmForm(Form):
+    nick = StringField(
+        "Alarm nick",
+        validators=[validators.required()],
+        description="A simple name to recognize this alarm",
+    )
+    cron_format = StringField(
+        "cron-like format",
+        validators=[validators.required()],
+        description="the frequency specification, as in the `cron` command",
+    )
+    submit = SubmitField("Submit")
+
+    def populate_from_timespec(self, timespec):
+        if "nick" in timespec:
+            self.nick.data = timespec["nick"]
+        if "cron_format" in timespec:
+            self.cron_format.data = timespec["cron_format"]
+
+    def validate_cron_format(self, field):
+        if not croniter.is_valid(field.data):
+            raise ValidationError("formato di cron non valido")
+
+
+def cronalarm_receive(form):
+    return {
+        "kind": "cron",
+        "nick": form.nick.data,
+        "cron_format": form.cron_format.data,
+    }

larigira/timegen_cron.py

@@ -0,0 +1,30 @@
+import logging
+from datetime import datetime
+
+from croniter import croniter
+
+from .timegen_every import Alarm
+
+log = logging.getLogger("time-cron")
+
+
+class CronAlarm(Alarm):
+
+    description = "Frequency specified by cron-like format. nerds preferred"
+
+    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,14 +74,17 @@ 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',
               'frequency = larigira.timeform_base:FrequencyAlarmForm',
+              'cron = larigira.timeform_cron:CronAlarmForm',
           ],
           'larigira.timeform_receive': [
               'single = larigira.timeform_base:singlealarm_receive',
               'frequency = larigira.timeform_base:frequencyalarm_receive',
+              'cron = larigira.timeform_cron:cronalarm_receive',
           ],
           'larigira.audioform_create': [
               'static = larigira.audioform_static:StaticAudioForm',