Revize 474551ae

re #9419 Moved the config manager into a class (better modularity)

client/api_client.py

@@ -5 +5 @@
 from diskcache import Deque
 from requests import HTTPError, ConnectionError
 
-from config_manager import server_url, server_port, server_endpoint, cache_dir, \
-    cache_retry_period_seconds, cache_max_retries, cache_max_entries
 
+_uri = None
+_cache = None
+_config = None
 
-_cache = Deque(directory=cache_dir)
-_uri = server_url + ":" + server_port + server_endpoint
+
+def api_client_set_config(config):
+    global _config, _cache, _uri
+    _config = config
+    _cache = Deque(directory=_config.cache_dir)
+    _uri = config.server_url + ":" + config.server_port + config.server_endpoint
 
 
 def send_data(payload: dict):
@@ -27 +32 @@
 
 
 def _cache_failed_payload(payload: dict):
-    if len(_cache) >= cache_max_entries:
+    if len(_cache) >= _config.cache_max_entries:
         oldest_payload = _cache.pop()
         logging.warning(f"cache is full - discarding payload = {oldest_payload}")
 
@@ -35 +40 @@
     _cache.append(payload)
 
 
+def _resend_cached_payloads():
+    retries = min(_config.cache_max_retries, len(_cache))
+    logging.info(f"emptying the cache ({retries} records)")
+    for _ in range(0, retries):
+        payload = _cache.pop()
+        send_data(payload)
+
+
 def api_client_run():
     while True:
-        retries = min(cache_max_retries, len(_cache))
-        logging.info(f"emptying the cache ({retries} records)")
-        for _ in range(0, retries):
-            payload = _cache.pop()
-            send_data(payload)
-        sleep(cache_retry_period_seconds)
+        _resend_cached_payloads()
+        sleep(_config.cache_retry_period_seconds)

client/config_manager.py

@@ +1 @@
+from sys import exit
 from configparser import RawConfigParser
 
 
-CONFIG_FILE = "config.ini"
-
-usb_detector_section = "usb_detector"
-server_section = "server"
-logger_section = "logger"
-cache_section = "cache"
-
-config = RawConfigParser()
-config.read(CONFIG_FILE)
-
-scan_period_seconds = float(config[usb_detector_section]["scan_period_seconds"])
-connected_devices_filename = config[usb_detector_section]["connected_devices_filename"]
-
-server_url = config[server_section]["url"]
-server_port = config[server_section]["port"]
-server_endpoint = config[server_section]["end_point"]
-
-logger_config = config[logger_section]["file"]
-logger_format = config[logger_section]["format"]
-logger_level_str = config[logger_section]["level"]
-logger_level = 0
-
-if logger_level_str == "DEBUG":
-    logger_level = 10
-elif logger_level_str == "INFO":
-    logger_level = 20
-elif logger_level_str == "WARNING":
-    logger_level = 30
-elif logger_level_str == "ERROR":
-    logger_level = 40
-elif logger_level_str == "CRITICAL":
-    logger_level = 50
-
-cache_dir = config[cache_section]["directory"]
-cache_max_entries = int(config[cache_section]["max_entries"])
-cache_max_retries = int(config[cache_section]["max_retries"])
-cache_retry_period_seconds = float(config[cache_section]["retry_period_seconds"])
+class Config:
+
+    def __init__(self, filepath):
+        self.config = RawConfigParser()
+        if not self.config.read(filepath):
+            print(f"Failed to parse the config file {filepath}. Make sure you entered a valid path.")
+            exit(1)
+
+        self._parse_usb_detector_section()
+        self._parse_server_section()
+        self._parse_logger_section()
+        self._parse_cache_section()
+
+    def _parse_usb_detector_section(self):
+        section_name = "usb_detector"
+        self.scan_period_seconds = float(self.config[section_name]["scan_period_seconds"])
+        self.connected_devices_filename = self.config[section_name]["connected_devices_filename"]
+
+    def _parse_server_section(self):
+        section_name = "server"
+        self.server_url = self.config[section_name]["url"]
+        self.server_port = self.config[section_name]["port"]
+        self.server_endpoint = self.config[section_name]["end_point"]
+
+    def _parse_logger_section(self):
+        section_name = "logger"
+        self.logger_config_file = self.config[section_name]["config_file"]
+
+    def _parse_cache_section(self):
+        section_name = "cache"
+        self.cache_dir = self.config[section_name]["directory"]
+        self.cache_max_entries = int(self.config[section_name]["max_entries"])
+        self.cache_max_retries = int(self.config[section_name]["max_retries"])
+        self.cache_retry_period_seconds = float(self.config[section_name]["retry_period_seconds"])

