feat: commands and persistence for mqtt

Author: sidey79

signalduino/commands.py

@@ -0,0 +1,147 @@
+"""
+Encapsulates all serial commands for the SIGNALDuino firmware.
+"""
+
+from typing import Any, Callable, Optional, Pattern
+import re
+
+class SignalduinoCommands:
+    """
+    Provides methods to construct and send commands to the SIGNALDuino.
+    
+    This class abstracts the raw serial commands documented in AI_AGENT_COMMANDS.md.
+    """
+
+    def __init__(self, send_command_func: Callable[[str, bool, float, Optional[Pattern[str]]], Any]):
+        """
+        Initialize with a function to send commands.
+        
+        Args:
+            send_command_func: A callable that accepts (payload, expect_response, timeout, response_pattern)
+                               and returns the response (if expected).
+        """
+        self._send = send_command_func
+
+    # --- System Commands ---
+
+    def get_version(self, timeout: float = 2.0) -> str:
+        """Query firmware version (V)."""
+        pattern = re.compile(r"V\s.*SIGNAL(?:duino|ESP|STM).*", re.IGNORECASE)
+        return self._send("V", expect_response=True, timeout=timeout, response_pattern=pattern)
+
+    def get_help(self) -> str:
+        """Show help (?)."""
+        return self._send("?", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def get_free_ram(self) -> str:
+        """Query free RAM (R)."""
+        return self._send("R", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def get_uptime(self) -> str:
+        """Query uptime in seconds (t)."""
+        return self._send("t", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def ping(self) -> str:
+        """Ping device (P)."""
+        return self._send("P", expect_response=True, timeout=2.0, response_pattern=re.compile(r"OK"))
+
+    def get_cc1101_status(self) -> str:
+        """Query CC1101 status (s)."""
+        return self._send("s", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def disable_receiver(self) -> None:
+        """Disable reception (XQ)."""
+        self._send("XQ", expect_response=False, timeout=0, response_pattern=None)
+
+    def enable_receiver(self) -> None:
+        """Enable reception (XE)."""
+        self._send("XE", expect_response=False, timeout=0, response_pattern=None)
+
+    def factory_reset(self) -> str:
+        """Factory reset CC1101 and load EEPROM defaults (e)."""
+        return self._send("e", expect_response=True, timeout=5.0, response_pattern=None)
+
+    # --- Configuration Commands ---
+
+    def get_config(self) -> str:
+        """Read configuration (CG)."""
+        return self._send("CG", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def set_decoder_state(self, decoder: str, enabled: bool) -> None:
+        """
+        Configure decoder (C<CMD><FLAG>).
+        
+        Args:
+            decoder: One of 'MS', 'MU', 'MC', 'Mred', 'AFC', 'WMBus', 'WMBus_T'
+                     Internal mapping: S=MS, U=MU, C=MC, R=Mred, A=AFC, W=WMBus, T=WMBus_T
+            enabled: True to enable, False to disable
+        """
+        decoder_map = {
+            "MS": "S",
+            "MU": "U",
+            "MC": "C",
+            "Mred": "R",
+            "AFC": "A",
+            "WMBus": "W",
+            "WMBus_T": "T"
+        }
+        if decoder not in decoder_map:
+            raise ValueError(f"Unknown decoder: {decoder}")
+        
+        cmd_char = decoder_map[decoder]
+        flag_char = "E" if enabled else "D"
+        command = f"C{cmd_char}{flag_char}"
+        self._send(command, expect_response=False, timeout=0, response_pattern=None)
+
+    def set_manchester_min_bit_length(self, length: int) -> str:
+        """Set MC Min Bit Length (CSmcmbl=<val>)."""
+        return self._send(f"CSmcmbl={length}", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def read_cc1101_register(self, register: int) -> str:
+        """Read CC1101 register (C<reg>). Register is int, sent as 2-digit hex."""
+        reg_hex = f"{register:02X}"
+        return self._send(f"C{reg_hex}", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def write_register(self, register: int, value: int) -> str:
+        """Write to EEPROM/CC1101 register (W<reg><val>)."""
+        reg_hex = f"{register:02X}"
+        val_hex = f"{value:02X}"
+        return self._send(f"W{reg_hex}{val_hex}", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def init_wmbus(self) -> str:
+        """Initialize WMBus mode (WS34)."""
+        return self._send("WS34", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def read_eeprom(self, address: int) -> str:
+        """Read EEPROM byte (r<addr>)."""
+        addr_hex = f"{address:02X}"
+        return self._send(f"r{addr_hex}", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def read_eeprom_block(self, address: int) -> str:
+        """Read EEPROM block (r<addr>n)."""
+        addr_hex = f"{address:02X}"
+        return self._send(f"r{addr_hex}n", expect_response=True, timeout=2.0, response_pattern=None)
+
+    def set_patable(self, value: int) -> str:
+        """Write PA Table (x<val>)."""
+        val_hex = f"{value:02X}"
+        return self._send(f"x{val_hex}", expect_response=True, timeout=2.0, response_pattern=None)
+
+    # --- Send Commands ---
+    # These typically don't expect a response, or the response is just an echo/OK which might be hard to sync with async rx
+    
+    def send_combined(self, params: str) -> None:
+        """Send Combined (SC...). params should be the full string after SC, e.g. ';R=4...'"""
+        self._send(f"SC{params}", expect_response=False, timeout=0, response_pattern=None)
+
+    def send_manchester(self, params: str) -> None:
+        """Send Manchester (SM...). params should be the full string after SM."""
+        self._send(f"SM{params}", expect_response=False, timeout=0, response_pattern=None)
+
+    def send_raw(self, params: str) -> None:
+        """Send Raw (SR...). params should be the full string after SR."""
+        self._send(f"SR{params}", expect_response=False, timeout=0, response_pattern=None)
+    
+    def send_xfsk(self, params: str) -> None:
+        """Send xFSK (SN...). params should be the full string after SN."""
+        self._send(f"SN{params}", expect_response=False, timeout=0, response_pattern=None)

signalduino/persistence.py

@@ -0,0 +1,44 @@
+import json
+import os
+import uuid
+import logging
+from typing import Optional
+
+# Todo: Pfad anpassen
+CLIENT_ID_FILE = os.path.join(os.path.expanduser("~"), ".signalduino_id")
+logger = logging.getLogger(__name__)
+
+def get_or_create_client_id() -> str:
+    """
+    Liest die persistente Client-ID aus der Datei oder generiert eine neue und speichert sie.
+    """
+    client_id = None
+    
+    # 1. Versuche, die ID aus der Konfigurationsdatei zu lesen
+    try:
+        if os.path.exists(CLIENT_ID_FILE):
+            with open(CLIENT_ID_FILE, "r", encoding="utf-8") as f:
+                config = json.load(f)
+                client_id = config.get("client_id")
+    except Exception as e:
+        logger.warning("Fehler beim Lesen der Client-ID aus %s: %s", CLIENT_ID_FILE, e)
+        
+    # 2. Wenn keine ID gefunden wurde, generiere eine neue
+    if not client_id:
+        client_id = f"signalduino-{uuid.uuid4().hex}"
+        logger.info("Neue Client-ID generiert: %s", client_id)
+        
+        # 3. Speichere die ID persistent
+        try:
+            with open(CLIENT_ID_FILE, "w", encoding="utf-8") as f:
+                json.dump({"client_id": client_id}, f, indent=4)
+            logger.info("Client-ID dauerhaft gespeichert in %s", CLIENT_ID_FILE)
+        except Exception as e:
+            logger.error("Fehler beim Speichern der Client-ID in %s: %s", CLIENT_ID_FILE, e)
+            
+    return client_id
+
+if __name__ == "__main__":
+    # Beispiel für die Verwendung
+    logging.basicConfig(level=logging.INFO)
+    print(f"Client ID: {get_or_create_client_id()}")