carinha-control-panel/cerebro.py

"""Cérebro local: decisões reais do Qwen, limitadas a expressões do personagem."""
import json
import os
from pathlib import Path
import threading
import time
from urllib.request import Request,urlopen
from comum import gravar_atomico, ler_json

GESTOS=('piscar','piscadinha','feliz','surpreso','coracao','olhar_esq','olhar_dir','olhar_baixo','acenar','esconder','pensar',
        'espiar','bocejar','confuso')
EVENTOS=('som','movimento','saudacao','sozinho')
# O Qwen pequeno ignora "varie" e repetia 'esconder' sem parar. Cada evento só oferece gestos que
# fazem sentido (barulho: susto ou olhar; movimento: olhar e acenar) e os dois últimos ficam de fora.
POR_EVENTO={
    'som':('surpreso','olhar_esq','olhar_dir','pensar','piscar','confuso'),
    'movimento':('olhar_esq','olhar_dir','acenar','feliz','piscadinha','surpreso','espiar'),
    'saudacao':('acenar','feliz','coracao','piscadinha'),
    'sozinho':('pensar','olhar_baixo','olhar_esq','olhar_dir','bocejar','piscar'),
}
INTERVALO=45            # s mínimos entre duas reações da IA
INTERVALO_SOZINHO=180   # s sem nada acontecendo até reagir a "sozinho"
def validar_intencao(data):
    if not isinstance(data,dict) or set(data)!= {'gesto'} or data['gesto'] not in GESTOS:
        raise ValueError('Intenção fora da lista de expressões')
    return data['gesto']

