Files
pizerovoice/steckdose.md
2026-01-18 16:27:21 +00:00

34 KiB

Pi Zero 2W + ReSpeaker - IKEA TRETAKT Integration via FritzBox

Sprachsteuerung von IKEA TRETAKT Smart Plugs

Hardware: IKEA TRETAKT (Zigbee 3.0, €8)
Integration: FritzBox Smart Gateway (Zigbee Bridge)
Steuerung: Pi Zero 2W → FritzBox → TRETAKT Steckdose


UNTERSCHIED: TRETAKT vs. FRITZ!DECT

Aspekt FRITZ!DECT 200 IKEA TRETAKT
Funkstandard DECT ULE Zigbee 3.0
Preis €25-35 €8 (super günstig!)
Reichweite ~300m ~100m
Netzwerk DECT Mesh Zigbee Mesh
Kompatibilität Nur FritzBox DECT Philips Hue, Dirigera, FritzBox + Gateway
API Direkt via TR-064 Via FritzBox Zigbee Integration

Vorteil TRETAKT: Deutlich günstiger, Zigbee-Standard, erweiterbares Netzwerk!
Nachteil: Braucht FRITZ!Smart Gateway (~€100) wenn du nur DECT-Geräte hast. Mit FritzBox 7690 oder neuer schon integriert.


VORAUSSETZUNGEN

  1. FritzBox mit Zigbee-Unterstützung:

    • FritzBox 7690 / 7680 / 7590 AX (ab 2021)
    • FRITZ!Smart Gateway (~€100, zusätzliche Zigbee-Bridge)
    • Ältere Modelle: Nicht unterstützt
  2. IKEA TRETAKT Steckdosen (€8 pro Stück)

    • Zigbee 3.0 Standard
    • Kein IKEA Hub nötig!

Überprüfe dein FritzBox-Modell:

# Später vom Pi aus:
curl -s "http://192.168.178.1:49000/tr64desc.xml" | grep modelName

TEIL 1: FritzBox vorbereiten

1.1 FritzBox Zigbee aktivieren

Je nach Modell unterschiedlich:

Option A: FritzBox 7690/7680 (integriertes Zigbee)

  1. Öffne: http://fritz.box → Admin-Passwort eingeben
  2. Gehe zu: HeimnetzGeräte und SensorenSmarthome Einstellungen
  3. Aktiviere: "Zigbee aktivieren"
  4. Klicke: "Speichern"

Option B: Mit FRITZ!Smart Gateway (für ältere Modelle)

  1. Gateway kaufen (~€100)
  2. Mit Ethernet an FritzBox verbinden (nicht WiFi!)
  3. FritzBox automatisch verbindet sich
  4. Einrichtung ähnlich wie Option A

Hinweis: Dieses Dokument geht von Option A aus (integriertes Zigbee).

1.2 Benutzer für API-Zugriff erstellen

  1. FritzBox öffnen: http://fritz.box
  2. SystemFRITZ!Box-Benutzer
  3. Neuer Benutzer klicken:
    • Benutzername: raspi_home
    • Passwort: Starkes Passwort (z.B. TreTakt2025Pi)
  4. Haken setzen:
    • Zugriff auf Smart Home Geräte (WICHTIG!)
  5. OK klicken

TEIL 2: TRETAKT Steckdose koppeln

2.1 Koppelmodus starten

  1. Öffne FritzBox: http://fritz.box
  2. Gehe zu: HeimnetzGeräte und Sensoren
  3. Klicke auf: "Neues Gerät anmelden" oder "Gerät koppeln"
  4. Die FritzBox sucht jetzt nach Zigbee-Geräten (30 Sekunden)

2.2 TRETAKT Steckdose anmelden

  1. TRETAKT Steckdose in die Steckdose stecken
  2. Reset-Taste auf der Rückseite 8-10 Sekunden halten
    • LED sollte schnell blinken (Kopplungsmodus aktiv)
  3. FritzBox sollte das Gerät erkennen
  4. Gib einen Namen ein (z.B. "Wohnzimmer Musik")
  5. Speichern

Steckdose ist jetzt gekoppelt!

2.3 Gerät-ID (Identifier) ermitteln

Dies brauchst du für die Python-Steuerung:

  1. FritzBox: HeimnetzGeräte und Sensoren
  2. Klicke auf die TRETAKT Steckdose
  3. Suche nach "Identifier" oder "ID"
    • Format: zigbee:ABC123:1:on oder ähnlich
  4. Notiere diese ID!

