feat: Implementiere MQTT Command-Handling und Architektur-Refactoring

Integriert einen dedizierten MQTT Command-Listener und refaktoriert die Publisher/Controller-Beziehung für direktes Command-Handling.

- feat(mqtt): MqttPublisher verarbeitet Commands nun intern über `_handle_command`
- refactor(controller): Controller injiziert MqttPublisher für Heartbeats und direkten Zugriff
- perf(protocols): Optimierung der Demodulationsschleife für inaktive Protokolle in `message_unsynced.py`
- fix(main): Auflösung der zirkulären Abhängigkeit zwischen Controller und Publisher
- test: Anpassung der Testsuite an neue Architektur und Mocking-Strategie

Author: sidey79

.devcontainer/devcontainer.json

@@ -3,7 +3,11 @@
 {
 	"name": "Python 3",
 	// Or use a Dockerfile or Docker Compose file. More info: https://containers.dev/guide/dockerfile
-	"image": "mcr.microsoft.com/devcontainers/python:2-3-bookworm",
+	"dockerComposeFile": "docker-compose.yml",
+	"workspaceFolder": "/workspaces/PySignalduino",
+	"service": "devcontainer",
+	// Features to add to the dev container. More info: https://containers.dev/features.
+	// "features": {},
 	"features": {
 		"ghcr.io/devcontainers/features/node:1": {
 			"nodeGypDependencies": true,
@@ -18,32 +22,15 @@
 		}
 		//"ghcr.io/hspaans/devcontainer-features/pytest:2": {}
 	},
-
-	// Features to add to the dev container. More info: https://containers.dev/features.
-	// "features": {},
-
 	// Use 'forwardPorts' to make a list of ports inside the container available locally.
 	// "forwardPorts": [],
-
-	// Use 'postCreateCommand' to run commands after the container is created.
-	"postCreateCommand": "pip3 install --user -r requirements-dev.txt  -r requirements.txt || exit 0",
 	"customizations": {
 		"vscode": {
 			"extensions": [
 				"RooVeterinaryInc.roo-cline"
 			]
 		}
 	},
-	"runArgs": [
-					"--env-file", 
-					".devcontainer/devcontainer.env", 
-					"--network=bridge",	 
-					"--memory=4gb"
-	],
-
-	// Configure tool-specific properties.
-	// "customizations": {},
-
-	// Uncomment to connect as root instead. More info: https://aka.ms/dev-containers-non-root.
-	// "remoteUser": "root"
+	// Use 'postCreateCommand' to run commands after the container is created.
+	"postCreateCommand": "pip3 install --user -r requirements-dev.txt  -r requirements.txt || exit 0"
 }

.devcontainer/docker-compose.yml

@@ -0,0 +1,31 @@
+services:
+  devcontainer:
+    # Build the image from the existing devcontainer setup
+    image: mcr.microsoft.com/devcontainers/python:3
+    # The current working directory is mounted automatically
+    volumes:
+      - ..:/workspaces/PySignalduino
+      
+    # Use the existing settings from devcontainer.json
+    # Overriding the entrypoint is necessary when using a non-Compose devcontainer base image
+    command: /bin/bash -c "sleep infinity"
+    
+    # Environment variables from .devcontainer/devcontainer.env
+    env_file:
+      - ./devcontainer.env
+
+    # This ensures services in the compose file can be reached by their service name
+    # The default bridge network is sufficient for this purpose
+    
+  mqtt:
+    image: eclipse-mosquitto:latest
+    container_name: mosquitto-dev-broker
+    #ports:
+      # Expose port 1883 on the host, so other local clients can also connect if needed
+      #- "1883:1883"
+    volumes:
+       - ./mosquitto/config:/mosquitto/config
+       - ./mosquitto/data:/mosquitto/data
+       - ./mosquitto/log:/mosquitto/log
+    command: mosquitto -c /mosquitto/config/mosquitto.conf
+    restart: unless-stopped

.devcontainer/mosquitto/config/mosquitto.conf

@@ -0,0 +1,7 @@
+listener 1883 0.0.0.0
+allow_anonymous true 
+
+# Mosquitto Standard-Pfade
+persistence true
+persistence_location /mosquitto/data/
+log_dest file /mosquitto/log/mosquitto.log

