julio/carinha-control-panel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
74ae2863a7d45d169cb505f78f66fac0ab134602
carinha-control-panel/kiosk-server.py
T
julioandClaude Opus 5.5 f7c4793dc2 feat(voz): voz de robo sintetizada no estilo dos robozinhos de cinema 
- voz-robo.js: silabas com deslize de tom, vibrato e filtro de formante
  ("uou", "ohh"), motorzinho de ruido filtrado e cliques; tudo gerado na
  hora com Web Audio, sem gravacoes nem arquivos de terceiros
- um som proprio para cada gesto, incluindo maos (motor ao subir e descer)
- balbucio de robo na resposta da conversa (mesma frase, mesmo som; sobe
  no fim quando e pergunta) e silencio dos gestos enquanto ele "fala"
- tela Rosto: ouvir a voz no PC e botao "Testar fala"; bipes antigos
  ficam como reserva

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-06 02:57:32 -03:00

493 lines
21 KiB
Python
Raw Blame History

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
kiosk-server.py — servidor do Carinha (Orange Pi, porta 8100): painel, rosto e controle.
Estado em tempo real:
GET /eventos -> Server-Sent Events (expressao, telemetria, conversa, reacao,
ia, motores, estop, aparencia, app_config, treinamento)
GET /estado -> foto de tudo para o primeiro carregamento (inclui 'seq')
Comandos:
POST /comando -> {"tipo": "gesto"|"rosto"|"cor"|"reacao"|"parar_reacao"|"recarregar", "valor": ...}
POST /parar-tudo -> E-Stop: desliga todos os servos, para reações e acalma o rosto
POST /rearmar -> libera os motores depois do E-Stop
GET /fila?aparelho=1 -> ordens pendentes para o celular (até o rosto migrar para /eventos)
Autenticação: se CARINHA_TOKEN (ambiente) ou seguranca.json {"token": ...} existir, toda
rota de controle/dados exige "Authorization: Bearer <token>" (ou ?t=<token> no EventSource
e no app). Sem token configurado o servidor fica aberto como antes.
Só usa a biblioteca padrão. Nada de instalação extra.
"""
import argparse
from collections import deque
import json
import os
import threading
import time
import uuid
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
from urllib.parse import unquote
from pathlib import Path
from comum import autorizado, carregar_token, gravar_atomico, ler_json
from controle_pc import ControlePC, versao
from eventos import Eventos
from servos import ControleServos
from braco import Braco
from cerebro import Cerebro
from conversa import Conversa
from treinamento import Treinamento
from api_config import ConfigAPI
TRAVA = threading.Lock()
# Espelho da biblioteca de rostos do carinha.html (para /faces funcionar)
ROSTOS = ["vivo", "bloco", "matriz", "lente", "eilik", "carinha", "miudo"]
# Formato do olho nos rostos novos (rosto-motor.js)
FORMATOS = ["padrao", "redondo", "pilula", "quadrado", "largo"]
# Espelho dos gestos do carinha.html (para o painel montar os botoes)
GESTOS = ["piscar", "piscar2", "olhar_esq", "olhar_dir", "olhar_baixo", "feliz",
"surpreso", "bravo", "coracao", "triste", "piscadinha", "alegre",
"dormir", "acordar", "aleatorio", "parar",
"esconder", "espiar", "acenar", "pensar", "musica"]
# Ordens internas que o rosto entende mas não viram botão.
ORDENS_INTERNAS = {"calma"}
# Aparência atual da carinha (personalizacao pelo painel). Fica gravada num
# arquivo ao lado do servidor, para sobreviver a reinicios.
APARENCIA = {"rosto": "carinha", "cor": [92, 214, 255], "formato": "padrao"}
ARQUIVO_APARENCIA = "aparencia.json"
# Arquivos servidos como estão: o rosto do celular e o rastreamento do painel.
ESTATICOS = {"/carinha.html", "/camera.html", "/reacoes-render.js", "/rastreador-worker.js", "/rosto-motor.js", "/voz-robo.js",
"/vendor/mediapipe/vision_bundle.mjs", "/vendor/mediapipe/face.tflite", "/vendor/mediapipe/hands.task",
"/vendor/mediapipe/wasm/vision_wasm_internal.js", "/vendor/mediapipe/wasm/vision_wasm_internal.wasm",
"/vendor/mediapipe/wasm/vision_wasm_nosimd_internal.js", "/vendor/mediapipe/wasm/vision_wasm_nosimd_internal.wasm"}
# Rotas que não exigem token: só o necessário para a página do rosto abrir e o app se achar.
PUBLICAS_GET = {"/app/health", "/faces", "/gestos", "/config", "/ia/modelo", "/app.apk"}
PREFIXOS_API = ("/eventos", "/estado", "/fila", "/rtc/", "/conversa/", "/treinamento/",
"/app/", "/braco/", "/servos/", "/ia/", "/api/", "/gesto/", "/face/", "/diag/")
class Fila:
"""Ordens para o celular: limitada e com validade, para não repetir ordens velhas
quando o aparelho volta depois de um tempo offline."""
def __init__(self, maximo=20, validade=10.0):
self.lock = threading.Lock()
self.itens = deque(maxlen=maximo)
self.validade = validade
def adicionar(self, ordem, agora=None):
with self.lock:
if ordem == "recarregar" and any(o == ordem for _, o in self.itens):
return
self.itens.append((agora if agora is not None else time.monotonic(), ordem))
def consumir(self, aparelho=True, agora=None):
"""O celular leva tudo; um rosto aberto no navegador (kiosk) leva tudo menos 'recarregar'."""
agora = agora if agora is not None else time.monotonic()
with self.lock:
validos = [(t, o) for t, o in self.itens if agora - t <= self.validade]
pendentes = [o for _, o in validos if aparelho or o != "recarregar"]
self.itens.clear()
if not aparelho:
self.itens.extend((t, o) for t, o in validos if o == "recarregar")
return pendentes
def ordens(self):
with self.lock:
return [o for _, o in self.itens]
def limpar(self):
with self.lock:
self.itens.clear()
FILA = Fila()
EVENTOS = Eventos()
ESTOP = {"ativo": False, "desde": None}
def enfileirar(ordem):
"""Caminho único de uma ordem para o rosto: fila do celular + evento para os painéis."""
FILA.adicionar(ordem)
EVENTOS.publicar("expressao", {"ordem": ordem})
def carregar_aparencia():
try:
with open(ARQUIVO_APARENCIA, "r", encoding="utf-8") as f:
dados = json.load(f)
if isinstance(dados, dict):
if dados.get("rosto") in ROSTOS:
APARENCIA["rosto"] = dados["rosto"]
if dados.get("formato") in FORMATOS:
APARENCIA["formato"] = dados["formato"]
cor = dados.get("cor")
if isinstance(cor, list) and len(cor) == 3:
APARENCIA["cor"] = [max(0, min(255, int(v))) for v in cor]
except Exception:
pass
def salvar_aparencia():
try:
gravar_atomico(ARQUIVO_APARENCIA, APARENCIA)
except OSError:
pass
def validar_cor(valor):
cor = [max(0, min(255, int(v))) for v in list(valor)[:3]]
if len(cor) != 3:
raise ValueError("Cor precisa de três números")
return cor
def aplicar_aparencia(dados):
"""Guarda rosto/cor e manda as ordens para a página aplicar na hora."""
ordens = []
with TRAVA:
if "rosto" in dados:
nome = str(dados["rosto"]).strip().lower()
if nome in ROSTOS:
APARENCIA["rosto"] = nome
ordens.append("rosto " + nome)
if "formato" in dados:
nome = str(dados["formato"]).strip().lower()
if nome in FORMATOS:
APARENCIA["formato"] = nome
ordens.append("formato " + nome)
if "cor" in dados:
try:
APARENCIA["cor"] = validar_cor(dados["cor"])
ordens.append("cor %d,%d,%d" % tuple(APARENCIA["cor"]))
except (TypeError, ValueError):
pass
aparencia = dict(APARENCIA)
for ordem in ordens:
enfileirar(ordem)
if ordens:
salvar_aparencia()
EVENTOS.publicar("aparencia", aparencia)
return ordens
def ordem_de_comando(tipo, valor):
"""Traduz {tipo, valor} para a ordem de texto que o rosto entende; ValueError se inválido."""
if tipo == "gesto":
nome = str(valor or "").strip().lower()
if nome not in GESTOS and nome not in ORDENS_INTERNAS:
raise ValueError("Gesto desconhecido: " + nome)
return nome
if tipo == "rosto":
nome = str(valor or "").strip().lower()
if nome not in ROSTOS:
raise ValueError("Rosto desconhecido: " + nome)
return "rosto " + nome
raise ValueError("Tipo de comando inválido")
class Handler(SimpleHTTPRequestHandler):
extensoes_map = {".html": "text/html; charset=utf-8",
".js": "text/javascript; charset=utf-8",
".mjs": "text/javascript; charset=utf-8",
".css": "text/css; charset=utf-8",
".json": "application/json; charset=utf-8",
".svg": "image/svg+xml",
".png": "image/png",
".woff2": "font/woff2",
".mp4": "video/mp4"}
def guess_type(self, caminho):
return self.extensoes_map.get(os.path.splitext(caminho)[1].lower(),
super().guess_type(caminho))
def enviar(self, corpo, tipo="application/json; charset=utf-8", codigo=200):
dados = corpo.encode("utf-8") if isinstance(corpo, str) else corpo
self.send_response(codigo)
self.send_header("Content-Type", tipo)
self.send_header("Content-Length", str(len(dados)))
self.send_header("Cache-Control", "no-store")
if codigo == 401:
self.send_header("WWW-Authenticate", 'Bearer realm="carinha"')
self.end_headers()
self.wfile.write(dados)
def json(self, dados, codigo=200):
self.enviar(json.dumps(dados, ensure_ascii=False), codigo=codigo)
def rota(self):
return self.path.split("?", 1)[0]
def acesso_negado(self, metodo):
"""True (e já respondeu 401) se a rota exige token e o pedido não trouxe o certo."""
rota = self.rota()
protegida = metodo == "POST" or (rota not in PUBLICAS_GET and rota.startswith(PREFIXOS_API))
if protegida and not autorizado(self):
self.json({"erro": "Token ausente ou inválido"}, 401)
return True
return False
def modulos(self):
s = self.server
return [getattr(s, nome) for nome in ("api_config", "treinamento", "conversa", "cerebro",
"braco", "servos", "controle") if hasattr(s, nome)]
def estado(self):
s = self.server
host = self.headers.get("Host", "").split(":")[0]
with TRAVA:
aparencia = dict(APARENCIA)
dados = dict(versao=versao(), seq=EVENTOS.seq, autenticacao=bool(getattr(s, "token", None)),
estop=dict(ESTOP), aparencia=aparencia, gestos=GESTOS, rostos=ROSTOS, formatos=FORMATOS)
if hasattr(s, "controle"):
dados.update(app=s.controle.resumo(host), app_config=s.controle.config.copy())
if hasattr(s, "servos"):
dados["motores"] = s.servos.motores()
if hasattr(s, "conversa"):
dados["conversa"] = s.conversa.status()
if hasattr(s, "cerebro"):
dados["ia"] = s.cerebro.status()
if hasattr(s, "treinamento"):
dados["reacao"] = s.treinamento.expressao
dados["treinamento"] = dict(ocupado=s.treinamento.busy, erro=s.treinamento.erro)
if hasattr(s, "api_config"):
dados["api"] = s.api_config.status()
return dados
def do_GET(self):
if self.acesso_negado("GET"):
return
for modulo in self.modulos():
if modulo.rotas(self, "GET"):
return
caminho = self.rota()
partes = [p for p in caminho.split("/") if p]
if caminho == "/eventos":
return EVENTOS.atender(self)
if caminho == "/estado":
return self.json(self.estado())
if partes and partes[0] == "fila":
# Painéis não leem a fila: observam o rosto por /eventos.
return self.enviar(json.dumps(FILA.consumir(aparelho="aparelho=1" in self.path)))
if partes and partes[0] == "faces":
return self.enviar(json.dumps({"rostos": ROSTOS}))
if partes and partes[0] == "config":
return self.enviar(json.dumps(APARENCIA))
if partes and partes[0] == "gestos":
return self.enviar(json.dumps({"gestos": GESTOS}))
if partes and partes[0] == "diag":
# a propria pagina relata o tamanho que recebeu (ajuda a achar
# barras/enquadramento errado no kiosk). Aparece no journal.
info = unquote("/".join(partes[1:]))
print("DIAG kiosk:", info, flush=True)
return self.enviar(json.dumps({"ok": True}))
# Obsoletos (GET que muda estado); ficam até o painel antigo sair. Use POST /comando.
if len(partes) == 2 and partes[0] in ("gesto", "face"):
tipo = "gesto" if partes[0] == "gesto" else "rosto"
try:
ordem = ordem_de_comando(tipo, unquote(partes[1]))
except ValueError as e:
return self.json({"erro": str(e)}, 400)
enfileirar(ordem)
return self.json({"ok": True, partes[0]: ordem})
# Painel (Vite/Svelte compilado em web/dist) em "/"; o resto é lista branca.
painel = Path("web/dist")
if caminho in ("/", "", "/index.html"):
if not (painel / "index.html").is_file():
return self.json({"erro": "Painel não compilado: rode 'npm run build' em web/ (ou use deploy.ps1)"}, 503)
self.path = "/web/dist/index.html"
elif caminho.startswith("/assets/"):
nome = caminho[len("/assets/"):]
if "/" in nome or "\\" in nome or nome.startswith(".") or not (painel / "assets" / nome).is_file():
return self.json({"erro": "Arquivo não disponível"}, 404)
self.path = "/web/dist/assets/" + nome
elif caminho in ("/dashboard", "/painel", "/conversa", "/emojis", "/treinamento", "/braco", "/servos", "/api"):
# Endereços das telas antigas levam à seção equivalente do painel novo.
secao = {"/conversa": "conversa", "/emojis": "reacoes", "/treinamento": "reacoes", "/braco": "movimento",
"/servos": "movimento", "/api": "ajustes"}.get(caminho, "ao-vivo")
self.send_response(302)
self.send_header("Location", "/#/" + secao)
self.send_header("Content-Length", "0")
self.end_headers()
return
elif caminho == '/qualidade-auto.js':
self.path = '/android/app/assets/qualidade-auto.js'
elif caminho == '/app.apk':
self.path = '/android/carinha-rosto.apk'
elif caminho not in ESTATICOS:
return self.enviar(json.dumps({'erro': 'Arquivo não disponível'}), codigo=404)
return super().do_GET()
def do_POST(self):
if self.acesso_negado("POST"):
return
for modulo in self.modulos():
if modulo.rotas(self, "POST"):
return
rota = self.rota().rstrip("/")
try:
if rota == "/comando":
return self.comando(ler_json(self, 2048))
if rota == "/parar-tudo":
ler_json(self, 2048)
return self.json(self.parar_tudo())
if rota == "/rearmar":
ler_json(self, 2048)
return self.json(self.rearmar())
if rota == '/app/recarregar':
ler_json(self, 2048)
return self.recarregar()
if rota == "/config":
return self.json({"ok": True, "ordens": aplicar_aparencia(ler_json(self, 2048)),
"aparencia": APARENCIA})
if rota in ("/gesto", "/face"):
# Obsoleto: corpo é o nome cru. Use POST /comando.
tamanho = int(self.headers.get("Content-Length") or 0)
nome = self.rfile.read(min(tamanho, 200)).decode("utf-8", "ignore")
ordem = ordem_de_comando("gesto" if rota == "/gesto" else "rosto", nome)
enfileirar(ordem)
return self.json({"ok": True, "ordem": ordem})
except ValueError as e:
return self.json({"erro": str(e)}, 400)
return self.json({"erro": "Rota POST desconhecida"}, 404)
def comando(self, dados):
tipo, valor = dados.get("tipo"), dados.get("valor")
s = self.server
if tipo == "rosto":
ordem_de_comando("rosto", valor) # valida e dá erro legível
aplicar_aparencia({"rosto": valor}) # salva, avisa os painéis e manda ao celular
return self.json({"ok": True})
if tipo in ("cor", "formato"):
if not aplicar_aparencia({tipo: valor}):
raise ValueError("Cor inválida" if tipo == "cor" else "Formato inválido")
return self.json({"ok": True})
if tipo == "recarregar":
return self.recarregar()
if tipo in ("reacao", "parar_reacao"):
if not hasattr(s, "treinamento"):
raise ValueError("Reações indisponíveis")
if tipo == "reacao":
if type(valor) is not int:
raise ValueError("ID de reação inválido")
s.treinamento.testar(valor)
else:
with s.treinamento.lock:
s.treinamento.expressao = dict(id=time.time_ns(), horario=time.time(), receita=dict(limpar=True))
return self.json({"ok": True})
ordem = ordem_de_comando(tipo, valor)
enfileirar(ordem)
return self.json({"ok": True, "ordem": ordem})
def parar_tudo(self):
s = self.server
if hasattr(s, "braco"):
s.braco.cancelar()
if hasattr(s, "servos"):
with s.servos.lock:
s.servos.travado = True
s.servos.stop()
if hasattr(s, "treinamento"):
with s.treinamento.lock:
s.treinamento.expressao = dict(id=time.time_ns(), horario=time.time(), receita=dict(limpar=True))
FILA.limpar()
enfileirar("calma")
ESTOP.update(ativo=True, desde=time.time())
EVENTOS.publicar("estop", dict(ESTOP))
return {"ok": True, "estop": dict(ESTOP)}
def rearmar(self):
s = self.server
if hasattr(s, "servos"):
with s.servos.lock:
s.servos.travado = False
EVENTOS.publicar("motores", s.servos.motores())
ESTOP.update(ativo=False, desde=None)
EVENTOS.publicar("estop", dict(ESTOP))
return {"ok": True, "estop": dict(ESTOP)}
def recarregar(self):
controle = self.server.controle
with controle.lock:
online = time.monotonic() - controle.status_at < 5
controle.recarga = dict(id=uuid.uuid4().hex, confirmada=False)
if not online:
return self.json({"erro": "Abra o Carinha e conecte ao servidor primeiro"}, 409)
FILA.adicionar("recarregar")
EVENTOS.publicar("recarregar", {"id": controle.recarga["id"]})
return self.json({"ok": True, "mensagem": "Solicitação enviada. Aguardando o celular recarregar a tela."})
def log_message(self, formato, *args):
# silencio por padrao (roda em segundo plano); --log liga o registro
if getattr(self.server, "registrar", False):
super().log_message(formato, *args)
class Servidor(ThreadingHTTPServer):
"""No Windows o SO_REUSEADDR deixaria DOIS servidores na mesma porta, e o
antigo continuaria respondendo. Aqui a porta e exclusiva.
Nota: no Linux (a placa) o SO_REUSEADDR e util para reiniciar rapido, entao
so o desligamos no Windows."""
daemon_threads = True
allow_reuse_address = (os.name != "nt")
token = None
def ligar_eventos(servidor):
for nome in ("controle", "servos", "cerebro", "conversa", "treinamento"):
if hasattr(servidor, nome):
getattr(servidor, nome).publicar = EVENTOS.publicar
def main():
ap = argparse.ArgumentParser(description="Servidor da carinha em HTML (kiosk).")
ap.add_argument("--porta", type=int, default=8100)
ap.add_argument("--pasta", default=os.path.dirname(os.path.abspath(__file__)))
ap.add_argument("--log", action="store_true",
help="registra cada pedido (util para investigar o kiosk)")
args = ap.parse_args()
os.chdir(args.pasta)
carregar_aparencia()
servidor = Servidor(("0.0.0.0", args.porta), Handler)
servidor.token = carregar_token(Path.cwd())
servidor.api_config = ConfigAPI(Path.cwd() / 'api-privada.json')
servidor.controle = ControlePC(Path.cwd() / 'app-config.json')
servidor.servos = ControleServos(Path.cwd() / 'servos.json')
servidor.braco = Braco(Path.cwd() / 'braco.json', servidor.servos)
servidor.cerebro = Cerebro(Path.cwd() / 'ia-config.json', servidor.controle, enfileirar)
if servidor.token and not servidor.cerebro.chave:
servidor.cerebro.chave = servidor.token
servidor.conversa = Conversa(Path.cwd() / 'memoria.db', servidor.cerebro)
servidor.treinamento = Treinamento(Path.cwd() / 'treinamento.db')
servidor.conversa.memoria.on_registrar = servidor.treinamento.registrar
servidor.conversa.treinamento = servidor.treinamento
ligar_eventos(servidor)
servidor.registrar = args.log
print(f"servindo {args.pasta} em http://0.0.0.0:{args.porta}/")
print("autenticação:", "token obrigatório" if servidor.token else "DESLIGADA (defina CARINHA_TOKEN)")
try:
servidor.serve_forever()
finally:
EVENTOS.fechar()
if __name__ == "__main__":
main()
Reference in New Issue View Git Blame Copy Permalink