Revize f7fb8759

re #9305 Added a python module for USB detection

client/main.py

@@ -1 +1 @@
 import logging
+from threading import Thread
 
 from tendo import singleton
 
 from config_manager import logger_format, logger_level
+from usb_detector.detector import register_listener, usb_detector_run
+from usb_detector.event_listener import usb_connected_callback, usb_disconnected_callback
+
 
 if __name__ == "__main__":
     app_instance = singleton.SingleInstance()
 
     logging.basicConfig(format=logger_format, level=logger_level)
 
+    register_listener(callback=usb_connected_callback, connected=True)
+    register_listener(callback=usb_disconnected_callback, connected=False)
+
+    usb_detector_thread = Thread(target=usb_detector_run)
+    usb_detector_thread.setDaemon(True)
+    usb_detector_thread.start()
 
+    usb_detector_thread.join()

client/usb_detector/detector.py

@@ +1 @@
+import os
+import json
+import logging
+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 = []
+
+
+def register_listener(callback, connected: bool = True):
+    if connected is True:
+        _listeners_connected.append(callback)
+    else:
+        _listeners_disconnected.append(callback)
+
+
+def _notify_listeners(listeners: list, devices: list):
+    for callback in listeners:
+        for device in devices:
+            callback(device)
+
+
+def _store_connected_devices(devices: list):
+    with open(connected_devices_filename, "w") as file:
+        json.dump(devices, file)
+
+
+def _load_last_connected_devices() -> list:
+    try:
+        with open(connected_devices_filename, "r") as file:
+            return json.loads(file.read())
+    except IOError:
+        return []
+
+
+def usb_detector_run():
+    logging.info("USB device detector is now running")
+
+    if not os.path.exists(connected_devices_filename):
+        dir_name = os.path.dirname(connected_devices_filename)
+        os.makedirs(dir_name)
+
+    while True:
+        last_connected_devices = _load_last_connected_devices()
+        detected_devices = read_connected_devices()
+
+        connected_devices = [device for device in detected_devices if device not in last_connected_devices]
+        disconnected_devices = [device for device in last_connected_devices if device not in detected_devices]
+
+        _notify_listeners(_listeners_connected, connected_devices)
+        _notify_listeners(_listeners_disconnected, disconnected_devices)
+
+        _store_connected_devices(detected_devices)
+        sleep(scan_period_seconds)

client/usb_detector/event_listener.py

@@ +1 @@
+import platform
+import logging
+import getpass
+from datetime import datetime
+
+
+def _get_metadata() -> dict:
+    return {
+        "username": getpass.getuser(),
+        "hostname": platform.uname().node,
+        "timestamp": str(datetime.now())
+    }
+
+
+def _send_payload_to_server(device: dict, status: str):
+    payload = _get_metadata()
+    payload["device"] = device
+    payload["status"] = status
+
+    # TODO send the payload off to the server
+
+
+def usb_connected_callback(device: dict):
+    logging.info(f"Device {device} has been connected")
+    _send_payload_to_server(device, "connected")
+
+
+def usb_disconnected_callback(device: dict):
+    logging.info(f"Device {device} has been disconnected")
+    _send_payload_to_server(device, "disconnected")

client/usb_detector/usb_reader.py

@@ +1 @@
+import usb.core
+import usb.util
+
+
+def read_connected_devices():
+    detected_devices = []
+
+    busses = usb.busses()
+
+    for bus in busses:
+        devices = bus.devices
+        for dev in devices:
+            detected_devices.append({
+                "vendor_id": dev.idVendor,
+                "product_id": dev.idProduct
+            })
+
+    return detected_devices