julioandClaude Opus 5.5
35d443a358
- 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>
472 lines
19 KiB
Python
472 lines
19 KiB
Python
#!/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 = ["eilik", "carinha", "miudo"]
|
|
|
|
# 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"]
|
|
# 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]}
|
|
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:
|
|
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"]
|
|
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 "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)
|
|
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})
|
|
|
|
if caminho in ("/dashboard", "/emojis", "/api"):
|
|
self.path = caminho + ".html"
|
|
elif caminho == "/painel":
|
|
self.path = "/painel.html" # painel de controle
|
|
elif caminho == '/servos':
|
|
self.path = '/servos.html'
|
|
elif caminho == '/conversa':
|
|
self.path = '/conversa.html'
|
|
elif caminho == '/treinamento':
|
|
self.path = '/treinamento.html'
|
|
elif caminho == '/braco':
|
|
self.path = '/braco.html'
|
|
elif caminho in ("/", ""):
|
|
self.path = "/dashboard.html"
|
|
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 in (
|
|
'/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'):
|
|
pass
|
|
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'):
|
|
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 == "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})
|
|
|
|
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")
|
|
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)
|
|
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()
|