added mypy ignore

.gitea/workflows/default.yaml

@@ -25,4 +25,6 @@ jobs:
       - name: PEP8 check
         run: flake8 ./code --benchmark
       - name: Type check
-        run: mypy ./code
+        run: |
+          python3 -m pip install types-python-dateutil
+          mypy ./code

.gitea/workflows/release.yaml

@@ -24,7 +24,9 @@ jobs:
       - name: PEP8 check
         run: flake8 ./code --benchmark
       - name: Type check
-        run: mypy ./code
+        run: |
+          python3 -m pip install types-python-dateutil
+          mypy ./code
   publish:
     name: Publish
     runs-on: ubuntu-latest

code/cron.py

@@ -0,0 +1,53 @@
+from cron_converter import Cron
+from configparser import ConfigParser
+from pathlib import Path
+from datetime import datetime
+import dateutil
+import time
+import functools
+import logging
+
+# load default values
+config = ConfigParser()
+config.read(Path(__file__).parent / 'defaults.ini')
+SCHEDULE = config.get('cron', 'schedule')
+TZ = config.get('cron', 'tz')
+
+
+def cronjob(
+    schedule: str = SCHEDULE,
+    tz: str = TZ
+):
+    """
+    decorator for running function as a cronjob.
+    """
+    logging.info(f'started cronjob with schedule {schedule} and tz {tz}')
+
+    def decorator_cronjob(func):
+        @functools.wraps(func)
+        def wrapper_cronjob(*args, **kwargs):
+            scheduler = Cron(schedule).schedule(timezone_str=tz)
+            trigger_time = scheduler.next()
+            while True:
+                while trigger_time > datetime.now(tz=dateutil.tz.gettz(tz)):
+                    # waiting loop
+                    time.sleep(1)
+                # actual execution
+                try:
+                    func(*args, **kwargs)
+                except Exception as e:
+                    logging.warning(f'failed executing {func}: {e}')
+                    raise
+                else:
+                    # update trigger time
+                    trigger_time = scheduler.next()
+        return wrapper_cronjob
+    return decorator_cronjob
+
+
+if __name__ == '__main__':
+    @cronjob(schedule='* * * * *', tz=TZ)
+    def test_loop(msg: str):
+        print(msg)
+
+    test_loop('hello world')

code/defaults.ini

@@ -2,8 +2,7 @@
 logger_level=debug
 user_id=0
 replicates=3
-trigger_interval_seconds=3600
+queue_flush_interval_seconds=600
-data_queue_flush_interval_seconds=600
 
 [database]
 db_ip_address=192.168.1.2
@@ -14,3 +13,7 @@ db_collection_name=data
 webhook_url=https://discord.com/api/webhooks/1127970367047225354/wNkoRex4OncMw11OCJg6atR_xgHS2VhLrku4jYIUw81Kr4sutdD3Tt-XAltG0UQy14rZ
 service_name=bandwidth_probing
 logger_level=warning
+
+[cron]
+schedule=* * * * *
+tz=Europe/Paris

code/main.py

@@ -6,6 +6,7 @@ from configparser import ConfigParser
 from pathlib import Path
 from queue import Queue
 from initialise_app import initialise_app
+from cron import cronjob
 from bandwidth import measure as measure_bandwidth
 from database import connect
 
@@ -14,18 +15,14 @@ from database import connect
 config = ConfigParser()
 config.read(Path(__file__).parent / 'defaults.ini')
 REPLICATES = int(config.get('main', 'replicates'))