Alternative (via XML):

curl -s "http://192.168.178.1:49000/tr64desc.xml" | grep -i zigbee

Beispiel-IDs:

zigbee:0013A20041C4ABCD:1:on

TEIL 3: Python-Installation

3.1 Bibliothek für FritzBox Zigbee installieren

# SSH zum Pi
ssh pi@<pi-ip>

# Installiere die FritzBox Bibliothek
pip3 install fritzconnection

# Optional: für vollständigere Zigbee-Unterstützung
pip3 install pyfritz

3.2 Test durchführen

Erstelle test_tretakt.py:

nano ~/voice_assistant/test_tretakt.py
#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Test-Skript: IKEA TRETAKT via FritzBox steuern
Zigbee-Steckdosen über REST API kontrollieren
"""

import sys
from fritzconnection.lib.homeauto import FritzHomeAutomation

# ============================================================================
# KONFIGURATION
# ============================================================================

FRITZBOX_IP = "192.168.178.1"
FRITZBOX_USER = "raspi_home"
FRITZBOX_PASSWORD = "TreTakt2025Pi"

# Die Identifier deiner TRETAKT Steckdosen (aus Schritt 2.3)
TRETAKT_DEVICES = {
    "musik": {
        "identifier": "zigbee:0013A20041C4ABCD:1:on",  # ← DEINE ID!
        "name": "Wohnzimmer Musik",
    },
    "licht": {
        "identifier": "zigbee:0013A20041C4ABCE:1:on",  # ← DEINE ID!
        "name": "Küche Licht",
    },
    "stopp": {
        "identifier": "zigbee:0013A20041C4ABCF:1:on",  # ← DEINE ID!
        "name": "Flur Steckdose",
    },
}

# ============================================================================
# FUNKTIONEN
# ============================================================================

def test_connection():
    """Teste Verbindung zur FritzBox"""
    print("Verbinde zur FritzBox...")
    try:
        fha = FritzHomeAutomation(
            avm_device_ip=FRITZBOX_IP,
            user=FRITZBOX_USER,
            password=FRITZBOX_PASSWORD
        )
        print("✓ Verbindung erfolgreich!\n")
        return fha
    except Exception as e:
        print(f"✗ Verbindungsfehler: {e}")
        return None

def list_all_devices(fha):
    """Zeige alle Zigbee-Geräte"""
    print("Verfügbare Zigbee-Geräte:")
    try:
        devices = fha.get_devices()
        if not devices:
            print("Keine Geräte gefunden")
            return
        
        for device in devices:
            print(f"\nGerät: {device.name}")
            print(f"  Identifier: {device.identifier}")
            print(f"  Typ: {device.device_type}")
            print(f"  Status: {'AN ✓' if device.state else 'AUS ✗'}")
            
            # Zusätzliche Infos
            if hasattr(device, 'power'):
                print(f"  Leistung: {device.power / 100}W")
            if hasattr(device, 'energy'):
                print(f"  Energie: {device.energy / 1000}kWh")
    except Exception as e:
        print(f"✗ Fehler: {e}")

def toggle_device(fha, identifier):
    """Schalte Gerät um"""
    try:
        device = fha.get_device_by_identifier(identifier)
        if not device:
            print(f"✗ Gerät nicht gefunden: {identifier}")
            return False
        
        current_state = device.state
        print(f"Status: {'AN' if current_state else 'AUS'}")
        
        if current_state:
            print("Schalte AUS...")
            device.turn_off()
        else:
            print("Schalte AN...")
            device.turn_on()
        
        import time
        time.sleep(1)
        
        device.update()
        new_state = device.state
        print(f"Neuer Status: {'AN ✓' if new_state else 'AUS ✗'}\n")
        return True
    except Exception as e:
        print(f"✗ Fehler beim Schalten: {e}")
        return False

def get_device_info(fha, identifier, name):
    """Zeige Geräteinformationen"""
    try:
        device = fha.get_device_by_identifier(identifier)
        if not device:
            print(f"✗ Gerät nicht gefunden")
            return
        
        print(f"\n{'='*50}")
        print(f"TRETAKT: {name}")
        print(f"{'='*50}")
        print(f"Identifier: {device.identifier}")
        print(f"Typ: {device.device_type}")
        print(f"Status: {'AN ✓' if device.state else 'AUS ✗'}")
        
        if hasattr(device, 'power'):
            print(f"Leistung: {device.power / 100:.1f}W")
        if hasattr(device, 'energy'):
            print(f"Verbrauch: {device.energy / 1000:.2f}kWh")
        
        print(f"{'='*50}\n")
    except Exception as e:
        print(f"✗ Fehler: {e}")

# ============================================================================
# HAUPTPROGRAMM
# ============================================================================

if __name__ == "__main__":
    print("="*60)
    print("IKEA TRETAKT via FritzBox - Test")
    print("="*60 + "\n")
    
    # Verbindung testen
    fha = test_connection()
    if not fha:
        sys.exit(1)
    
    # Alle Geräte auflisten
    list_all_devices(fha)
    
    # Info zu Testgerät
    if "musik" in TRETAKT_DEVICES:
        device_config = TRETAKT_DEVICES["musik"]
        get_device_info(fha, device_config["identifier"], device_config["name"])
        
        # Test schalten
        print("Teste Schalten...\n")
        toggle_device(fha, device_config["identifier"])
        
        print("✓ Test erfolgreich!")

Speichere und teste:

# Bearbeite die IDs
nano ~/voice_assistant/test_tretakt.py
# Trage deine Identifiers ein

# Teste
python3 ~/voice_assistant/test_tretakt.py

Erwartete Ausgabe:

============================================================
IKEA TRETAKT via FritzBox - Test
============================================================

Verbinde zur FritzBox...
✓ Verbindung erfolgreich!

Verfügbare Zigbee-Geräte:

Gerät: Wohnzimmer Musik
  Identifier: zigbee:0013A20041C4ABCD:1:on
  Typ: HAN_FUN
  Status: AUS ✗

==================================================
TRETAKT: Wohnzimmer Musik
==================================================
Identifier: zigbee:0013A20041C4ABCD:1:on
Typ: HAN_FUN
Status: AUS ✗
==================================================

Teste Schalten...

Status: AUS
Schalte AN...
Neuer Status: AN ✓

✓ Test erfolgreich!

TEIL 4: Integration in Keyword Spotting

4.1 Modifizierte keyword_spotting.py

Beginne mit deiner bisherigen keyword_spotting.py und ersetze die Konfiguration:

nano ~/voice_assistant/keyword_spotting.py

Ersetze die Konfiguration:

from fritzconnection.lib.homeauto import FritzHomeAutomation

class Config:
    # ... Audio-Config wie zuvor ...
    
    # === FritzBox Smart Home (Zigbee) ===
    FRITZBOX_IP = "192.168.178.1"
    FRITZBOX_USER = "raspi_home"
    FRITZBOX_PASSWORD = "TreTakt2025Pi"
    
    # === Deine IKEA TRETAKT Steckdosen ===
    TRETAKT_DEVICES = {
        "musik": {
            "identifier": "zigbee:0013A20041C4ABCD:1:on",  # ← DEINE ID!
            "name": "Wohnzimmer Musik",
        },
        "stopp": {
            "identifier": "zigbee:0013A20041C4ABCE:1:on",  # ← DEINE ID!
            "name": "Küche Musik",
        },
        "licht": {
            "identifier": "zigbee:0013A20041C4ABCF:1:on",  # ← DEINE ID!
            "name": "Flur Licht",
        },
    }
    
    # Keywords (wie zuvor)
    KEYWORDS = {
        "musik": {
            "sound": "music.wav",
            "action": "toggle_tretakt",
            "device": "musik",
            "confidence": 0.65,
        },
        "stopp": {
            "sound": "stopped.wav",
            "action": "toggle_tretakt",
            "device": "stopp",
            "confidence": 0.70,
        },
        "licht": {
            "sound": "light.wav",
            "action": "toggle_tretakt",
            "device": "licht",
            "confidence": 0.68,
        },
    }

4.2 FritzBox-Klasse für Zigbee hinzufügen

Füge diese Klasse VOR KeywordSpotter ein:

# ============================================================================
# FRITZBOX ZIGBEE SMART HOME (TRETAKT)
# ============================================================================

class FritzBoxZigbee:
    """Steuere IKEA TRETAKT Steckdosen über FritzBox"""
    
    def __init__(self):
        """Initialisiere FritzBox Verbindung"""
        logger.info("Initialisiere FritzBox Zigbee-Verbindung...")
        try:
            self.fha = FritzHomeAutomation(
                avm_device_ip=Config.FRITZBOX_IP,
                user=Config.FRITZBOX_USER,
                password=Config.FRITZBOX_PASSWORD
            )
            logger.info("✓ FritzBox Zigbee verbunden")
        except Exception as e:
            logger.warning(f"⚠ FritzBox nicht verfügbar: {e}")
            self.fha = None
    
    def toggle_tretakt(self, device_name):
        """Schalte TRETAKT Steckdose um"""
        if not self.fha:
            logger.error("✗ FritzBox nicht verbunden")
            return False
        
        if device_name not in Config.TRETAKT_DEVICES:
            logger.error(f"✗ Gerät '{device_name}' nicht konfiguriert")
            return False
        
        try:
            device_config = Config.TRETAKT_DEVICES[device_name]
            identifier = device_config["identifier"]
            name = device_config["name"]
            
            # Hole Gerät
            device = self.fha.get_device_by_identifier(identifier)
            if not device:
                logger.error(f"✗ Gerät {name} nicht gefunden")
                return False
            
            current_state = device.state
            
            if current_state:
                logger.info(f"🔴 Schalte '{name}' AUS")
                device.turn_off()
            else:
                logger.info(f"🟢 Schalte '{name}' AN")
                device.turn_on()
            
            # Status aktualisieren
            import time
            time.sleep(0.3)
            device.update()
            new_state = device.state
            
            logger.info(f"✓ '{name}' ist jetzt: {'AN' if new_state else 'AUS'}")
            return True
        
        except Exception as e:
            logger.error(f"✗ Fehler beim Schalten: {e}")
            return False
    
    def get_status(self, device_name):
        """Hole Status einer Steckdose"""
        if not self.fha:
            return None
        
        try:
            device_config = Config.TRETAKT_DEVICES[device_name]
            device = self.fha.get_device_by_identifier(device_config["identifier"])
            if device:
                return device.state
        except:
            pass
        
        return None

4.3 ActionHandler aktualisieren

Ersetze die komplette ActionHandler Klasse:

class ActionHandler:
    """Führe erkannte Kommandos aus"""
    
    def __init__(self, sound_player, fritz_zigbee):
        self.sound_player = sound_player
        self.fritz_zigbee = fritz_zigbee
    
    def execute(self, keyword):
        """Führe Aktion aus"""
        if keyword not in Config.KEYWORDS:
            return False
        
        config = Config.KEYWORDS[keyword]
        logger.info(f"🎯 Erkannt: {keyword.upper()}")
        
        # Spiele Sound ab
        if config.get("sound"):
            self.sound_player.play_sound(config["sound"])
        
        # Führe Aktion aus
        action = config.get("action")
        
        if action == "toggle_tretakt":
            device_name = config.get("device")
            self.fritz_zigbee.toggle_tretakt(device_name)
        
        return True

4.4 VoiceControllerLite aktualisieren

Ersetze die __init__ Methode:

def __init__(self):
    logger.info("=" * 70)
    logger.info("Voice Controller - IKEA TRETAKT Integration")
    logger.info("=" * 70)
    
    try:
        self.spotter = KeywordSpotter()
        self.sound_player = SoundPlayer()
        self.fritz_zigbee = FritzBoxZigbee()  # ← NEU!
        self.action_handler = ActionHandler(self.sound_player, self.fritz_zigbee)  # Geändert
        
        self.last_detection = 0
        self.detection_cooldown = 1.0
    except Exception as e:
        logger.error(f"✗ Initialisierungsfehler: {e}")
        raise

4.5 Kompletter Code

Hier ist die komplette modifizierte keyword_spotting.py mit TRETAKT-Integration:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-

"""
Keyword Spotting für Raspberry Pi Zero 2W
+ IKEA TRETAKT Zigbee Integration via FritzBox
"""

import json
import os
import sys
import logging
import subprocess
import time
import numpy as np
import sounddevice as sd
from pathlib import Path
from collections import deque
from fritzconnection.lib.homeauto import FritzHomeAutomation

# ============================================================================
# KONFIGURATION
# ============================================================================

class Config:
    BASE_DIR = Path(__file__).parent
    SOUNDS_DIR = BASE_DIR / "sounds"
    LOGS_DIR = BASE_DIR / "logs"
    
    SAMPLERATE = 16000
    CHUNK_SIZE = 512
    CHANNELS = 1
    DEVICE_INDEX = None
    
    # === FritzBox Smart Home (Zigbee) ===
    FRITZBOX_IP = "192.168.178.1"
    FRITZBOX_USER = "raspi_home"
    FRITZBOX_PASSWORD = "TreTakt2025Pi"
    
    # === Deine IKEA TRETAKT Steckdosen ===
    TRETAKT_DEVICES = {
        "musik": {
            "identifier": "zigbee:0013A20041C4ABCD:1:on",  # ← DEINE ID!
            "name": "Wohnzimmer Musik",
        },
        "stopp": {
            "identifier": "zigbee:0013A20041C4ABCE:1:on",  # ← DEINE ID!
            "name": "Küche Musik",
        },
        "licht": {
            "identifier": "zigbee:0013A20041C4ABCF:1:on",  # ← DEINE ID!
            "name": "Flur Licht",
        },
    }
    
    # Keywords
    KEYWORDS = {
        "musik": {
            "sound": "music.wav",
            "action": "toggle_tretakt",
            "device": "musik",
            "confidence": 0.65,
        },
        "stopp": {
            "sound": "stopped.wav",
            "action": "toggle_tretakt",
            "device": "stopp",
            "confidence": 0.70,
        },
        "licht": {
            "sound": "light.wav",
            "action": "toggle_tretakt",
            "device": "licht",
            "confidence": 0.68,
        },
    }
    
    LOG_FILE = LOGS_DIR / "keyword_spotting.log"
    LOG_LEVEL = logging.INFO

# ============================================================================
# LOGGING
# ============================================================================

def setup_logging():
    """Einfaches Logging"""
    Config.LOGS_DIR.mkdir(exist_ok=True)
    
    logger = logging.getLogger("KeywordSpotter")
    logger.setLevel(Config.LOG_LEVEL)
    
    fh = logging.FileHandler(Config.LOG_FILE)
    fh.setLevel(Config.LOG_LEVEL)
    
    ch = logging.StreamHandler()
    ch.setLevel(Config.LOG_LEVEL)
    
    formatter = logging.Formatter(
        '%(asctime)s - %(levelname)s - %(message)s',
        datefmt='%Y-%m-%d %H:%M:%S'
    )
    fh.setFormatter(formatter)
    ch.setFormatter(formatter)
    
    logger.addHandler(fh)
    logger.addHandler(ch)
    
    return logger

logger = setup_logging()

# ============================================================================
# AUDIO-GERÄTE
# ============================================================================

def find_respeaker_device():
    """Finde ReSpeaker"""
    logger.info("Suche ReSpeaker...")
    try:
        for index, name in enumerate(sd.query_devices()):
            if isinstance(name, dict):
                device_name = name.get('name', '')
            else:
                device_name = str(name)
            
            if 'seeed' in device_name.lower():
                logger.info(f"✓ ReSpeaker gefunden: Index {index}")
                return index
    except:
        pass
    
    logger.warning("⚠ ReSpeaker nicht gefunden")
    return None

# ============================================================================
# AKUSTISCHE FINGERPRINTS
# ============================================================================

class AudioFingerprint:
    """Extrahiere akustische Features"""
    
    @staticmethod
    def extract_features(audio_chunk):
        """Extrahiere Features"""
        audio = np.array(audio_chunk, dtype=np.float32) / 32768.0
        
        zcr = np.mean(np.abs(np.diff(np.sign(audio))))
        energy = np.sqrt(np.mean(audio ** 2))
        
        fft = np.abs(np.fft.fft(audio[:512]))
        freq_energy = [
            np.sum(fft[0:50]),
            np.sum(fft[50:150]),
            np.sum(fft[150:256]),
        ]
        
        return np.array([zcr, energy] + freq_energy, dtype=np.float32)
    
    @staticmethod
    def compare_fingerprints(fp1, fp2):
        """Vergleiche zwei Fingerprints"""
        fp1_norm = (fp1 - np.mean(fp1)) / (np.std(fp1) + 1e-6)
        fp2_norm = (fp2 - np.mean(fp2)) / (np.std(fp2) + 1e-6)
        
        similarity = np.dot(fp1_norm, fp2_norm) / (
            np.linalg.norm(fp1_norm) * np.linalg.norm(fp2_norm) + 1e-6
        )
        
        similarity = (similarity + 1.0) / 2.0
        return max(0.0, min(1.0, similarity))

# ============================================================================
# REFERENCE FINGERPRINTS
# ============================================================================

class ReferenceDatabase:
    """Speichert Reference-Fingerprints"""
    
    def __init__(self):
        self.db_file = Config.BASE_DIR / "reference_fingerprints.npy"
        self.fingerprints = {}
        self.load_or_create()
    
    def load_or_create(self):
        """Lade oder erstelle Fingerprints"""
        if self.db_file.exists():
            logger.info("Lade Reference-Fingerprints...")
            try:
                data = np.load(self.db_file, allow_pickle=True).item()
                self.fingerprints = data
                logger.info(f"✓ {len(self.fingerprints)} Keywords geladen")
            except Exception as e:
                logger.warning(f"Konnte nicht laden: {e}")
                self.create_default_fingerprints()
        else:
            logger.info("Erstelle Default-Fingerprints...")
            self.create_default_fingerprints()
    
    def create_default_fingerprints(self):
        """Erstelle Default-Fingerprints"""
        logger.warning("⚠ Benutze prepare_training.py für Training!")
        
        self.fingerprints = {
            "musik": np.array([0.05, 0.3, 100, 500, 200], dtype=np.float32),
            "stopp": np.array([0.02, 0.2, 150, 400, 300], dtype=np.float32),
            "licht": np.array([0.04, 0.25, 120, 450, 250], dtype=np.float32),
        }
        
        self.save()
    
    def save(self):
        """Speichere Fingerprints"""
        try:
            np.save(self.db_file, self.fingerprints)
            logger.info(f"✓ Reference-Fingerprints gespeichert")
        except Exception as e:
            logger.error(f"Fehler beim Speichern: {e}")

# ============================================================================
# FRITZBOX ZIGBEE (TRETAKT) ← NEU
# ============================================================================

class FritzBoxZigbee:
    """Steuere IKEA TRETAKT Steckdosen über FritzBox"""
    
    def __init__(self):
        """Initialisiere FritzBox Verbindung"""
        logger.info("Initialisiere FritzBox Zigbee-Verbindung...")
        try:
            self.fha = FritzHomeAutomation(
                avm_device_ip=Config.FRITZBOX_IP,
                user=Config.FRITZBOX_USER,
                password=Config.FRITZBOX_PASSWORD
            )
            logger.info("✓ FritzBox Zigbee verbunden")
        except Exception as e:
            logger.warning(f"⚠ FritzBox nicht verfügbar: {e}")
            self.fha = None
    
    def toggle_tretakt(self, device_name):
        """Schalte TRETAKT Steckdose um"""
        if not self.fha:
            logger.error("✗ FritzBox nicht verbunden")
            return False
        
        if device_name not in Config.TRETAKT_DEVICES:
            logger.error(f"✗ Gerät '{device_name}' nicht konfiguriert")
            return False
        
        try:
            device_config = Config.TRETAKT_DEVICES[device_name]
            identifier = device_config["identifier"]
            name = device_config["name"]
            
            device = self.fha.get_device_by_identifier(identifier)
            if not device:
                logger.error(f"✗ Gerät {name} nicht gefunden")
                return False
            
            current_state = device.state
            
            if current_state:
                logger.info(f"🔴 Schalte '{name}' AUS")
                device.turn_off()
            else:
                logger.info(f"🟢 Schalte '{name}' AN")
                device.turn_on()
            
            time.sleep(0.3)
            device.update()
            new_state = device.state
            
            logger.info(f"✓ '{name}' ist jetzt: {'AN' if new_state else 'AUS'}")
            return True
        
        except Exception as e:
            logger.error(f"✗ Fehler beim Schalten: {e}")
            return False

# ============================================================================
# KEYWORD SPOTTER
# ============================================================================

class KeywordSpotter:
    """Höre nach Keywords"""
    
    def __init__(self):
        logger.info("Initialisiere Keyword Spotter...")
        
        Config.DEVICE_INDEX = find_respeaker_device()
        self.ref_db = ReferenceDatabase()
        
        self.stream = None
        self.is_running = False
        self.buffer = deque(maxlen=Config.SAMPLERATE)
    
    def audio_callback(self, indata, frames, time_info, status):
        """Callback beim Audio-Input"""
        if status:
            logger.warning(f"Audio-Status: {status}")
        
        audio_data = indata[:, 0]
        for sample in audio_data:
            self.buffer.append(int(sample * 32767))
    
    def start(self):
        """Starte Audio-Listening"""
        logger.info("Starte Audio-Listening...")
        try:
            self.stream = sd.InputStream(
                samplerate=Config.SAMPLERATE,
                blocksize=Config.CHUNK_SIZE,
                channels=Config.CHANNELS,
                device=Config.DEVICE_INDEX,
                callback=self.audio_callback
            )
            self.stream.start()
            self.is_running = True
            logger.info("✓ Audio-Listening aktiv")
        except Exception as e:
            logger.error(f"Fehler beim Starten: {e}")
            raise
    
    def stop(self):
        """Stoppe Audio-Listening"""
        logger.info("Stoppe Audio-Listening...")
        if self.stream:
            self.stream.stop()
            self.stream.close()
        self.is_running = False
    
    def detect_keywords(self):
        """Erkenne Keywords"""
        if len(self.buffer) < Config.SAMPLERATE:
            return None, 0
        
        audio_chunk = list(self.buffer)
        current_fp = AudioFingerprint.extract_features(audio_chunk)
        
        best_keyword = None
        best_confidence = 0
        
        for keyword, threshold_config in Config.KEYWORDS.items():
            ref_fp = self.ref_db.fingerprints.get(keyword)
            
            if ref_fp is None:
                continue
            
            similarity = AudioFingerprint.compare_fingerprints(current_fp, ref_fp)
            required_threshold = threshold_config.get("confidence", 0.7)
            
            if similarity > best_confidence and similarity >= required_threshold:
                best_keyword = keyword
                best_confidence = similarity
        
        return best_keyword, best_confidence

# ============================================================================
# SOUND-AUSGABE
# ============================================================================

class SoundPlayer:
    """Spiele Sounds ab"""
    
    def __init__(self):
        self.sounds_dir = Config.SOUNDS_DIR
        self.sounds_dir.mkdir(exist_ok=True)
    
    def play_sound(self, filename):
        """Spiele Sound ab"""
        sound_path = self.sounds_dir / filename
        
        if not sound_path.exists():
            logger.warning(f"⚠ Sound nicht gefunden: {filename}")
            return False
        
        try:
            logger.info(f"♪ Spiele Sound ab: {filename}")
            subprocess.run(
                ['aplay', '-D', 'hw:1,0', str(sound_path)],
                check=True,
                capture_output=True,
                timeout=10
            )
            return True
        except Exception as e:
            logger.error(f"✗ Fehler beim Abspielen: {e}")
            return False

# ============================================================================
# AKTION-HANDLER ← GEÄNDERT
# ============================================================================

class ActionHandler:
    """Führe Aktionen aus"""
    
    def __init__(self, sound_player, fritz_zigbee):
        self.sound_player = sound_player
        self.fritz_zigbee = fritz_zigbee
    
    def execute(self, keyword):
        """Führe Aktion aus"""
        if keyword not in Config.KEYWORDS:
            return False
        
        config = Config.KEYWORDS[keyword]
        logger.info(f"🎯 Erkannt: {keyword.upper()}")
        
        # Spiele Sound ab
        if config.get("sound"):
            self.sound_player.play_sound(config["sound"])
        
        # Führe Aktion aus
        action = config.get("action")
        
        if action == "toggle_tretakt":
            device_name = config.get("device")
            self.fritz_zigbee.toggle_tretakt(device_name)
        
        return True

# ============================================================================
# HAUPTPROGRAMM ← GEÄNDERT
# ============================================================================

class VoiceControllerLite:
    """Hauptprogramm"""
    
    def __init__(self):
        logger.info("=" * 70)
        logger.info("Voice Controller - IKEA TRETAKT Integration")
        logger.info("=" * 70)
        
        try:
            self.spotter = KeywordSpotter()
            self.sound_player = SoundPlayer()
            self.fritz_zigbee = FritzBoxZigbee()  # ← NEU!
            self.action_handler = ActionHandler(self.sound_player, self.fritz_zigbee)  # Geändert
            
            self.last_detection = 0
            self.detection_cooldown = 1.0
        except Exception as e:
            logger.error(f"✗ Initialisierungsfehler: {e}")
            raise
    
    def run(self):
        """Hauptschleife"""
        logger.info("Starte Hauptschleife...")
        
        try:
            self.spotter.start()
            
            detection_count = 0
            
            while True:
                try:
                    keyword, confidence = self.spotter.detect_keywords()
                    
                    if keyword and confidence > 0.5:
                        current_time = time.time()
                        if current_time - self.last_detection > self.detection_cooldown:
                            detection_count += 1
                            logger.info(
                                f"[#{detection_count}] ✓ {keyword.upper()} "
                                f"({confidence:.1%})"
                            )
                            
                            self.action_handler.execute(keyword)
                            self.last_detection = current_time
                    
                    time.sleep(0.1)
                
                except KeyboardInterrupt:
                    logger.info("\n⏹ Unterbrochen")
                    break
                except Exception as e:
                    logger.error(f"Fehler: {e}")
                    time.sleep(1)
        
        finally:
            self.spotter.stop()
            logger.info("✓ Voice Controller beendet")

# ============================================================================
# EINSTIEGSPUNKT
# ============================================================================

if __name__ == "__main__":
    try:
        controller = VoiceControllerLite()
        controller.run()
    except KeyboardInterrupt:
        logger.info("Beendet")
        sys.exit(0)
    except Exception as e:
        logger.error(f"✗ Fehler: {e}", exc_info=True)
        sys.exit(1)

Speichere die Datei (Ctrl+X, Y, Enter).

4.6 Konfiguration anpassen

nano ~/voice_assistant/keyword_spotting.py

Trage deine Daten ein:

FRITZBOX_IP = "192.168.178.1"           # Deine FritzBox IP
FRITZBOX_USER = "raspi_home"            # Dein Benutzer
FRITZBOX_PASSWORD = "TreTakt2025Pi"     # Dein Passwort

TRETAKT_DEVICES = {
    "musik": {
        "identifier": "zigbee:0013A20041C4ABCD:1:on",  # ← DEINE ID!
        "name": "Wohnzimmer Musik",
    },
    # ... weitere Geräte ...
}

4.7 Starten und testen

cd ~/voice_assistant
python3 keyword_spotting.py

Jetzt:

  1. Sprich: "musik" → TRETAKT schaltet! ✓
  2. Sprich: "stopp" → TRETAKT schaltet aus!
  3. Sprich: "licht" → Andere TRETAKT schaltet!

VORTEIL TRETAKT vs. FRITZ!DECT

Feature TRETAKT FRITZ!DECT
Preis €8 €25-35
Netzwerk-Erweiterung Ja (Zigbee Mesh) Nein (nur DECT)
Mit Hue kompatibel Ja Nein
Weniger Strom Ja Etwas mehr
Physischer Schalter Ja Ja
Reichweite 100m 300m

Du sparst damit €17-27 pro Steckdose! 🎉


TROUBLESHOOTING

Problem: "Identifier nicht gefunden"

# Zeige alle Zigbee-Geräte mit Identifiers
python3 << 'EOF'
from fritzconnection.lib.homeauto import FritzHomeAutomation

fha = FritzHomeAutomation(
    avm_device_ip="192.168.178.1",
    user="raspi_home",
    password="TreTakt2025Pi"
)

for device in fha.get_devices():
    if "tretakt" in device.name.lower() or "zigbee" in str(device.identifier):
        print(f"Name: {device.name}")
        print(f"Identifier: {device.identifier}\n")
EOF

Problem: "FritzBox erkundet TRETAKT nicht"

  1. Steckdose Reset (8-10 Sekunden Knopf drücken)
  2. LED sollte schnell blinken
  3. In FritzBox "Neues Gerät anmelden" starten
  4. Warten (30 Sekunden)

Problem: "Schalten funktioniert nicht"

  1. Test ob Verbindung funktioniert: python3 test_tretakt.py
  2. Überprüfe Passwort in FritzBox
  3. Überprüfe Identifier (muss exakt stimmen!)

ZUSAMMENFASSUNG

IKEA TRETAKT (€8 pro Stück)
FritzBox Zigbee Integration
Sprachgesteuert über Pi Zero 2W
Offline im heimischen LAN
Super günstig und einfach

Der perfekte Start in Smart Home! 🏠

Viel Erfolg! 🚀