.vscode/settings.json

@@ -45,5 +45,14 @@
         "tests"
     ],
     "python.testing.unittestEnabled": false,
-    "python.testing.pytestEnabled": true
+    "python.testing.pytestEnabled": true,
+    "vsmqtt.brokerProfiles": [
+        {
+            "name": "devmqtt",
+            "host": "mqtt",
+            "port": 1883,
+            "clientId": "vsmqtt_client_db93",
+            "savedSubscriptions": ['signalduino/']
+        }
+    ]
 }

AGENTS.md

@@ -253,22 +253,32 @@ Dieser Architecture-First Development Process ist für **alle** neuen Funktionen
 
 Die Einhaltung dieses Prozesses gewährleistet, dass Design-Entscheidungen bewusst getroffen, dokumentiert und nachvollziehbar sind, was die langfristige Wartbarkeit, Skalierbarkeit und Qualität des PySignalduino-Projekts sicherstellt.
 
-## Fehlerbehebungsprozess für fehlende Abhängigkeiten
-
+## Fehlerbehebungsprozess
 ### Problemidentifikation
 1. **Symptom:** ImportError oder ModuleNotFoundError während der Testausführung
 2. **Ursachenanalyse:**
    - Überprüfen der Traceback-Meldung auf fehlende Module
    - Vergleich mit requirements.txt und requirements-dev.txt
    - Prüfen der Dokumentation auf Installationsanweisungen
 
-### Lösungsimplementierung
+### Lösungsimplementierung (Abhängigkeiten)
 1. **requirements-dev.txt aktualisieren:**
    - Modulname zur Datei hinzufügen
    - Commit mit Conventional Commits Syntax erstellen (z.B. "fix: add <module> to requirements-dev.txt")
 2. **Dokumentation prüfen:**
    - Sicherstellen, dass Installationsanweisungen in README.md und docs/ aktuell sind
 
