Update

.github/workflows/dev-ci.yml

@@ -28,6 +28,10 @@ jobs:
           cache: pip
           cache-dependency-path: scripts/requirements-dev.txt
 
+      - name: Install system dependencies
+        if: matrix.test-type == 'pytest'
+        run: sudo apt-get install -y libgl1 libglib2.0-0 libegl1
+
       - name: Install dependencies
         run: |
           echo "Installing dependencies"

BlocksScreen.cfg

@@ -3,4 +3,7 @@ host: localhost
 port: 7125
 
 [screensaver]
-timeout: 5000
+timeout: 5000
+
+[usb_manager]
+gcodes_dir: ~/printer_data/gcodes/

BlocksScreen/BlocksScreen.py

@@ -2,11 +2,28 @@
 import sys
 import typing
 
-import logger
-from lib.panels.mainWindow import MainWindow
-from PyQt6 import QtCore, QtGui, QtWidgets
+from logger import CrashHandler, LogManager, install_crash_handler, setup_logging
+
+install_crash_handler()
+
+from lib.panels.mainWindow import MainWindow  # noqa: E402
+from PyQt6 import QtCore, QtGui, QtWidgets  # noqa: E402
+
+
+class BlocksScreenApp(QtWidgets.QApplication):
+    """QApplication subclass that routes unhandled slot exceptions to CrashHandler."""
+
+    def notify(self, a0: QtCore.QObject, a1: QtCore.QEvent) -> bool:  # type: ignore[override]
+        try:
+            return super().notify(a0, a1)
+        except Exception:
+            exc_type, exc_value, exc_tb = sys.exc_info()
+            handler = CrashHandler._instance
+            if handler is not None and exc_type is not None and exc_value is not None:
+                handler._exception_hook(exc_type, exc_value, exc_tb)
+            return False
+
 
