221 lines
6.6 KiB
Plaintext
221 lines
6.6 KiB
Plaintext
# 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
|
|
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()
|
|
|
|
def led_worker():
|
|
while True:
|
|
cmd = led_queue.get()
|
|
if cmd is None:
|
|
break # arrêt propre
|
|
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()
|
|
|
|
# --- LOGGING ---
|
|
def log(*args):
|
|
if DEBUG:
|
|
print(*args)
|
|
|
|
# --- BOUTON MUTE ---
|
|
def toggle_mute(ch):
|
|
etat['mute'] = not etat['mute']
|
|
led_queue.put({'action': 'set', 'color': 'R' if etat['mute'] else 'O'})
|
|
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)
|
|
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
|
|
|
|
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 not etat['parle'] and not audio_q.full():
|
|
audio_q.put(bytes(indata))
|
|
|
|
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):
|
|
etat['parle'] = True
|
|
etat['dernier_tick'] = time.time()
|
|
audio = b''.join(voice.synthesize_stream_raw(text))
|
|
sd_stream.write(np.frombuffer(audio, dtype=np.int16))
|
|
etat['parle'] = False
|
|
|
|
# --- CLEANUP ---
|
|
def handle_exit(sig, frame):
|
|
led_queue.put(None)
|
|
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']:
|
|
led_queue.put({'action': 'set', 'color': 'BL'}) # blanc = veille
|
|
etat['veille'] = True
|
|
time.sleep(5)
|
|
|
|
# --- MAIN LOOP ---
|
|
if __name__ == '__main__':
|
|
print("Init…")
|
|
led_queue.put({'action': 'set', 'color': 'O'}) # Orange init
|
|
|
|
if not test_ally():
|
|
led_queue.put({'action': 'blink', 'rgb': (255, 0, 0), 'count': 3, 'duration': 0.3})
|
|
log("ERREUR: API Ally indisponible")
|
|
sys.exit(1)
|
|
|
|
Init()
|
|
time.sleep(0.5)
|
|
sd_stream.start()
|
|
stream.start()
|
|
threading.Thread(target=led_worker, daemon=True).start()
|
|
threading.Thread(target=surveiller_inactivite, daemon=True).start()
|
|
led_queue.put({'action': 'set', 'color': ''})
|
|
|
|
while True:
|
|
if etat['mute'] or etat['parle']:
|
|
time.sleep(0.05)
|
|
continue
|
|
|
|
led_queue.put({'action': 'set', 'color': 'VI'}) # écoute active
|
|
|
|
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)
|
|
led_queue.put({'action': 'blink', 'rgb': (255, 105, 180), 'count': 2, 'duration': 0.5})
|
|
resp = ally(txt)
|
|
log("R>", resp)
|
|
if resp:
|
|
led_queue.put({'action': 'wave', 'rgb': (255, 105, 180)})
|
|
led_queue.put({'action': 'set', 'color': 'B'})
|
|
piper_talk(resp, voice, sd_stream)
|
|
else:
|
|
led_queue.put({'action': 'blink', 'rgb': (255, 0, 0), 'count': 3, 'duration': 0.3})
|
|
time.sleep(min(0.5, len(resp)*0.01))
|