changed default values

Reviewed-on: http://192.168.1.2:3000/brian/bandwidth_probing/pulls/13

.gitea/workflows/default.yaml

@@ -23,6 +23,8 @@ jobs:
           pip install flake8 mypy pytest
           if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
       - name: PEP8 check
-        run: flake8 ./code --benchmark --exit-zero
+        run: flake8 ./code --benchmark
       - name: Type check
-        run: mypy ./code
+        run: |
+          python3 -m pip install types-python-dateutil
+          mypy ./code

.gitea/workflows/release.yaml

@@ -22,9 +22,11 @@ jobs:
           pip install flake8 mypy pytest
           if [ -f requirements.txt ]; then pip install -r requirements.txt; fi
       - name: PEP8 check
-        run: flake8 ./code --benchmark --exit-zero
+        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

README.md

@@ -1,3 +1,21 @@
 # bandwidth_probing
 
 Service that runs in a container and measures network bandwidth at a specified interval. The results are sent to a MongoDB database. Logs are sent to discord channel.
+
+## Environment Variables
+
+```python
+LOGGER_LEVEL: Literal['debug', 'info', 'warning', 'error', 'critical'] = 'debug'
+USER_ID: int = 0 # id number 0 is reserved for testing
+REPLICATES: int = 3 # number of replicates to be measured per event
+TRIGGER_INTERVAL_SECONDS: float = 3600 # seconds between events
+DATA_QUEUE_FLUSH_INTERVAL_SECONDS: float = 600 # seconds between attempts to push data stuck in data queue
+
+DB_IP_ADDRESS: str = '192.168.1.2'
+DB_NAME: str = 'bandwidth_probing'
+DB_COLLECTION_NAME: str = 'data'
+
+DISCORD_SERVICE_NAME: str = 'bandwidth_probing
+DISCORD_WEBHOOK_URL: str = 'https://discord.com/api/webhooks/1127970367047225354/wNkoRex4OncMw11OCJg6atR_xgHS2VhLrku4jYIUw81Kr4sutdD3Tt-XAltG0UQy14rZ'
+DISCORD_LOGGER_LEVEL: Literal['debug', 'info', 'warning', 'error', 'critical'] = 'warning'
+```

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/database.py

@@ -3,6 +3,7 @@ from dotenv import load_dotenv
 from configparser import ConfigParser
 from pathlib import Path
 import logging
+import os
 
 
 # load default values
@@ -21,6 +22,20 @@ def connect(
     """
     Connect to MongoDB database and return collection.
     """
+    # load environment variables
+    load_dotenv()
+    ip_addr = os.getenv(
+        'DB_IP_ADDRESS',
+        default=DB_IP_ADDRESS
+    )
+    db_name = os.getenv(
+        'DB_NAME',
+        default=DB_NAME
+    )
+    collection_name = os.getenv(
+        'DB_COLLECTION_NAME',
+        default=DB_COLLECTION_NAME
+    )
     # connect to database
     client: MongoClient = MongoClient(ip_addr)
     db = client[db_name]

code/defaults.ini

@@ -1,8 +1,7 @@
 [main]
-logger_level=info
+logger_level=debug
 user_id=0
 replicates=3
-trigger_interval_seconds=60
 
 [database]
 db_ip_address=192.168.1.2
@@ -13,3 +12,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=05 * * * *
+tz=Europe/Paris

code/initialise_app.py

@@ -20,13 +20,18 @@ def initialise_app(
     discord_service_name: str = DISCORD_SERVICE_NAME,
     discord_webhook_url: str = DISCORD_WEBHOOK_URL,
     discord_logger_level: str = DISCORD_LOGGER_LEVEL
-):
+) -> None:
     """
     Convienience function that ensures eveything is ready
     before running the main loop.
     """
     # load environment variables
     load_dotenv()
