julio/carinha-control-panel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(servidor): adiciona SSE, token, E-Stop real e trava de hardware

Browse Source 
- comum.py: ler_json (Origin/JSON/tamanho), gravar_atomico e token
  (CARINHA_TOKEN ou seguranca.json), substituindo 6 copias nos modulos
- eventos.py: hub SSE em /eventos com heartbeat e Last-Event-ID;
  modulos publicam expressao, telemetria, conversa, reacao, ia, motores
- /estado agrega o snapshot inicial; /comando unifica gesto/rosto/cor/reacao
- /parar-tudo desliga o PCA9685, para reacoes e trava motores ate /rearmar
- braco: lease (ttl_ms) e interpolacao com limite de velocidade no servidor
- braco e diagnostico nao disputam o PCA9685 (409 para o segundo dono)
- fila limitada (20) com validade de 10 s; POST /face usa prefixo "rosto"
- /app/config passa a checar Origin e JSON; versao vem de version.json
- figurinhas automaticas do Gemini limitadas a 6 por hora
- rosto entende a ordem "calma" enviada pelo E-Stop

Token so e exigido quando configurado (docs/SEGURANCA.md).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
This commit is contained in:
julio
2026-10-05 19:34:35 -03:00
co-authored by Claude Opus 5.5
parent 6c5bd6f432
commit 35d443a358
15 changed files with 875 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
kiosk-server.py
+281 -113
View File
@@ -1,38 +1,39 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
kiosk-server.py — servidor local do PC: painel, carinha e fila de ordens.
kiosk-server.py — servidor do Carinha (Orange Pi, porta 8100): painel, rosto e controle.
Alem de entregar os arquivos da pasta (padrao /root/carinha), ele expõe:
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')
GET /fila -> devolve e limpa a fila, em JSON: ["feliz","piscar"]
GET /gesto/<nome> -> poe <nome> na fila (ex.: /gesto/coracao)
POST /gesto -> corpo = nome do gesto
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)
A pagina da carinha pergunta /fila a cada meio segundo; assim da para mandar
um gesto do PC com um curl, igual ao antigo FIFO /tmp/eilik.fifo:
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.
curl http://orangepi:8099/gesto/feliz
Alem disso guarda a APARENCIA da carinha (rosto + cor dos olhos), para o painel
personalizar e a pagina aplicar sozinha ao carregar:
GET /config -> {"rosto":"carinha","cor":[92,214,255]}
POST /config -> corpo JSON {"rosto":"miudo","cor":[255,120,0]}
E serve o painel de controle em http://<maquina>:8099/painel .
So usa a biblioteca padrao (http.server). Nada de instalacao extra.
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 controle_pc import ControlePC
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
@@ -41,7 +42,6 @@ from treinamento import Treinamento
from api_config import ConfigAPI
TRAVA = threading.Lock()
FILA = []
# Espelho da biblioteca de rostos do carinha.html (para /faces funcionar)
ROSTOS = ["eilik", "carinha", "miudo"]
@@ -50,12 +50,65 @@ ROSTOS = ["eilik", "carinha", "miudo"]
GESTOS = ["piscar", "piscar2", "olhar_esq", "olhar_dir", "olhar_baixo", "feliz",
"surpreso", "bravo", "coracao", "triste", "piscadinha", "alegre",
"dormir", "acordar", "aleatorio", "parar"]
# 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
# 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]}
ARQUIVO_APARENCIA = "aparencia.json"
# 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/", "/camera/", "/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:
@@ -73,14 +126,20 @@ def carregar_aparencia():
def salvar_aparencia():
try:
with open(ARQUIVO_APARENCIA, "w", encoding="utf-8") as f:
json.dump(APARENCIA, f, ensure_ascii=False, indent=2)
except Exception:
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 poe as ordens na fila para a pagina aplicar na hora."""
"""Guarda rosto/cor e manda as ordens para a página aplicar na hora."""
ordens = []
with TRAVA:
if "rosto" in dados:
@@ -90,24 +149,43 @@ def aplicar_aparencia(dados):
ordens.append("rosto " + nome)
if "cor" in dados:
try:
cor = [max(0, min(255, int(v))) for v in list(dados["cor"])[:3]]
if len(cor) == 3:
APARENCIA["cor"] = cor
ordens.append("cor %d,%d,%d" % tuple(cor))
except Exception:
APARENCIA["cor"] = validar_cor(dados["cor"])
ordens.append("cor %d,%d,%d" % tuple(APARENCIA["cor"]))
except (TypeError, ValueError):
pass
FILA.extend(ordens)
salvar_aparencia()
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):
@@ -120,32 +198,71 @@ class Handler(SimpleHTTPRequestHandler):
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)
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 hasattr(self.server, "api_config") and self.server.api_config.rotas(self, "GET"):return
if hasattr(self.server, 'treinamento') and self.server.treinamento.rotas(self, 'GET'):
if self.acesso_negado("GET"):
return
if hasattr(self.server, 'conversa') and self.server.conversa.rotas(self, 'GET'):
return
if hasattr(self.server, 'cerebro') and self.server.cerebro.rotas(self, 'GET'):
return
if hasattr(self.server, 'braco') and self.server.braco.rotas(self, 'GET'):
return
if hasattr(self.server, 'servos') and self.server.servos.rotas(self, 'GET'):
return
if self.server.controle.rotas(self, 'GET'):
return
caminho = self.path.split("?", 1)[0]
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":
with TRAVA:
aparelho = 'aparelho=1' in self.path
pendentes = [x for x in FILA if x!='recarregar' or aparelho]
FILA[:] = [x for x in FILA if x=='recarregar' and not aparelho]
return self.enviar(json.dumps(pendentes))
# 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}))
@@ -156,12 +273,6 @@ class Handler(SimpleHTTPRequestHandler):
if partes and partes[0] == "gestos":
return self.enviar(json.dumps({"gestos": GESTOS}))
if len(partes) == 2 and partes[0] == "face":
nome = unquote(partes[1]).lower()
with TRAVA:
FILA.append("rosto " + nome)
return self.enviar(json.dumps({"ok": True, "rosto": nome}))
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.
@@ -169,11 +280,15 @@ class Handler(SimpleHTTPRequestHandler):
print("DIAG kiosk:", info, flush=True)
return self.enviar(json.dumps({"ok": True}))
if len(partes) == 2 and partes[0] == "gesto":
nome = unquote(partes[1]).lower() # aceita %20 para espacos
with TRAVA:
FILA.append(nome)
return self.enviar(json.dumps({"ok": True, "gesto": nome}))
# 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})
if caminho in ("/dashboard", "/emojis", "/api"):
self.path = caminho + ".html"
@@ -201,60 +316,103 @@ class Handler(SimpleHTTPRequestHandler):
elif caminho in ('/dashboard.html','/emojis.html','/api.html','/dashboard.css','/dashboard.js','/overview.js','/biblioteca.js','/api-painel.js','/painel.css','/conversa.css','/treinamento.css','/servos.css','/braco.css','/audio-robot.js'):
pass
elif caminho not in ('/conversa.html', '/treinamento.html', '/treinamento.js', '/reacoes-render.js', '/painel-conversa.js', '/braco-calibracao.js', '/movimentos-core.js', '/braco-movimentos.js', '/braco.html', '/braco.js', '/servos.html', '/servos.js', '/carinha.html', '/camera.html', '/painel.html', '/painel-controle.js', '/painel-video.js',
'/rastreamento-core.js', '/rastreador-worker.js', '/painel-rastreamento.js', '/audio-robot.js'):
'/rastreamento-core.js', '/rastreador-worker.js', '/painel-rastreamento.js'):
return self.enviar(json.dumps({'erro': 'Arquivo não disponível'}), codigo=404)
return super().do_GET()
def do_POST(self):
if hasattr(self.server, "api_config") and self.server.api_config.rotas(self, "POST"):return
if hasattr(self.server, 'treinamento') and self.server.treinamento.rotas(self, 'POST'):
if self.acesso_negado("POST"):
return
if hasattr(self.server, 'conversa') and self.server.conversa.rotas(self, 'POST'):
return
if hasattr(self.server, 'cerebro') and self.server.cerebro.rotas(self, 'POST'):
return
if hasattr(self.server, 'braco') and self.server.braco.rotas(self, 'POST'):
return
if hasattr(self.server, 'servos') and self.server.servos.rotas(self, 'POST'):
return
if self.server.controle.rotas(self, 'POST'):
return
rota = self.path.split("?", 1)[0].rstrip("/")
tamanho = int(self.headers.get("Content-Length") or 0)
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)
if rota == '/app/recarregar':
origin=self.headers.get('Origin')
if origin and origin not in ('http://'+self.headers.get('Host',''),'https://'+self.headers.get('Host','')):
return self.enviar(json.dumps(dict(erro='Origem inválida')),codigo=400)
import time
with self.server.controle.lock:
online=time.monotonic()-self.server.controle.status_at<5
import uuid
self.server.controle.recarga = dict(id=uuid.uuid4().hex, confirmada=False)
if not online:return self.enviar(json.dumps(dict(erro='Abra o Carinha e conecte ao servidor primeiro')),codigo=409)
with TRAVA:
if 'recarregar' not in FILA:FILA.append('recarregar')
return self.enviar(json.dumps(dict(ok=True,mensagem='Solicitação enviada. Aguardando o celular recarregar a tela.')))
def comando(self, dados):
tipo, valor = dados.get("tipo"), dados.get("valor")
s = self.server
if tipo == "cor":
if not aplicar_aparencia({"cor": valor}):
raise ValueError("Cor inválida")
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})
if rota == "/config":
try:
dados = json.loads(self.rfile.read(tamanho).decode("utf-8", "ignore") or "{}")
if not isinstance(dados, dict):
raise ValueError("esperava um objeto JSON")
except Exception as e:
return self.enviar(json.dumps({"erro": "json invalido: %s" % e}), codigo=400)
ordens = aplicar_aparencia(dados)
return self.enviar(json.dumps({"ok": True, "aparencia": APARENCIA,
"ordens": ordens}))
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)}
if rota not in ("/gesto", "/face"):
return self.enviar(json.dumps({"erro": "rotas POST: /gesto, /face, /config"}), codigo=404)
corpo = self.rfile.read(tamanho).decode("utf-8", "ignore").strip().lower()
if not corpo:
return self.enviar(json.dumps({"erro": "corpo vazio"}), codigo=400)
with TRAVA:
FILA.append(corpo)
return self.enviar(json.dumps({"ok": True, "ordem": corpo}))
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")
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
@@ -270,6 +428,13 @@ class Servidor(ThreadingHTTPServer):
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():
@@ -282,21 +447,24 @@ def main():
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)
def expressao_ia(gesto):
with TRAVA: FILA.append(gesto)
servidor.cerebro = Cerebro(Path.cwd() / 'ia-config.json', servidor.controle, expressao_ia)
servidor.cerebro = Cerebro(Path.cwd() / 'ia-config.json', servidor.controle, enfileirar)
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}/carinha.html")
print(f"painel de controle em http://0.0.0.0:{args.porta}/painel")
servidor.serve_forever()
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__":