-_logger = logging.getLogger(name="logs/BlocksScreen.log")
 QtGui.QGuiApplication.setAttribute(
     QtCore.Qt.ApplicationAttribute.AA_SynthesizeMouseForUnhandledTouchEvents,
     True,
@@ -22,13 +39,6 @@
 RESET = "\033[0m"
 
 
-def setup_app_loggers():
-    """Setup logger"""
-    _ = logger.create_logger(name="logs/BlocksScreen.log", level=logging.DEBUG)
-    _logger = logging.getLogger(name="logs/BlocksScreen.log")
-    _logger.info("============ BlocksScreen Initializing ============")
-
-
 def show_splash(window: typing.Optional[QtWidgets.QWidget] = None):
     """Show splash screen on app initialization"""
     logo = QtGui.QPixmap("BlocksScreen/BlocksScreen/lib/ui/resources/logoblocks.png")
@@ -38,13 +48,28 @@ def show_splash(window: typing.Optional[QtWidgets.QWidget] = None):
         splash.finish(window)
 
 
+def on_quit() -> None:
+    logging.info("Final exit cleanup")
+    LogManager.shutdown()
+
+
 if __name__ == "__main__":
-    setup_app_loggers()
-    BlocksScreen = QtWidgets.QApplication([])
+    setup_logging(
+        filename="logs/BlocksScreen.log",
+        level=logging.DEBUG,
+        console_output=True,
+        console_level=logging.DEBUG,
+        capture_stderr=True,
+        capture_stdout=False,
+    )
+    _logger = logging.getLogger(__name__)
+    _logger.info("============ BlocksScreen Initializing ============")
+    BlocksScreen = BlocksScreenApp([])
     BlocksScreen.setApplicationName("BlocksScreen")
     BlocksScreen.setApplicationDisplayName("BlocksScreen")
     BlocksScreen.setDesktopFileName("BlocksScreen")
     main_window = MainWindow()
     BlocksScreen.processEvents()
+    BlocksScreen.aboutToQuit.connect(on_quit)
     main_window.show()
     sys.exit(BlocksScreen.exec())

BlocksScreen/configfile.py

@@ -37,6 +37,8 @@
 
 from helper_methods import check_file_on_path
 
+logger = logging.getLogger(__name__)
+
 HOME_DIR = os.path.expanduser("~/")
 WORKING_DIR = os.getcwd()
 DEFAULT_CONFIGFILE_PATH = pathlib.Path(HOME_DIR, "printer_data", "config")
@@ -56,11 +58,19 @@ class ConfigError(Exception):
     """Exception raised when Configfile errors exist"""
 
     def __init__(self, msg) -> None:
+        """Store the error message on both the exception and the ``msg`` attribute."""
         super().__init__(msg)
         self.msg = msg
 
 
 class BlocksScreenConfig:
+    """Thread-safe wrapper around :class:`configparser.ConfigParser` with raw-text tracking.
+
+    Maintains a ``raw_config`` list that mirrors the on-disk file so that
+    ``add_section``, ``add_option``, and ``update_option`` can write back
+    changes without losing comments or formatting.
+    """
+
     config = configparser.ConfigParser(
         allow_no_value=True,
     )
@@ -70,16 +80,19 @@ class BlocksScreenConfig:
     def __init__(
         self, configfile: typing.Union[str, pathlib.Path], section: str
     ) -> None:
+        """Initialise with the path to the config file and the default section name."""
         self.configfile = pathlib.Path(configfile)
         self.section = section
         self.raw_config: typing.List[str] = []
         self.raw_dict_config: typing.Dict = {}
         self.file_lock = threading.Lock()  # Thread safety for future work
 
     def __getitem__(self, key: str) -> BlocksScreenConfig:
+        """Return a :class:`BlocksScreenConfig` for *key* section (same as ``get_section``)."""
         return self.get_section(key)
 
     def __contains__(self, key):
+        """Return True if *key* is a section in the underlying ConfigParser."""
         return key in self.config
 
     def sections(self) -> typing.List[str]:
@@ -91,7 +104,7 @@ def get_section(
     ) -> BlocksScreenConfig:
         """Get configfile section"""
         if not self.config.has_section(section):
-            raise configparser.NoSectionError(f"No section with name: {section}")
+            return fallback
         return BlocksScreenConfig(self.configfile, section)
 
     def get_options(self) -> list:
@@ -193,12 +206,14 @@ def getboolean(
         )
 
     def _find_section_index(self, section: str) -> int:
+        """Return the index of the ``[section]`` header line in ``raw_config``."""
         try:
             return self.raw_config.index("[" + section + "]")
         except ValueError as e:
             raise configparser.Error(f'Section "{section}" does not exist: {e}')
 
     def _find_section_limits(self, section: str) -> typing.Tuple:
+        """Return ``(start_index, end_index)`` of *section* in ``raw_config``."""
         try:
             section_start = self._find_section_index(section)
             buffer = self.raw_config[section_start:]
@@ -212,6 +227,7 @@ def _find_section_limits(self, section: str) -> typing.Tuple:
     def _find_option_index(
         self, section: str, option: str
     ) -> typing.Union[Sentinel, int, None]:
+        """Return the index of the *option* line within *section* in ``raw_config``."""
         try:
             start, end = self._find_section_limits(section)
             section_buffer = self.raw_config[start:][:end]
@@ -253,9 +269,9 @@ def add_section(self, section: str) -> None:
                 self.config.add_section(section)
                 self.update_pending = True
         except configparser.DuplicateSectionError as e:
-            logging.error(f'Section "{section}" already exists. {e}')
+            logger.error(f'Section "{section}" already exists. {e}')
         except configparser.Error as e:
-            logging.error(f'Unable to add "{section}" section to configuration: {e}')
+            logger.error(f'Unable to add "{section}" section to configuration: {e}')
 
     def add_option(
         self,
@@ -283,12 +299,46 @@ def add_option(
                 self.config.set(section, option, value)
                 self.update_pending = True
         except configparser.DuplicateOptionError as e:
-            logging.error(f"Option {option} already present on {section}: {e}")
+            logger.error(f"Option {option} already present on {section}: {e}")
         except configparser.Error as e:
-            logging.error(
+            logger.error(
                 f'Unable to add "{option}" option to section "{section}": {e} '
             )
 
+    def update_option(
+        self,
+        section: str,
+        option: str,
+        value: typing.Any,
+    ) -> None:
+        """Update an existing option's value in both raw tracking and configparser."""
+        try:
+            with self.file_lock:
+                if not self.config.has_section(section):
+                    self.add_section(section)
+
+                if not self.config.has_option(section, option):
+                    self.add_option(section, option, str(value))
+                    return
+
+                line_idx = self._find_option_line_index(section, option)
+                self.raw_config[line_idx] = f"{option}: {value}"
+                self.config.set(section, option, str(value))
+                self.update_pending = True
+        except Exception as e:
+            logger.error(
+                f'Unable to update option "{option}" in section "{section}": {e}'
+            )
+
+    def _find_option_line_index(self, section: str, option: str) -> int:
+        """Find the index of an option line within a specific section."""
+        start, end = self._find_section_limits(section)
+        opt_regex = re.compile(rf"^\s*{re.escape(option)}\s*[:=]")
+        for i in range(start + 1, end):
+            if opt_regex.match(self.raw_config[i]):
+                return i
+        raise configparser.Error(f'Option "{option}" not found in section "{section}"')
+
     def save_configuration(self) -> None:
         """Save teh configuration to file"""
         try:
@@ -301,7 +351,7 @@ def save_configuration(self) -> None:
                 self.config.write(sio)
                 sio.close()
         except Exception as e:
-            logging.error(
+            logger.error(
                 f"ERROR: Unable to save new configuration, something went wrong while saving updated configuration. {e}"
             )
         finally:
@@ -319,6 +369,14 @@ def load_config(self):
             raise configparser.Error(f"Error loading configuration file: {e}")
 
     def _parse_file(self) -> typing.Tuple[typing.List[str], typing.Dict]:
+        """Read and normalise the config file into a raw line list and a nested dict.
+
+        Strips comments, normalises ``=`` to ``:`` separators, deduplicates
+        sections/options, and ensures the buffer ends with an empty line.
+
+        Returns:
+            A tuple of (raw_lines, dict_representation).
+        """
         buffer = []
         dict_buff: typing.Dict = {}
         curr_sec: typing.Union[Sentinel, str] = Sentinel.MISSING
@@ -336,17 +394,18 @@ def _parse_file(self) -> typing.Tuple[typing.List[str], typing.Dict]:
                     if not line:
                         continue
                 # remove leading and trailing white spaces
-                line = re.sub(r"\s*([:=])\s*", r"\1", line)
+                line = re.sub(r"\s*([:=])\s*", r"\1 ", line)
                 line = re.sub(r"=", r":", line)
                 # find the beginning of sections
                 section_match = re.compile(r"[^\s]*\[([^]]+)\]")
                 match_sec = re.match(section_match, line)  #
                 if match_sec:
                     sec_name = re.sub(r"[\[*\]]", r"", line)
                     if sec_name not in dict_buff.keys():
-                        buffer.extend(
-                            [""]
-                        )  # REFACTOR: Just add some line separation between sections
+                        if buffer:
+                            buffer.extend(
+                                [""]
+                            )  # REFACTOR: Just add some line separation between sections
                         dict_buff.update({sec_name: {}})
                         curr_sec = sec_name
                     else:
@@ -386,6 +445,6 @@ def get_configparser() -> BlocksScreenConfig:
     config_object = BlocksScreenConfig(configfile=configfile, section="server")
     config_object.load_config()
     if not config_object.has_section("server"):
-        logging.error("Error loading configuration file for the application.")
+        logger.error("Error loading configuration file for the application.")
         raise ConfigError("Section [server] is missing from configuration")
-    return BlocksScreenConfig(configfile=configfile, section="server")
+    return config_object

BlocksScreen/devices/storage/__init__.py

@@ -0,0 +1,24 @@
+from .usb_controller import USBManager
+
+__doc__ = """
+
+The storage package contains a tool that monitors 
+pluggable usb devices via python-sdbus library. 
+While offering an automounting option.
+The package is also capable of creating a symlink that 
+points directly to the mounted usb drive on the gcodes
+directory. 
+
+
+There is still a lot of functionality missing, that may 
+be added in the future, but for now it just automounts, 
+creates symlinks, cleans up broken symlinks on the 
+gcodes directory.
+
+
+All tools related to storage devices should be contained 
+in this package directory. 
+"""
+__version__ = "0.0.1"
+__all__ = ["USBManager"]
+__name__ = "storage"