-TRIGGER_INTERVAL_SECONDS = float(
+QUEUE_FLUSH_INTERVAL_SECONDS = float(
     config.get(
         'main',
-        'trigger_interval_seconds'
+        'queue_flush_interval_seconds'
-    )
-)
-DATA_QUEUE_FLUSH_INTERVAL_SECONDS = float(
-    config.get(
-        'main',
-        'data_queue_flush_interval_seconds'
     )
 )
+SCHEDULE = config.get('cron', 'schedule')
+TZ = config.get('cron', 'tz')
 
 
 def flush_data(
@@ -34,8 +31,13 @@ def flush_data(
 ) -> None:
     # connect to database
     db = connect()
-    # flush queue
+    # stop early
     queue_size = data_queue.qsize()
+    if queue_size == 0:
+        logging.debug('queue already empty')
+        logging.debug('finished')
+        return
+    # flush queue
     for i in range(queue_size):
         # get data
         data = data_queue.get()
@@ -48,13 +50,13 @@ def flush_data(
         try:
             db_id = db.insert_one(payload_dict).inserted_id
         except Exception as e:
+            data_queue.put(data)    # put data back in queue
             logging.error('failed sending data to database')
-            logging.info(
+            logging.debug(
                 'failed pushing to database:\n'
                 f'{json.dumps(payload_dict, indent=4)}\n'
                 f'with error: {e}'
             )
-            data_queue.put(data)    # put data back in queue
             continue
         else:
             logging.debug(f'data sent to database received id: {db_id}')
@@ -80,59 +82,62 @@ def event(
         # polite pause
         time.sleep(1)
     # flush queue
-    if data_queue.qsize() > 0:
+    flush_data(
-        flush_data(
+        data_queue=data_queue,
-            data_queue=data_queue,
+        user_id=user_id
-            user_id=user_id
+    )
-        )
     logging.debug('finished')
 
 
 if __name__ == '__main__':
     # setup
     initialise_app()
+    data_queue: Queue = Queue()
     user_id = int(os.getenv('USER_ID', default=0))
     replicates = int(os.getenv('REPLICATES', default=REPLICATES))
-    trigger_interval = float(
+    queue_flush_interval = float(
         os.getenv(
-            'TRIGGER_INTERVAL_SECONDS',
+            'QUEUE_FLUSH_INTERVAL_SECONDS',
-            default=TRIGGER_INTERVAL_SECONDS
+            default=QUEUE_FLUSH_INTERVAL_SECONDS
         )
     )
-    data_flush_interval = float(
+    schedule = os.getenv(
-        os.getenv(
+        key='SCHEDULE',
-            'DATA_QUEUE_FLUSH_INTERVAL_SECONDS',
+        default=SCHEDULE
-            default=DATA_QUEUE_FLUSH_INTERVAL_SECONDS
-        )
     )
-    data_queue: Queue = Queue()
+    tz = os.getenv(
+        key='TZ',
+        default=TZ
+    )
+
+    @cronjob(schedule=schedule, tz=tz)
+    def main_loop(
+        data_queue: Queue,
+        user_id: int,
+        replicates: int = REPLICATES
+    ) -> None:
+        # run event
+        for rep_num in range(replicates):
+            logging.debug(f'running replicate {rep_num+1} of {replicates}')
+            # do measurement
+            try:
+                data = measure_bandwidth()
+            except Exception as e:
+                logging.error(f'failed to measure bandwidth: {e}')
+                continue
+            # add data to queue
+            data_queue.put(data)
+            # polite pause
+            time.sleep(1)
+        # flush queue
+        flush_data(
+            data_queue=data_queue,
+            user_id=user_id
+        )
+        logging.debug('finished')
     # start main loop
-    trigger_time = time.time()
+    main_loop(
-    data_flush_time = trigger_time + data_flush_interval
+        data_queue=data_queue,
-    while True:
+        user_id=user_id,
-        # measure event
+        replicates=replicates
-        if time.time() >= trigger_time:
+    )
-            logging.debug('triggered event')
-            # set new trigger time
-            trigger_time += trigger_interval
-            # run event
-            event(
-                data_queue=data_queue,
-                user_id=user_id,
-                replicates=replicates
-            )
-        # extra data flush
-        if (
-            time.time() >= data_flush_time and
-            data_queue.qsize() > 0
-        ):
-            logging.debug('triggered extra data flush')
-            # set new trigger time
-            data_flush_time += data_flush_interval
-            # flush data
-            flush_data(
-                data_queue=data_queue,
-                user_id=user_id
-            )
-        # polite pause
-        time.sleep(1)

mypy.ini

@@ -5,3 +5,6 @@ ignore_missing_imports = True
 
 [mypy-speedtest.*]
 ignore_missing_imports = True
+
+[mypy-cron_converter.*]
+ignore_missing_imports = True