- Improve speed testing with using 2 threads, one to saturate connection with speedtest.net and another to get rates from router's API

- Logging fix dual entries

.gitignore

@@ -136,3 +136,4 @@ test.py
 test.png
 data/database.db-shm
 data/database.db-wal
+test2.py

api.py

@@ -0,0 +1,80 @@
+'''
+Main application file
+'''
+
+import eventlet
+eventlet.monkey_patch()
+
+from flask import Flask, jsonify
+from flask_socketio import SocketIO
+
+from datetime import datetime
+import dbo
+
+# region Logger
+import logging
+from debug import setup_logging
+
+log = logger = logging.getLogger("default")
+setup_logging()
+# endregion
+
+from configuration import read_config
+from peewee import fn
+
+config = read_config()
+
+app = Flask(__name__)
+app.config['SECRET_KEY'] = 'BLAAAA_GeneerateMeDynamicallyForBetterSecurity'
+
+socketio = SocketIO(app, async_mode='eventlet')
+
+
+# app.jinja_env.filters['html_line_breaks'] = jinja_filters.html_line_breaks
+# app.jinja_env.filters['zfill'] = jinja_filters.zfill
+
+@app.context_processor
+def inject_global_variables():
+    return dict(
+        config=config,
+        now=datetime.utcnow(),
+        Accounting=dbo.Accounting,
+        MonthlyArchive=dbo.MonthlyArchive,
+        fn=fn
+    )
+
+
+def add_background_task(task, interval):
+    def tsk():
+        while True:
+            try:
+                log.debug(f"Running background task {task.__name__}...")
+                task()
+                log.debug(f"Completed background task {task.__name__}!")
+            except Exception as e:
+                log.error(f"Can't run background task '{task.__name__}': {e}", exc_info=True)
+            socketio.sleep(interval)
+
+    socketio.start_background_task(tsk)
+
+@app.route("/api/")
+def api_index():
+    return jsonify(
+        {""}
+    )
+
+if __name__ == '__main__':
+    config = read_config()
+
+    try:
+        if config['host'] == "0.0.0.0":
+            host = 'localhost'
+        else:
+            host = config['host']
+        log.info(f"Running at http://{host}:{config['port']}")
+        socketio.run(app, debug=False, host=config['host'], port=config['port'])
+    except:
+        print("Unable to start", exc_info=True)
+
+
+

config/config.yaml

@@ -13,6 +13,9 @@ ros_minimum_speed: 300000
 # starts to show high CPU usage at close to 20mbit.
 # Not implemented yet
 ros_maximum_speed: 15000000 # 15 mbit
+ros_wan_interface: lte1
+api_host: 0.0.0.0
+api_port: 1357
 
 
 # Conversion rate 1 mbit : 1000000 bits

config/logger.yaml

@@ -23,9 +23,9 @@ handlers:
 
 root:
     level: ERROR
-    handlers: [debug_file_handler]
+    handlers: [console]
 
 loggers:
   "default":
     level: DEBUG
-    handlers: [debug_file_handler, console]
+    handlers: [debug_file_handler]

paths.py

@@ -9,3 +9,7 @@ import os
 APP_DIR = os.path.dirname(os.path.realpath(__file__))
 CONFIG_DIR = os.path.join(APP_DIR, "config")
 DATA_DIR = os.path.join(APP_DIR, "data")
+
+
+cache1 = os.path.join(APP_DIR, 'cache1.yaml')
+cache2 = os.path.join(APP_DIR, 'cache2.yaml')

speedtester.py

@@ -6,7 +6,9 @@ from debug import setup_logging, catch_errors
 log = logger = logging.getLogger("default")
 setup_logging()
 # endregion
-
+from statistics import median
+from threading import Thread
+import speedtest
 from dbo import Entry
 import routeros_api
 
@@ -16,6 +18,10 @@ secrets = read_config('secrets')
 import time
 
 
+def mbits(bits):
+    return round(bits / 1000000, 2)
+
+
 @catch_errors
 def gather_data():
     log.debug("Gathering data...")
