Revize 1526cb8b

re #9422 Commented detector.py, event_listener.py, and usb_reader.py

client/src/usb_detector/detector.py

@@ -5 +5 @@
 from .usb_reader import read_connected_devices
 
 
-_listeners_connected = []
-_listeners_disconnected = []
-_last_connected_devices = []
-_config = None
+_listeners_connected = []       # list of listeners (USB devices is connected)
+_listeners_disconnected = []    # list of listeners (USB devices is disconnected)
+
+_last_connected_devices = []    # list of the lastly connected USB devices
+_config = None                  # instance of Config (config manager)
 
 
 def usb_detector_set_config(config):
+    """Initializes the usb detector module (file).
+
+    This function is meant to be called prior to calling
+    any other function of the detector module. It stores
+    an instance of the Config class which is then used
+    by other functions within this file.
+
+    :param config: instance of Config (config manager)
+    """
+    # Store the instance into the global variable.
     global _config
     _config = config
 
 
 def register_listener(callback, connected: bool = True):
+    """Registers a new event listener.
+
+    The caller is supposed to passed in a function they
+    wish to be called whenever an event occurs. What kind
+    of event will trigger (call) the callback function is
+    determined by the second parameter.
+
+    :param callback: Function that is called whenever the desired event happens.
+    :param connected: If the value is set to True, the callback function
+                      will be called whenever a USB device is connected.
+                      If it is set to False, it will be triggered whenever a
+                      USB device is disconnected.
+    """
     logging.info(f"Registering callback: {callback}.")
+
     if connected is True:
+        # Register the callback for "connected devices"
         _listeners_connected.append(callback)
     else:
+        # Register the callback for "disconnected devices"
         _listeners_disconnected.append(callback)
 
 
 def _notify_listeners(listeners: list, devices: list):
+    """ Notifies (calls) all listeners based on the even that just occurred.
+
+    This function is called whenever a USB device is plugged or unplugged.
+    Based on the type of the event, the corresponding list of listeners
+    is passed in along with a list of all devices involved in the event.
+
+    :param listeners: List of listeners registered for the event.
+    :param devices: List of all USB devices involved in the event
+                    (usually a single device).
+    """
+    # Make sure both lists are not None
     if listeners is None or devices is None:
         return
+
+    # Iterate over the listeners and notify them
+    # of all USB devices involved in the event.
     for callback in listeners:
         for device in devices:
             callback(device)
 
 
 def _store_connected_devices(devices: list):
+    """Stores the list of currently connected USB devices into a file.
+
+    This function is called whenever a device is connected or disconnected.
+    Its main purpose is to dump the list of the currently plugged devices
+    on the disk (so it's not kept in RAM when the computer shuts down). The
+    list is then loaded upon every start of the application.
+
+    :param devices: List of the devices that are currently connected to the PC.
+    """
     logging.debug("storing newly connected devices")
+
+    # Dump the list into a JSON format.
     with open(_config.connected_devices_filename, "w") as file:
         json.dump(devices, file)
 
 
 def _load_last_connected_devices() -> list:
+    """Loads the list of the connected devices from the disk.
+
+    This function is called with every start of the application.
+    It ensures that the application remembers the USB devices
+    that were connected to the PC before it was turned off
+    (persistent memory).
+
+    :return: List of the lastly connected USB devices.
+    """
     logging.debug("loading last connected devices")
     try:
         with open(_config.connected_devices_filename, "r") as file:
@@ -49 +110 @@
 
 
 def _get_connected_devices(detected_devices: list, last_connected_devices: list) -> list:
