# Init => Orange # Ecoute => Violet # Mute => Rouge # Reflechi(Ally) => Cligno Rose # Repond => Bleu # Erreur cligno rouge from driver_led import APA102 import RPi.GPIO as GPIO import time import queue import sounddevice as sd import numpy as np import json import requests, urllib3 import signal, sys import threading # --- CONFIGURATION --- #DEBUG = True DEBUG = False url_ally = "https://192.168.1.12:8000/mic" piper_model_path = "/data/piper_model/fr_FR-siwis-low.onnx" vosk_model_path = "vosk-model/vosk-model-small-fr-0.22" num_leds = 12 BUTTON = 17 # Initialisation GPIO et LED GPIO.setmode(GPIO.BCM) GPIO.setup(BUTTON, GPIO.IN) urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning) led = APA102(num_led=num_leds) # --- ETAT GLOBAL --- etat = { 'mute': False, 'parle': False, 'dernier_tick': time.time(), 'veille': False } # --- LED QUEUE ET THREAD --- led_queue = queue.Queue(maxsize=50) def led_worker(): while True: try: cmd = led_queue.get() if cmd is None: break action = cmd.get('action') color = cmd.get('color', '') if action == 'set': rgb_map = { 'R': (255, 0, 0), 'V': (0, 255, 0), 'B': (0, 0, 255), 'J': (255, 255, 0), 'VI': (128, 0, 128), 'RO': (255, 105, 180), 'O': (255, 165, 0), 'BL': (255, 255, 255), '': (0, 0, 0) } rgb = rgb_map.get(color.upper(), (0, 0, 0)) for i in range(num_leds): led.set_pixel(i, *rgb) led.show() elif action == 'blink': rgb = cmd.get('rgb', (255, 0, 0)) count = cmd.get('count', 3) duration = cmd.get('duration', 0.3) for _ in range(count): for i in range(num_leds): led.set_pixel(i, *rgb) led.show() time.sleep(duration) for i in range(num_leds): led.set_pixel(i, 0, 0, 0) led.show() time.sleep(duration) elif action == 'wave': rgb = cmd.get('rgb', (255, 105, 180)) for _ in range(2): for i in range(num_leds): led.set_pixel(i, *rgb) led.show() time.sleep(0.05) led.set_pixel(i, 0, 0, 0) led.show() except queue.Empty: continue # --- LOGGING --- def log(*args): if DEBUG: print(*args) # --- BOUTON MUTE --- def toggle_mute(ch): etat['mute'] = not etat['mute'] try: led_queue.put_nowait({'action': 'set', 'color': 'R' if etat['mute'] else 'O'}) except queue.Full: pass log("Mute:", etat['mute']) etat['dernier_tick'] = time.time() GPIO.add_event_detect(BUTTON, GPIO.FALLING, callback=toggle_mute, bouncetime=300) # --- Ally API --- def test_ally(): try: r = requests.post(url_ally, data={"texto": "test"}, verify=False, timeout=2) return r.status_code == 200 except: return False def ally(text): try: r = requests.post(url_ally, data={"texto": text}, verify=False) log("Q>", text) return r.json().get("ora", "") if r.status_code == 200 else "" except Exception as e: log("Ally exception", e) return "" # --- INIT TTS/STT --- rec = None audio_q = None stream = None sd_stream = None voice = None stream_started = False def Init(): global rec, audio_q, stream, sd_stream, voice import vosk model = vosk.Model(vosk_model_path) rec = vosk.KaldiRecognizer(model, 16000) audio_q = queue.Queue(maxsize=10) def audio_callback(indata, frames, time_info, status): if status: log(status) if etat['mute'] or etat['parle'] or audio_q.full(): return audio_q.put(bytes(indata)) global stream stream = sd.RawInputStream(samplerate=16000, blocksize=4096, device=None, dtype='int16', channels=1, callback=audio_callback) from piper.voice import PiperVoice voice = PiperVoice.load(piper_model_path) sd_stream = sd.OutputStream(samplerate=voice.config.sample_rate, channels=1, dtype='int16') # --- PAROLE --- def piper_talk(text, voice, sd_stream): global stream_started etat['parle'] = True etat['dernier_tick'] = time.time() try: if stream_started: stream.stop() stream_started = False except Exception as e: log("Erreur stream.stop():", e) try: audio = b''.join(voice.synthesize_stream_raw(text)) sd_stream.write(np.frombuffer(audio, dtype=np.int16)) except Exception as e: log("Erreur TTS:", e) time.sleep(0.3) try: stream.start() stream_started = True except Exception as e: log("Erreur stream.start():", e) while not audio_q.empty(): try: audio_q.get_nowait() except queue.Empty: break etat['parle'] = False # --- CLEANUP --- def handle_exit(sig, frame): try: led_queue.put(None) except: pass for i in range(num_leds): led.set_pixel(i, 0, 0, 0) led.show() led.cleanup() GPIO.cleanup() try: if stream: stream.stop() if sd_stream: sd_stream.stop() except: pass sys.exit(0) signal.signal(signal.SIGINT, handle_exit) signal.signal(signal.SIGTERM, handle_exit) # --- VEILLE AUTOMATIQUE --- def surveiller_inactivite(): while True: if not etat['mute'] and not etat['parle']: if time.time() - etat['dernier_tick'] > 300 and not etat['veille']: try: led_queue.put_nowait({'action': 'set', 'color': 'BL'}) except queue.Full: pass etat['veille'] = True time.sleep(5) # --- MAIN LOOP --- if __name__ == '__main__': print("Init…") try: led_queue.put_nowait({'action': 'set', 'color': 'O'}) except queue.Full: pass if not test_ally(): try: led_queue.put_nowait({'action': 'blink', 'rgb': (255, 0, 0), 'count': 3, 'duration': 0.3}) except queue.Full: pass log("ERREUR: API Ally indisponible") sys.exit(1) Init() time.sleep(0.5) sd_stream.start() stream.start() stream_started = True threading.Thread(target=led_worker, daemon=True).start() threading.Thread(target=surveiller_inactivite, daemon=True).start() try: led_queue.put_nowait({'action': 'set', 'color': ''}) except queue.Full: pass while True: if etat['mute'] or etat['parle']: # purge de la file si mute while not audio_q.empty(): try: audio_q.get_nowait() except queue.Empty: break time.sleep(0.1) continue try: led_queue.put_nowait({'action': 'set', 'color': 'VI'}) except queue.Full: pass try: data = audio_q.get(timeout=1) except queue.Empty: continue if rec.AcceptWaveform(data): txt = json.loads(rec.Result()).get("text", "") if txt: etat['dernier_tick'] = time.time() etat['veille'] = False log("Q>", txt) try: led_queue.put_nowait({'action': 'blink', 'rgb': (255, 105, 180), 'count': 2, 'duration': 0.5}) except queue.Full: pass resp = ally(txt) log("R>", resp) if resp: try: led_queue.put_nowait({'action': 'wave', 'rgb': (255, 105, 180)}) led_queue.put_nowait({'action': 'set', 'color': 'B'}) except queue.Full: pass piper_talk(resp, voice, sd_stream) else: try: led_queue.put_nowait({'action': 'blink', 'rgb': (255, 0, 0), 'count': 3, 'duration': 0.3}) except queue.Full: pass time.sleep(min(0.5, len(resp) * 0.01))