class Cerebro:
    def __init__(self,path,controle,enviar):
        self.path=Path(path);self.controle=controle;self.enviar=enviar
        self.endpoint=os.environ.get('CARINHA_IA_URL','http://127.0.0.1:8101')
        # O llama-server do S10 fica na rede: com token, ele exige a mesma chave (--api-key).
        self.chave=os.environ.get('CARINHA_IA_CHAVE') or os.environ.get('CARINHA_TOKEN','')
        self.executor='S10' if '192.168.1.16' in self.endpoint else 'Orange Pi'
        self.lock=threading.Lock();self.ativo=False;self.pending=None;self.busy=False
        self.conversando=False;self.history=[];self.result=None;self.error='';self.last=0;self.generation=0
        self.publicar=lambda tipo,dados:None
        self.vida_ligada=lambda:False      # com o modo vida ligado, ficar sozinho é com ele
        try:self.ativo=json.loads(self.path.read_text())['ativo'] is True
        except (OSError,ValueError,KeyError,TypeError):pass
        threading.Thread(target=self.loop,daemon=True).start()
    def cabecalhos(self):
        h={'Content-Type':'application/json'}
        if self.chave:h['Authorization']='Bearer '+self.chave
        return h
    def config(self,ativo):
        if type(ativo) is not bool:raise ValueError('ativo deve ser booleano')
        with self.lock:
            gravar_atomico(self.path,dict(ativo=ativo))
            self.ativo=ativo;self.generation+=1;self.pending=None
        status=self.status();self.publicar('ia',status)
        return status
    def status(self):
        with self.lock:return dict(ativo=self.ativo,ocupado=self.busy,modelo='Qwen3-0.6B Q4_K_M',executor=self.executor,ultima=self.result,erro=self.error)
    def event(self,event):
        if event not in EVENTOS:raise ValueError('Evento desconhecido')
        with self.lock:
            if not self.ativo:raise ValueError('Ative a personalidade primeiro')
            if self.busy or self.pending or self.conversando:raise ValueError('IA processando; aguarde')
            self.pending=event
        return dict(ok=True)
    def opcoes(self,event):
        recentes=set(self.history[-2:])
        return [g for g in POR_EVENTO.get(event,GESTOS) if g not in recentes] or list(POR_EVENTO.get(event,GESTOS))
    def decide(self,event):
        opcoes=self.opcoes(event)
        payload=dict(model='local',messages=[dict(role='system',content='Você é Carinha, um pequeno robô curioso, gentil e expressivo, inspirado em WALL-E. Escolha uma expressão adequada ao evento. Responda somente JSON.'),dict(role='user',content=json.dumps(dict(evento=event),ensure_ascii=False)+' /no_think')],max_tokens=32,temperature=.9,chat_template_kwargs=dict(enable_thinking=False),response_format=dict(type='json_schema',json_schema=dict(name='intencao',strict=True,schema=dict(type='object',properties=dict(gesto=dict(type='string',enum=opcoes)),required=['gesto'],additionalProperties=False))))
        request=Request(self.endpoint+'/v1/chat/completions',data=json.dumps(payload).encode(),headers=self.cabecalhos())
        start=time.monotonic()
        with urlopen(request,timeout=90) as response:data=json.load(response)
        content=data['choices'][0]['message']['content']
        gesto=validar_intencao(json.loads(content))
        if gesto not in opcoes:raise ValueError('Gesto repetido ou fora do evento')
        return dict(evento=event,gesto=gesto,segundos=round(time.monotonic()-start,2),horario=time.time())
    def loop(self):
        while True:
            time.sleep(1)
            with self.lock:
                if not self.ativo or self.conversando:continue
                event=self.pending;self.pending=None;generation=self.generation
            # Só observar sensores enquanto o app está online e acordado.
            with self.controle.lock:
                online=time.monotonic()-self.controle.status_at<5
                state=dict(self.controle.estado)
            # ocupado com uma atividade (lendo, jogando...): não interrompe com gesto
            livre=online and not state.get('dormindo') and not state.get('atividade')
            if not event and livre and time.monotonic()-self.last>INTERVALO:
                if state.get('microfone_ativo') and state.get('microfone',0)>.35:event='som'
                elif state.get('camera_ativa') and state.get('movimento',0)>.05:event='movimento'
                elif time.monotonic()-self.last>INTERVALO_SOZINHO and not self.vida_ligada():event='sozinho'
            if not event:continue
            with self.lock:
                if self.conversando:continue
                self.busy=True
            try:
                result=self.decide(event)
                with self.lock:
                    if self.ativo and generation==self.generation:
                        self.result=result;self.error='';self.history.append(result['gesto']);self.history=self.history[-6:]
                        with self.controle.lock:
                            if time.monotonic()-self.controle.status_at<5 and not self.controle.estado.get('dormindo') and not self.controle.estado.get('atividade'):
                                self.enviar(result['gesto'])
            except Exception as e:
                with self.lock:self.error=str(e)[:200]
            finally:
                self.last=time.monotonic()
                with self.lock:self.busy=False
                self.publicar('ia',self.status())
    def rotas(self,h,method):
        route=h.path.split('?',1)[0]
        if route=='/ia/modelo' and method=='GET':
            model=Path('/opt/carinha-ai/models/qwen3-0.6b-q4.gguf')
            if not model.exists():h.enviar(json.dumps(dict(erro='Modelo indisponível')),codigo=404);return True
            h.send_response(200);h.send_header('Content-Type','application/octet-stream');h.send_header('Content-Length',str(model.stat().st_size));h.end_headers()
            try:
                with model.open('rb') as source:
                    while True:
                        chunk=source.read(65536)
                        if not chunk:break
                        h.wfile.write(chunk)
            except (BrokenPipeError,ConnectionResetError):pass
            return True
        if route not in ('/ia/status','/ia/config','/ia/evento'):return False
        try:
            if method=='GET' and route=='/ia/status':result=self.status()
            elif method=='POST':
                data=ler_json(h,2048)
                if route=='/ia/config':result=self.config(data.get('ativo'))
                elif route=='/ia/evento':result=self.event(data.get('evento'))
                else:raise ValueError('Rota inválida')
            else:raise ValueError('Método inválido')
            h.enviar(json.dumps(result,ensure_ascii=False))
        except (ValueError,TypeError) as e:h.enviar(json.dumps(dict(erro=str(e))),codigo=400)
        return True