+    """Returns a list of USB devices that were just plugged into the computer.
+
+    Using the two lists passed in as parameters, it figures out what devices
+    were just connected to the PC. Essentially, any device of the detected_devices list
+    that does not apper in the last_connected_devices list must have been just plugged in.
+
+    :param detected_devices: list of currently plugged USB devices
+    :param last_connected_devices: list of the lastly connected USB devices
+    :return: list of all USB devices that were just plugged into the computer
+    """
+    # If there is no previous record of what USB devices where connected to the PC,
+    # all newly-connected devices are treated as if they were just plugged in.
     if last_connected_devices is None and detected_devices is not None:
         return detected_devices
+
+    # Return an empty list if no devices were detected.
     if detected_devices is None:
         return []
+
+    # Return a list of all devices that were just plugged into the PC.
     return [device for device in detected_devices if device not in last_connected_devices]
 
 
 def _get_disconnected_devices(detected_devices: list, last_connected_devices: list) -> list:
+    """Returns a list of USB devices that were just disconnected from the computer.
+
+    Using the two lists passed in as parameters, it figures out what devices where just
+    disconnected from the computer. Basically, any device that was seen connected to the PC
+    and does not apper in the detected_devices list must have been just unplugged from the PC.
+
+    :param detected_devices: list of currently plugged USB devices
+    :param last_connected_devices: list of the lastly connected USB devices
+    :return: list of all USB devices that were just disconnected from the PC
+    """
+    # If there is no previous record of what USB devices where connected to the PC,
+    # no devices were unplugged
     if last_connected_devices is None:
         return []
+
+    # Return last_connected_devices if no devices were detected.
     if detected_devices is None:
         return last_connected_devices
+
+    # Return a list of all devices that were just disconnected.
     return [device for device in last_connected_devices if device not in detected_devices]
 
 
 def _update():
+    """Updates the USB detector.
+
+    This function is periodically called from the usb_detector_run function.
+    It uses the other functions of this file to figure out if there have
+    been any changes since the last time this function was called - what
+    USB devices have been connected/disconnected.
+    """
+    # Retrieve a list of the currently plugged USB devices
+    # and store it globally within the file.
     global _last_connected_devices
     detected_devices = read_connected_devices()
 
+    # Figure out what USB devices were connected to the PC since
+    # the last time this function was called.
     connected_devices = _get_connected_devices(detected_devices, _last_connected_devices)
+
+    # Figure out what USB devices were disconnected from the PC since
+    # the last time this function was called.
     disconnected_devices = _get_disconnected_devices(detected_devices, _last_connected_devices)
 
+    # Notify both kinds of listeners (call the registered callback functions).
     _notify_listeners(_listeners_connected, connected_devices)
     _notify_listeners(_listeners_disconnected, disconnected_devices)
 
+    # If there have been any changes, update the file on the disk (persistent memory).
     if len(connected_devices) > 0 or len(disconnected_devices) > 0:
         _store_connected_devices(detected_devices)
         _last_connected_devices = detected_devices
 
 
 def usb_detector_run():
+    """Keeps detecting what USB devices were plugged/unplugged.
+
+    This function is instantiated as a thread that periodically scans
+    what USB devices are connected to the PC. The scan period can be
+    found and modified in the configuration file of the application.
+    """
     logging.info("USB device detector is now running")
 
+    # Read the list of the lastly connected USB devices from the disk (once).
     global _last_connected_devices
     _last_connected_devices = _load_last_connected_devices()
 
     while True:
+        # Update the USB detector.
         _update()
+
+        # Sleep for a predefined amount of seconds
         sleep(_config.scan_period_seconds)

client/src/usb_detector/event_listener.py

@@ -7 +7 @@
 
 
 def _get_metadata() -> dict:
+    """Returns metadata associated with the computer.
+
+    This metadata is sent to the server as a part
+    of each payload. It includes the username, hostname,
+    and timestamp.
+
+    :return: metadata associated with the PC
+    """
     logging.debug("getting computer metadata")
     return {
-        "username": getpass.getuser(),
-        "hostname": platform.uname().node,
-        "timestamp": str(datetime.now()).split('.')[0]
+        "username": getpass.getuser(),                  # username
+        "hostname": platform.uname().node,              # hostname
+        "timestamp": str(datetime.now()).split('.')[0]  # timestamp (format: 2022-04-07 12:11:02)
     }
 
 
 def _send_payload_to_server(device: dict, status: str):