@@ -90,7 +96,7 @@ def ros_dynamic_speed(upload, download):
     :param upload: Upload speed from speedtest
     :param download: Download speed from speedtest
     '''
-
+    log.debug(f"Set Dynamic Speed to: DOWN {mbits(download)} mbps; UP {mbits(upload)} mbps")
     connection = routeros_api.RouterOsApiPool(config['ros_ip'], username=secrets["ros_login"],
                                               password=secrets["ros_password"], plaintext_login=True)
     api = connection.get_api()
@@ -100,7 +106,7 @@ def ros_dynamic_speed(upload, download):
     for queue in list_queues.get():
         if queue['name'] in config['ros_queues']:
             log.debug(
-                f"Adjust Queue {queue['name']}: limit_at {int(upload) - int(upload / 10)}/{int(download) - int(download / 10)}; max_limit {int(upload)}/{int(download)}")
+                f"Adjust Queue {queue['name']}: max_limit {int(upload)}/{int(download)}")
 
             if config["ros_du_invert"] == True:
                 # Inverting upload and download values, because master queue is most likely applied to the bridge
@@ -111,35 +117,82 @@ def ros_dynamic_speed(upload, download):
     connection.disconnect()
 
 
-def speedtest():
+wan_upload = None
-    import speedtest
+wan_download = None
+results_dict = None
+test_started = False
+downloading = True  # True for download, False for upload
+threads = None
+servers = []
 
-    servers = []
-    threads = None
 
-    for i in range(0, 3):
+def threaded_speedtest():
-        try:
+    global test_started
-            log.debug("Initializing speedtest...")
+    global downloading
-            s = speedtest.Speedtest()
 
-            log.debug(f"Running test...")
+    s = speedtest.Speedtest()
-            s.get_servers(servers)
+    s.get_servers(servers)
-            s.get_best_server()
+    s.get_best_server()
-            s.download(threads=threads)
+    print(f"Running test...")
-            s.upload(threads=threads, pre_allocate=False)
+    test_started = True
+    downloading = True
+    s.download(threads=threads)
+    downloading = False
+    s.upload(threads=threads, pre_allocate=False)
 
-            results_dict = s.results.dict()
+    global results_dict
+    results_dict = s.results.dict()
+    print(
+        f"Speedtest.net result: DOWN {mbits(results_dict['download'])} mbps; UP {mbits(results_dict['upload'])} mbps;")
+    return results_dict
 
-            if results_dict['download'] >= config['ros_minimum_speed']:
-                download = round(results_dict['download'] / 1000000, 2)
-                upload = round(results_dict['upload'] / 1000000, 2)
-                break
-            time.sleep(10)
-        except:
-            log.error(f"Test failed, try {i + 1}/3", exc_info=True)
 
-    log.debug(f"{download}mbps, {upload}mbps")
+def threaded_wan_speed():
-    return download, upload, results_dict
+    global test_started
+    global results_dict
+    global downloading
+    print("Waiting for test to start...")
+    while not test_started:
+        time.sleep(1)
+    print("Allow warm-up...")
+    time.sleep(2)
+    uploads = []
+    downloads = []
+    print("Monitoring...")
+    while not results_dict:
+        connection = routeros_api.RouterOsApiPool(config['ros_ip'], username=secrets["ros_login"],
+                                                  password=secrets["ros_password"], plaintext_login=True)
+        api = connection.get_api()
+
+        traffic = api.get_resource('/').call('interface/monitor-traffic',
+                                             {'interface': config['ros_wan_interface'], 'once': ' '})[0]
+
+        traffic['rx-bits-per-second'] = int(traffic['rx-bits-per-second'])
+        traffic['tx-bits-per-second'] = int(traffic['tx-bits-per-second'])
+
+        if downloading:
+            downloads.append(traffic['rx-bits-per-second'])
+            print(f"DL: {mbits(traffic['rx-bits-per-second'])} mbps;")
+        else:
+            uploads.append(traffic['tx-bits-per-second'])
+            print(f"UP: {mbits(traffic['tx-bits-per-second'])} mbps;")
+
+        time.sleep(1)
+    global wan_download
+    global wan_upload
+    wan_download = median(downloads)
+    wan_upload = median(uploads)
+    print(f"Monitor result: {mbits(wan_download)} mbps;  {mbits(wan_upload)} mbps;")
+
+
+def test_speed():
+    sws = Thread(target=threaded_wan_speed)
+    st = Thread(target=threaded_speedtest)
+    st.start()
+    sws.start()
+    st.join()
+    sws.join()
+
 
 if __name__ == "__main__":
     '''
@@ -157,21 +210,18 @@ if __name__ == "__main__":
             ros_fastrack_enable(True)
             time.sleep(5)
 
-        download, upload, results_dict = speedtest()
+        test_speed()
 
         entry = Entry()
-        entry.upload = upload
+        entry.upload = mbits(wan_upload)
-        entry.download = download
+        entry.download = mbits(wan_download)
         entry.save()
 
-        ros_upload = results_dict['upload']
+        if wan_download < config['ros_minimum_speed']:
-        ros_download = results_dict['download']
+            wan_download = config['ros_minimum_speed']
-
-        if ros_download < config['ros_minimum_speed']:
-            ros_download = config['ros_minimum_speed']
 
         if config["ros_dynamic_speed"]:
-            ros_dynamic_speed(results_dict['upload'], results_dict['download'])
+            ros_dynamic_speed(wan_upload, wan_download)
             ros_fastrack_enable(False)
 
         dates, downloads, uploads = gather_data()