+### Problemidentifikation (Hohe CPU-Last im Parser)
+1. **Symptom:** Anhaltende 100% CPU-Auslastung auf einem oder mehreren Kernen während des Parsens von MU/MC-Nachrichten.
+2. **Ursachenanalyse:**
+   - **Parser-Architektur prüfen:** Der gesamte Parservorgang sollte in [`signalduino/controller.py`](signalduino/controller.py) über `asyncio.to_thread` abgewickelt werden.
+   - **Protokoll-Ineffizienz:** Die synchrone Demodulationsschleife in [`sd_protocols/message_unsynced.py`](sd_protocols/message_unsynced.py) oder [`sd_protocols/manchester.py`](sd_protocols/manchester.py) blockiert den Worker-Thread zu lange.
+3. **Validierung:** Temporäres Hinzufügen von Zeit-Logging (z.B. mit `time.perf_counter()`) in der Protokollschleife in `demodulate_mu` zur Identifizierung des blockierenden Protokolls.
+
+### Lösungsimplementierung (Parser-Performance)
+1. **Backtracking-Hölle vermeiden:** Wenn ein Protokoll eine sehr lange Demodulationszeit (z.B. > 10ms) aufweist, liegt wahrscheinlich ein Catastrophic Backtracking in einem regulären Ausdruck vor.
+2 **Deaktivierung inaktiver Protokolle:** Die `active`-Prüfung in `demodulate_mu` sollte verwendet werden, um inaktive Protokolle auszuschließen.
+
 ### Verifikation
 1. **Installation testen:**
    ```bash

docs/ASYNCIO_MIGRATION.md

@@ -194,6 +194,30 @@ Vergessene `await`‑Schlüsselwörter führen zu `RuntimeWarning` oder hängen
 
 Wenn Sie Threads und asyncio mischen müssen (z.B. für Legacy‑Code), verwenden Sie `asyncio.run_coroutine_threadsafe()` oder `loop.call_soon_threadsafe()`.
 
+### 4. Async-Busy-Loops und CPU-Auslastung (100%)
+
+Wenn `asyncio.Queue.get()` in einer `while True`-Schleife ständig Elemente zurückgibt (z.B. bei hohem Nachrichtenaufkommen), kann die Co-Routine den Event-Loop dominieren, selbst wenn die schwere Arbeit in einem Thread-Pool ausgelagert wird. Dies führt zu hoher CPU-Auslastung und sporadischer Bearbeitung anderer Async-Tasks.
+
+**Lösung:** Stellen Sie in schnell laufenden Verarbeitungsschleifen sicher, dass ein expliziter Yield-Punkt vorhanden ist, um anderen Tasks die Kontrolle zu übergeben.
+
+```python
+# Falsch (potenzielle Busy-Loop bei vollem Buffer)
+# while not self._stop_event.is_set():
+#     item = await queue.get()
+#     process_item(item) # Wenn schnell, dominiert diese Task
+
+# Korrekt
+while not self._stop_event.is_set():
+    try:
+        line = await self._raw_message_queue.get()
+        # ... Verarbeitung (kann await asyncio.to_thread enthalten) ...
+        
+        # Sicherstellen, dass andere Tasks Zeit bekommen
+        await asyncio.sleep(0.01) 
+    except Exception:
+        break
+```
+
 ## Vollständiges Migrationsbeispiel
 
 Hier ein komplettes Beispiel, das einen einfachen MQTT‑Bridge‑Service migriert:

main.py

@@ -10,6 +10,7 @@
 from signalduino.constants import SDUINO_CMD_TIMEOUT
 from signalduino.controller import SignalduinoController
 from signalduino.exceptions import SignalduinoConnectionError, SignalduinoCommandTimeout
+from signalduino.mqtt import MqttPublisher
 from signalduino.transport import SerialTransport, TCPTransport
 from signalduino.types import DecodedMessage, RawFrame # NEU: RawFrame
 
@@ -58,7 +59,7 @@ async def _async_run(args: argparse.Namespace):
         logger.info(f"Initialisiere serielle Verbindung auf {args.serial} mit {args.baud} Baud...")
         transport = SerialTransport(port=args.serial, baudrate=args.baud)
     elif args.tcp:
-        logger.info(f"Initialisiere TCP Verbindung zu {args.tcp}:{args.port}...")
+        logger.info(f"Initializing TCP connection to {args.tcp}:{args.port}...")
         transport = TCPTransport(host=args.tcp, port=args.port)
 
     # Wenn weder --serial noch --tcp (oder deren ENV-Defaults) gesetzt sind
@@ -67,18 +68,42 @@ async def _async_run(args: argparse.Namespace):
         sys.exit(1)
 
     # Controller initialisieren
+    # Wir initialisieren den Controller zuerst (mit mqtt_publisher=None),
+    # um ihn als Argument an MqttPublisher übergeben zu können (zirkuläre Abhängigkeit).
     controller = SignalduinoController(
         transport=transport,
         message_callback=message_callback,
-        logger=logger
+        logger=logger,
+        mqtt_publisher=None # Wird später zugewiesen
     )
+
+    # MQTT Publisher explizit initialisieren, falls Host in Argumenten gesetzt
+    mqtt_publisher = None
+    # args.mqtt_host ist gesetzt, wenn es entweder als CLI-Argument übergeben wurde
+    # oder wenn es als Umgebungsvariable gesetzt war (Standardwert in main()).
+    # Wir prüfen hier nur, ob ein Wert vorhanden ist, da der Controller sonst
+    # intern die Umgebungsvariable MQTT_HOST prüft.
+    if args.mqtt_host:
+        logger.info(f"Initializing MQTT publisher for host: {args.mqtt_host}")
+        mqtt_publisher = MqttPublisher(
+            logger=logger,
+            controller=controller, # Korrektur: Fügt das fehlende 'controller'-Argument hinzu
+            host=args.mqtt_host,
+            port=args.mqtt_port,
+            username=args.mqtt_username,
+            password=args.mqtt_password,
+            topic=args.mqtt_topic,
+        )
+        # Weisen Sie den Publisher dem Controller nachträglich zu.
+        controller.mqtt_publisher = mqtt_publisher # NEU: Nachträgliche Zuweisung
+
 
     # Starten
     try:
-        logger.info("Verbinde zum Signalduino...")
+        logger.info("Connecting to Signalduino...")
         # NEU: Verwende async with Block
         async with controller:
-            logger.info("Verbunden! Starte Initialisierung und Hauptschleife...")
+            logger.info("Connected! Starting initialization and main loop...")
 
             # Starte die Hauptschleife, warte auf deren Beendigung oder ein Timeout
             await controller.run(timeout=args.timeout)

sd_protocols/message_unsynced.py

@@ -45,6 +45,8 @@ def demodulate_mu(self, msg_data: Dict[str, Any], msg_type: str = "MU") -> List[
         mu_protocols = self.get_keys('clockabs')
 
         for pid in mu_protocols:
+            if not self.check_property(pid, 'active', True):
+                continue
             self._logging(f"MU checking PID {pid}", 5)
             # Prepare working copy of raw_data and patterns
             # (Perl does this per protocol iteration because filterfunc might modify them)
@@ -144,63 +146,40 @@ def demodulate_mu(self, msg_data: Dict[str, Any], msg_type: str = "MU") -> List[
             # Construct Regex
             # Perl: $regex="(?:$startStr)($signalRegex)"; where signalRegex is (one|zero|float){min,}
 
-            signal_or_group = "|".join(signal_regex_parts)
-            if self.get_property(pid, 'reconstructBit'):
-                 # Add endPatternLookup keys
-                 extras = [re.escape(k) for k in end_pattern_lookup.keys()]
-                 if extras:
-                     signal_or_group += "|" + "|".join(extras)
-            
-            length_min = self.check_property(pid, 'length_min', 0)
-            # length_max = self.check_property(pid, 'length_max', '') 
-            
-            # Python re doesn't support variable length lookbehind or similar easily, 
-            # but here we are matching forward.
-            # Perl loop: while ( $rawData =~ m/$regex/g)
-            # regex = (?:$startStr)((?:p1|p2|...){min,})
-            
-            # We already sliced raw_data to start at startStr if present.
-            # So startStr is at the beginning of current_raw_data.
-            # However, if startStr was found, it is consumed? 
-            # Perl: $rawData = substr($rawData, $message_start);
-            # regex = "(?:$startStr)($signalRegex)";
-            # So it matches startStr again at the beginning? 
-            # Wait, if we sliced it, the first chars ARE startStr.
-            
-            # Let's try to match iteratively
-            
-            full_regex_str = f"(?:{re.escape(start_str)})((?:{signal_or_group}){{ {length_min}, }})"
-            if self.get_property(pid, 'reconstructBit'):
-                 # Perl: $signalRegex .= '(?:' . join('|',keys %endPatternLookupHash) . ')?';
-                 # This is appended to the repeating group? No.
-                 # Perl code:
-                 # $signalRegex .= qq[{$length_min,}];
-                 # if (defined(...reconstructBit...)) { $signalRegex .= '(?:' . join('|',keys %endPatternLookupHash) . ')?'; }
-                 # So it's ((?:p1|p2){min,}(?:partial)?)
-                 pass # Logic handled below manually or we construct regex precisely
-                 
-            # It seems cleaner to just use the regex to find the data part
-            # Constructing complex regex in Python from dynamic parts
-            
-            # Simplified approach:
-            # 1. We are at start of potential message (startStr)
-            # 2. Extract as many valid chunks as possible
-            
-            # Re-implementing Perl's while loop over matches
-            # The regex matches the *entire* message (start + data).
+            # Build the base repeating pattern (signal_group_inner)
+            # Optimization for catastrophic backtracking (e.g., P61: '12|11' -> '1(2|1)') 
+            # Only apply if all parts share the same length and single-character prefix.
 
-            # Adjust signal_or_group for the repeating part
-            signal_group_inner = "|".join(signal_regex_parts)
+            unescaped_parts = list(pattern_lookup.keys())
+            signal_group_inner = "|".join(signal_regex_parts) # Default: unoptimized
 
+            try:
+                # Check if optimization is possible (all same length, same prefix, length > 1)
+                if unescaped_parts and all(len(p) == len(unescaped_parts[0]) for p in unescaped_parts) and len(unescaped_parts[0]) > 1:
+                    first_part = unescaped_parts[0]
+                    prefix = first_part[0]
+                    
+                    if all(p.startswith(prefix) for p in unescaped_parts):
+                        suffixes = [p[1:] for p in unescaped_parts]
+                        
+                        # Reconstruct the inner group: prefix(?:suffix1|suffix2|...)
+                        # Note: re.escape is safe even for single characters
+                        signal_group_inner = re.escape(prefix) + "(?:" + "|".join(re.escape(s) for s in suffixes) + ")"
+                        self._logging(f"MU Demod: Optimized repeating pattern for PID {pid}: {signal_group_inner}", 5)
+            except Exception:
+                # Fallback to default in case of unexpected pattern data
+                pass
+                
             # Handle reconstructBit logic for regex end
             reconstruct_part = ""
             if self.get_property(pid, 'reconstructBit') and end_pattern_lookup:
                 reconstruct_part = "(?:" + "|".join([re.escape(k) for k in end_pattern_lookup.keys()]) + ")?"
 
-            # We need to compile this regex
+            length_min = self.check_property(pid, 'length_min', 0)
+
             # Note: Python f-string braces need escaping
             regex_pattern = f"(?:{re.escape(start_str)})((?:{signal_group_inner}){{{length_min},}}{reconstruct_part})"
-            
+
             try:
                 # print(f"DEBUG: Compiling regex for {pid}: {regex_pattern[:50]}...")
                 matcher = re.compile(regex_pattern)

signalduino/controller.py

@@ -7,7 +7,7 @@
 from datetime import datetime, timedelta, timezone
 from typing import Any, Awaitable, Callable, List, Optional, Dict, Tuple, Pattern
 
-from .commands import SignalduinoCommands, MqttCommandDispatcher
+from .commands import SignalduinoCommands
 from .constants import (
     SDUINO_CMD_TIMEOUT,
     SDUINO_INIT_MAXRETRY,
@@ -56,7 +56,7 @@ def __init__(
         # NEU: Automatische Initialisierung des MqttPublisher, wenn keine Instanz übergeben wird und
         # die Umgebungsvariable MQTT_HOST gesetzt ist.
         if mqtt_publisher is None and os.environ.get("MQTT_HOST"):
-            self.mqtt_publisher = MqttPublisher(logger=self.logger)
+            self.mqtt_publisher = MqttPublisher(controller=self, logger=self.logger)
         else:
             self.mqtt_publisher = mqtt_publisher
 
@@ -78,8 +78,10 @@ def __init__(
 
         mqtt_topic_root = self.mqtt_publisher.base_topic if self.mqtt_publisher else None
         self.commands = SignalduinoCommands(self.send_command, mqtt_topic_root)
-        if mqtt_publisher:
-            self.mqtt_dispatcher = MqttCommandDispatcher(self)
+
+    def get_version(self) -> Optional[str]:
+        """Returns the cached firmware version string."""
+        return self.init_version_response
 
     async def send_command(
         self,
@@ -148,7 +150,7 @@ async def _reader_task(self) -> None:
                     self.logger.debug(f"Reader task received line: {line}")
                     await self._raw_message_queue.put(line)
 
-                await asyncio.sleep(0)  # Ensure yield, even if readline returns immediately without data
+                await asyncio.sleep(0.01)  # Ensure minimal yield time to prevent 100% CPU usage
             except Exception as e:
                 self.logger.error(f"Reader task error: {e}")
                 break
@@ -167,6 +169,9 @@ async def _parser_task(self) -> None:
                         # Verwende die neue MqttPublisher.publish(message: DecodedMessage) Signatur
                         await self.mqtt_publisher.publish(decoded[0])
                     await self._handle_as_command_response(line)
+                
+                # Ensure a minimal yield time for other tasks when the queue is rapidly processed.
+                await asyncio.sleep(0.01)
             except Exception as e:
                 self.logger.error(f"Parser task error: {e}")
                 break
@@ -369,12 +374,4 @@ async def _publish_status_heartbeat(self) -> None:
                 "version": self.init_version_response,
                 "connected": not self.transport.closed()
             }
-            await self.mqtt_publisher.publish_simple("status/heartbeat", json.dumps(status))
-
-    async def _handle_mqtt_command(self, topic: str, payload: str) -> None:
-        """Handle incoming MQTT commands."""
-        if self.mqtt_dispatcher:
-            try:
-                await self.mqtt_dispatcher.dispatch(topic, payload)
-            except CommandValidationError as e:
-                self.logger.error(f"Invalid MQTT command: {e}")
+            await self.mqtt_publisher.publish_simple("status/heartbeat", json.dumps(status))