client/main.py

@@ -5 +5 @@
 from tendo import singleton
 from sys import exit
 
-from config_manager import logger_config
-from usb_detector.detector import register_listener, usb_detector_run
+from config_manager import Config
+from usb_detector.detector import register_listener, usb_detector_run, usb_detector_set_config
 from usb_detector.event_listener import usb_connected_callback, usb_disconnected_callback
-from api_client import api_client_run
+from api_client import api_client_run, api_client_set_config
 
 
-def init_logging():
-    if exists(logger_config):
-        logging.config.fileConfig(fname=logger_config)
+def init_logging(app_config: Config):
+    if exists(app_config.logger_config_file):
+        logging.config.fileConfig(fname=app_config.logger_config_file)
+        api_client_set_config(app_config)
+        usb_detector_set_config(app_config)
     else:
-        print(f"Cant find logger configuration \"{logger_config}\"! Please specify valid path or define new.")
+        print(f"Cannot find logger configuration \"{app_config.logger_config_file}\"! Please specify valid a path or define a new one.")
         exit(1)
 
 
@@ -25 +27 @@
     except singleton.SingleInstanceException:
         exit(1)
 
-    init_logging()
+    config = Config("config.ini")
+    init_logging(config)
 
     register_listener(callback=usb_connected_callback, connected=True)
     register_listener(callback=usb_disconnected_callback, connected=False)
@@ -36 +39 @@
     api_thread = Thread(target=api_client_run)
     api_thread.setDaemon(True)
 
-    logging.info('starting USB detector.')
+    logging.info("Starting USB detector.")
     usb_detector_thread.start()
 
-    logging.info('starting API communication manager.')
+    logging.info("Starting API communication manager.")
     api_thread.start()
 
     usb_detector_thread.join()
     api_thread.join()
-
-    logging.info('application exit.')

client/usb_detector/detector.py

@@ -3 +3 @@
 from time import sleep
 
 from .usb_reader import read_connected_devices
-from config_manager import scan_period_seconds, connected_devices_filename
 
 
 _listeners_connected = []
 _listeners_disconnected = []
 _last_connected_devices = []
+_config = None
 
 
-def register_listener(callback, connected: bool = True):
-    logging.info(f'Registering callback: {callback}.')
+def usb_detector_set_config(config):
+    global _config
+    _config = config
+
 
+def register_listener(callback, connected: bool = True):
+    logging.info(f"Registering callback: {callback}.")
     if connected is True:
         _listeners_connected.append(callback)
     else:
@@ -21 +25 @@
 
 
 def _notify_listeners(listeners: list, devices: list):
+    if listeners is None or devices is None:
+        return
     for callback in listeners:
         for device in devices:
             callback(device)
 
 
 def _store_connected_devices(devices: list):
-    logging.debug('storing newly connected devices')
-    with open(connected_devices_filename, "w") as file:
+    logging.debug("storing newly connected devices")
+    with open(_config.connected_devices_filename, "w") as file:
         json.dump(devices, file)
 
 
 def _load_last_connected_devices() -> list:
-    logging.debug('loading last connected devices')
+    logging.debug("loading last connected devices")
     try:
-        with open(connected_devices_filename, "r") as file:
+        with open(_config.connected_devices_filename, "r") as file:
             return json.loads(file.read())
     except IOError:
-        logging.error('loading of last connected devices failed')
+        logging.error("loading of last connected devices failed")
         return []
 
 
@@ -73 +79 @@
 
     while True:
         _update()
-        sleep(scan_period_seconds)
+        sleep(_config.scan_period_seconds)

client/usb_detector/event_listener.py

@@ -7 +7 @@
 
 
 def _get_metadata() -> dict:
-    logging.debug('getting computer metadata')
+    logging.debug("getting computer metadata")
     return {
         "username": getpass.getuser(),
         "hostname": platform.uname().node,
@@ -16 +16 @@
 
 
 def _send_payload_to_server(device: dict, status: str):
-    logging.debug('payload send preparation')
+    logging.debug("payload send preparation")
     payload = _get_metadata()
     payload["device"] = device
     payload["status"] = status