+    """ Creates a payload and calls the send_data function to send it to the server.
+
+    Each payload consists of metadata, status (connected/disconnected), and device,
+    which contains a vendor id, product id, and serial number.
+
+    :param device: USB device that has been detected
+    :param status: status of the USB device (connected/disconnected)
+    """
     logging.debug("payload send preparation")
+
+    # Get metadata that will be put into the payload.
     payload = _get_metadata()
+
+    # Add information about the USB device.
     payload["device"] = device
+
+    # Add the status of the USB device (connected/disconnected).
     payload["status"] = status
+
+    # Send the payload off to the server.
     send_data(payload)
 
 
 def usb_connected_callback(device: dict):
+    """Callback function for a connected USB device.
+
+    This function gets called whenever a USB device is
+    plugged into the computer (it is registered as a listener in
+    the USB detector). The device consists of a vendor id, product id,
+    and serial number.
+
+    :param device: USB device that was just plugged into the PC.
+    """
     logging.info(f"Device {device} has been connected")
+
+    # Create a payload and send it to the API (server).
     _send_payload_to_server(device, "connected")
 
 
 def usb_disconnected_callback(device: dict):
+    """Callback function for a disconnected USB device.
+
+    This function gets called whenever a USB device is
+    disconnected from the computer (it is registered as a listener in
+    the USB detector). The device consists of a vendor id, product id,
+    and serial number.
+
+    :param device: USB device that was just disconnected from the PC.
+    """
     logging.info(f"Device {device} has been disconnected")
+
+    # Create a payload and send it to the API (server).
     _send_payload_to_server(device, "disconnected")

client/src/usb_detector/usb_reader.py

@@ -4 +4 @@
 import usb.util
 
 
+# list of devices from which the application
+# could not retrieve a serial number
 _invalid_devices = []
 
 
 def read_connected_devices():
+    """Reads and returns all USB devices currently connected to the computer.
+
+    It iterates over devices connected to individual buses and for each of
+    them, it tries to retrieve its vendor id, product id, and serial number.
+    If the application fails to retrieve the serial number from a device, it
+    will be stored into an in-RAM list to prevent "spam" logs. Once the application
+    successes to read the serial number, the device will be removed from the list.
+
+    :return: list of all USB devices connected to the PC
+    """
     logging.debug("reading all currently connected devices")
+
+    # Create an empty list of USB devices.
     detected_devices = []
 
+    # Get a list of all buses.
     busses = usb.busses()
 
     for bus in busses:
+        # Get all devices connected to the current bus.
         devices = bus.devices
         for dev in devices:
+            # Create a record of the device.
             device = {
                 "vendor_id": dev.idVendor,
                 "product_id": dev.idProduct
             }
+
+            # Try to retrieve the serial number of the device.
             serial_number = None
             device_info = usb.core.find(idProduct=dev.idProduct)
             try:
                 serial_number = usb.util.get_string(device_info, device_info.iSerialNumber)
             except ValueError:
+                # If you fail, store the device into the in-RAM list (if it's
+                # not already there).
                 if device not in _invalid_devices:
                     logging.warning(f"Could not retrieve serial number from device {device}")
                     _invalid_devices.append(device)
 
             if serial_number is not None:
+                # If you manage to read the serial number of a USB device
+                # that was previously stored into the list of "failures", remove it.
                 if device in _invalid_devices:
                     _invalid_devices.remove(device)
 
+                # Add the serial number into to USB device record.
                 device["serial_number"] = serial_number
+
+                # Append the record into the list of the connected USB devices.
                 detected_devices.append(device)
 
+    # Return the list of currently plugged USB devices.
     return detected_devices