Set video overlay when speakers are connected (#205)

.github/workflows/test.yml

@@ -0,0 +1,40 @@
+name: Test
+
+on:
+  pull_request:
+  push:
+    branches:
+      - master
+
+jobs:
+  test:
+    runs-on: ubuntu-20.04
+    steps:
+      - name: Checkout code
+        uses: actions/checkout@v3
+
+      - name: Setup Python
+        uses: actions/setup-python@v4
+        with:
+          python-version: 3.11
+
+      - name: Install dependencies
+        run: |
+          sudo apt update
+          sudo apt install python3 python3-pip -y
+          pip3 install pytest pytest-cov codecov
+
+      - name: Run tests
+        run: |
+          pytest --verbose --cov-report term-missing --cov=pitopd
+          coverage xml
+
+      - name: Upload coverage reports to Codecov
+        uses: codecov/codecov-action@v4
+        with:
+          files: ./coverage.xml
+          flags: python-tests
+          env_vars: OS,PYTHON
+          fail_ci_if_error: true
+          verbose: true
+          token: ${{ secrets.CODECOV_TOKEN }}

.pre-commit-config.yaml

@@ -33,17 +33,17 @@ repos:
     - id: isort
 
 -   repo: https://github.com/psf/black
-    rev: 24.4.2
+    rev: 24.8.0
     hooks:
     - id: black
 
 -   repo: https://github.com/PyCQA/flake8
-    rev: 7.1.0
+    rev: 7.1.1
     hooks:
     - id: flake8
 
 -   repo: https://github.com/pre-commit/mirrors-mypy
-    rev: v1.10.1
+    rev: v1.11.1
     hooks:
     -   id: mypy
-        additional_dependencies: [types-all]
+        additional_dependencies: [types-setuptools]

pitopd/config_parser.py

@@ -0,0 +1,181 @@
+#!/usr/bin/python3
+from dataclasses import dataclass, field
+from itertools import count
+from typing import Optional
+
+
+@dataclass
+class Section:
+    name: str
+    settings: list[str] = field(default_factory=list)
+    id: int = field(default_factory=count().__next__, init=False)
+
+    def cleanup(self):
+        return [x for x in self.settings if len(x) > 0 and not x.startswith("#")]
+
+    def find(self, setting_name: str):
+        for setting in self.settings:
+            if setting.startswith(setting_name):
+                return setting
+        return None
+
+    def add(self, setting: str):
+        self.settings.append(setting)
+
+    def remove(self, setting_name: str):
+        setting = self.find(setting_name)
+        if setting is not None:
+            self.settings.remove(setting)
+        else:
+            print(
+                f"Section.remove(): '{setting_name}' not found in section '{self.name}'"
+            )
+
+    def comment(self, setting_name: str):
+        setting = self.find(setting_name)
+        if setting is not None:
+            self.settings[self.settings.index(setting)] = f"#{setting}"
+        else:
+            print(
+                f"Section.comment(): '{setting_name}' not found in section '{self.name}'"
+            )
+
+    def uncomment(self, setting_name: str):
+        if not setting_name.startswith("#"):
+            setting_name = f"#{setting_name}"
+        setting = self.find(setting_name)
+        if setting is not None:
+            self.settings[self.settings.index(setting)] = setting[1:]
+        else:
+            print(
+                f"Section.uncomment(): '{setting_name}' not found in section '{self.name}'"
+            )
+
+
+class RpiConfigParser:
+    def __init__(self, config_file_path):
+        self.sections = []
+        self.path = config_file_path
+        self.load()
+
+    def load(self):
+        self.sections = []
+        section_name = "all"
+        section = Section(name=section_name, settings=[])
+        with open(self.path, "r") as f:
+            lines = f.readlines()
+        for i, line in enumerate(lines):
+            line = line.strip()
+            if (
+                not line.startswith("#")
+                and len(line) >= 2
+                and line[0] == "["
+                and line[-1] == "]"
+            ):
+                # Found a section tag
+                if line[1:-1] == "all" or section_name != line[1:-1]:
+                    self.add(section)
+                    section_name = line[1:-1]
+                section = Section(name=section_name, settings=[])
+                continue
+            section.settings.append(line)
+            if i == len(lines) - 1:
+                self.add(section)
+
+    def add(self, section: Section) -> None:
+        if not isinstance(section, Section):
+            raise TypeError(
+                "RpiConfigParser.add_section(): section must be of type Section"
+            )
+        self.sections.append(section)
+
+    def remove(self, section: Section) -> None:
+        if not isinstance(section, Section):
+            raise TypeError(
+                "RpiConfigParser.add_section(): section must be of type Section"
+            )
+        self.sections.remove(section)
+
+    def update(self, section: Section) -> None:
+        if not isinstance(section, Section):
+            raise TypeError(
+                "RpiConfigParser.add_section(): section must be of type Section"
+            )
+        for i, s in enumerate(self.sections):
+            if s.id == section.id:
+                self.sections[i] = section
+                return
+        self.add(section)
+
+    def find(self, section_name: str) -> Optional[Section]:
+        # A config file can have multiple sections with the same name.
+        # This function returns the last one since repeated settings are overwritten.
+        for section in reversed(self.sections):
+            if section.name == section_name:
+                return section
+        return None
+
+    def find_all(self, section_name: str) -> list[Section]:
+        # Return an array of sections with the same name
+        response = []
+        sections = reversed(self.sections)
+        for section in sections:
+            if section.name == section_name:
+                response.append(section)
+        return response
+
+    def write(self, path: Optional[str] = None):
+        if path is None:
+            path = self.path
+        with open(path, "w") as f:
+            for i, section in enumerate(self.sections):
+                # don't print the "all" section if it's the first one
+                if not (i == 0 and section.name == "all"):
+                    f.write(f"[{section.name}]\n")
+                for setting in section.settings:
+                    f.write(f"{setting}\n")
+
+    def find_all_sections_with_setting(
+        self, section_name: str, setting: str
+    ) -> list[Section]:
+        matching_sections = []
+        sections = self.find_all(section_name)
+        for section in sections:
+            if section.find(setting) is not None:
+                matching_sections.append(section)
+        return matching_sections
+
+    def add_or_uncomment(self, section_name: str, setting: str) -> None:
+        # check if already exists
+        sections = self.find_all_sections_with_setting(section_name, setting)
+        if len(sections) > 0:
+            print(f"Setting '{setting}' already exists in section '{section_name}'")
+            return
+
+        # check if it's commented
+        sections = self.find_all_sections_with_setting(section_name, f"#{setting}")
+        if len(sections) > 0:
+            print(f"Uncommenting setting '{setting}' in section '{section_name}'")
+            for section in sections:
+                section.uncomment(setting)
+            return
+
+        # didn't find it; add it
+        sec = self.find("all")
+        if sec is None:
+            sec = Section("all")
+        print(f"Adding setting '{setting}' to section '{section_name}'")
+        sec.add(f"{setting}")
+        self.update(sec)
+
+    def comment_setting(self, section_name: str, setting: str) -> None:
+        for section in self.find_all_sections_with_setting(section_name, setting):
+            print(f"Commenting setting '{setting}' in section '{section_name}'")
+            section.comment(setting)
+
+    def uncomment_setting(self, section_name: str, setting: str) -> None:
+        if not setting.startswith("#"):
+            setting = f"#{setting}"
+        for section in self.find_all_sections_with_setting(section_name, setting):
+            print(f"Uncommenting setting '{setting}' in section '{section_name}'")
+            section.uncomment(setting)

pitopd/peripheral_manager.py

@@ -349,12 +349,14 @@ def auto_initialise_peripherals(self):
                 self.attempt_enable_peripheral_by_name(current_peripheral.name)
 
     def configure_hifiberry(self):
+        logger.info("Configuring HiFiBerry audio output")
         if I2S.get_current_state() is True:
             if state.get("sound", "i2s_configured", fallback="false") == "false":
                 call(("/usr/sbin/alsactl", "-f", self._i2s_config_file_path, "restore"))
                 state.set("sound", "i2s_configured", "true")
                 System.reboot_system()
             else:
+                logger.info("Setting HiFiBerry as audio output")
                 Hifiberry.set_as_audio_output(user=get_user_using_first_display())
 
     def get_peripheral_enabled(self, peripheral):

pitopd/pthub/pthub_i2c.py

@@ -105,7 +105,7 @@ def _refresh_state(self):
 
         logger.debug("Getting battery time...")
         if not self._get_battery_data(BatteryDataType.time):
-            logger.warning("Unable to get battery time")
+            logger.debug("Unable to get battery time")
             # return False - Don't return False - non essential
 
         logger.debug("Wattage set from battery voltage and current")