+    assert (
+        'USER_ID' in os.environ
+    ), (
+        'environment variable USER_ID must be specified'
+    )
     logger_level = os.getenv(
         'LOGGER_LEVEL',
         default=LOGGER_LEVEL
@@ -58,6 +63,7 @@ def initialise_app(
     level = getattr(logging, discord_logger_level.upper())
     discord_handler.setLevel(level=level)
     logger.addHandler(discord_handler)
+    logging.debug('finished')
 
 
 if __name__ == '__main__':

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
 
@@ -13,26 +14,55 @@ from database import connect
 # load default values
 config = ConfigParser()
 config.read(Path(__file__).parent / 'defaults.ini')
-REPLICATES = int(config.get('main', 'replicates'))
 USER_ID = int(config.get('main', 'user_id'))
-TRIGGER_INTERVAL_SECONDS = float(
+REPLICATES = int(config.get('main', 'replicates'))
-    config.get(
+SCHEDULE = config.get('cron', 'schedule')
-        'main',
+TZ = config.get('cron', 'tz')
-        'trigger_interval_seconds'
+
-    )
+
+def flush_data(
+    data_queue: Queue,
+    user_id: int
+) -> None:
+    # connect to database
+    db = connect()
+    # 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()
+        # prepare payload
+        payload_dict = {
+            'user_id': user_id,
+            'data': data
+        }
+        # send to database
+        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.debug(
+                'failed pushing to database:\n'
+                f'{json.dumps(payload_dict, indent=4)}\n'
+                f'with error: {e}'
             )
+            continue
+        else:
+            logging.debug(f'data sent to database received id: {db_id}')
+    logging.debug('finished')
 
 
 def event(
-    replicates: int = REPLICATES,
+    data_queue: Queue,
-    user_id: int = USER_ID
+    user_id: int,
-):
+    replicates: int = REPLICATES
-    logging.debug('started event')
+) -> None:
-    # load environment variables
-    replicates = int(os.getenv('REPLICATES', default=REPLICATES))
-    user_id = int(os.getenv('USER_ID', default=USER_ID))
-    # setup data queue
-    data_queue: Queue = Queue()
     # run event
     for rep_num in range(replicates):
         logging.debug(f'running replicate {rep_num+1} of {replicates}')
@@ -46,58 +76,57 @@ def event(
         data_queue.put(data)
         # polite pause
         time.sleep(1)
-    # check if queue should be flushed
+    # flush queue
-    queue_size = data_queue.qsize()
+    flush_data(
-    if queue_size == 0:
+        data_queue=data_queue,
-        logging.debug('queue empty, continuing')
+        user_id=user_id
-        return
-    # flush queue to database
-    db = connect()
-    for i in range(queue_size):
-        # get data
-        data = data_queue.get()
-        # prepare payload
-        payload_dict = {
-            'user_id': user_id,
-            'data': data
-        }
-        # send to database
-        try:
-            db_id = db.insert_one(payload_dict).inserted_id
-        except Exception as e:
-            logging.error('failed sending data to database')
-            logging.info(
-                '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}')
     logging.debug('finished')
 
 
 if __name__ == '__main__':
+    # setup
     initialise_app()
-    trigger_time = time.time()
+    data_queue: Queue = Queue()
-    while True:
+    user_id = int(os.getenv('USER_ID', default=USER_ID))
-        if time.time() >= trigger_time:
+    replicates = int(os.getenv('REPLICATES', default=REPLICATES))
-            logging.debug('triggered event')
+    schedule = os.getenv(
-            # set new trigger time
+        key='SCHEDULE',
-            trigger_interval = float(
+        default=SCHEDULE
-                os.getenv(
-                    'TRIGGER_INTERVAL_SECONDS',
-                    default=TRIGGER_INTERVAL_SECONDS
     )
+    tz = os.getenv(
+        key='TZ',
+        default=TZ
     )
-            trigger_time += trigger_interval
+
+    @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:
-                event()
+                data = measure_bandwidth()
             except Exception as e:
-                logging.error(e)
+                logging.error(f'failed to measure bandwidth: {e}')
-            else:
+                continue
-                logging.info('finished event')
+            # 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
+    main_loop(
+        data_queue=data_queue,
+        user_id=user_id,
+        replicates=replicates
+    )

mypy.ini

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