hal90/interrupt_detector.py at master · neal3000/hal90 · GitHub

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
"""
Interrupt Detector - Lightweight listener for stop commands during TTS
Uses PocketSphinx with limited vocabulary for low CPU overhead
"""
import logging
import threading
import time
from pocketsphinx import LiveSpeech

logger = logging.getLogger(__name__)


class InterruptDetector:
    """Detects interrupt commands (stop, cancel, etc.) during TTS playback"""

    def __init__(self, sample_rate=16000, audio_device=None):
        """
        Initialize interrupt detector

        Args:
            sample_rate: Audio sample rate (16000 recommended for PocketSphinx)
            audio_device: Audio input device index (None for default)
        """
        self.sample_rate = sample_rate
        self.audio_device = audio_device
        self.interrupted = False
        self.monitoring = False
        self.monitor_thread = None

        # Interrupt keywords - use simple keyphrase
        # Note: Using single words works better with PocketSphinx keyphrase spotting
        self.keyphrase = "stop"  # Primary interrupt keyword

        logger.info(f"InterruptDetector created with keyphrase: '{self.keyphrase}'")

    def start_monitoring(self):
        """Start monitoring for interrupt commands in background thread"""
        if self.monitoring:
            logger.warning("Already monitoring for interrupts")
            return

        self.interrupted = False
        self.monitoring = True

        # Start monitoring thread
        self.monitor_thread = threading.Thread(target=self._monitor_loop, daemon=True)
        self.monitor_thread.start()

        logger.info(f"🎤 Started interrupt monitoring (listening for '{self.keyphrase}')")

    def stop_monitoring(self):
        """Stop monitoring for interrupt commands"""
        self.monitoring = False

        if self.monitor_thread and self.monitor_thread.is_alive():
            self.monitor_thread.join(timeout=2.0)

        logger.info("🛑 Stopped interrupt monitoring")

    def reset(self):
        """Reset interrupt flag"""
        self.interrupted = False

    def is_interrupted(self):
        """Check if interrupt command was detected"""
        return self.interrupted

    def _monitor_loop(self):
        """Main monitoring loop - runs in background thread"""
        logger.info("🔍 Interrupt detector monitor_loop started")

        try:
            # Configure PocketSphinx for interrupt detection
            # Use very sensitive threshold to catch "stop" even during TTS
            config = {
                'verbose': False,
                'audio_device': self.audio_device,
                'sampling_rate': self.sample_rate,
                'buffer_size': 2048,
                'no_search': False,
                'full_utt': False,
                # Use keyphrase spotting for "stop"
                'keyphrase': self.keyphrase,
                'kws_threshold': 1e-20  # Very sensitive threshold
            }

            logger.info(f"🔍 Interrupt detector config: device={self.audio_device}, rate={self.sample_rate}, keyphrase='{self.keyphrase}', threshold=1e-20")

            # Create LiveSpeech object
            logger.info("🔍 Creating LiveSpeech object...")
            speech = LiveSpeech(**config)
            logger.info("🔍 LiveSpeech created successfully")

            logger.info("🔍 Starting to iterate over speech phrases...")
            phrase_count = 0

            # Listen for interrupt keyword
            for phrase in speech:
                phrase_count += 1
                logger.info(f"🔍 Phrase #{phrase_count} received from LiveSpeech")

                if not self.monitoring:
                    logger.info("🔍 Monitoring stopped, breaking loop")
                    break

                detected_text = str(phrase).strip().lower()
                logger.info(f"🎤 Interrupt detector heard: '{detected_text}'")

                # Check if interrupt keyword detected
                if self.keyphrase in detected_text:
                    logger.info(f"🛑 Interrupt keyword detected: '{detected_text}'")
                    self.interrupted = True
                    break
                else:
                    logger.info(f"🔍 No match - looking for '{self.keyphrase}' in '{detected_text}'")

            logger.info(f"🔍 Exited speech loop after {phrase_count} phrases")

        except Exception as e:
            logger.error(f"❌ Interrupt detector error: {e}", exc_info=True)

        finally:
            self.monitoring = False
            logger.info("🔍 Interrupt detector monitor_loop finished")