julio/carinha-control-panel
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
b456f807164004d2b9d418c82beec112ee4d8b84
carinha-control-panel/kiosk-server.py
T
julioandClaude Opus 5.5 b456f80716 feat(vida): modo autônomo, o Carinha como bichinho virtual 
Novo módulo vida.py: energia, alegria, companhia e diversão caem com o
tempo e sobem com presença (PIR), movimento, conversa e carinho. Delas sai
o humor e, do humor, os atos: gestos, cochilo quando cansado, dormir à
noite sem ninguém por perto, acenar pedindo atenção quando carente com
gente perto, tocar reações da biblioteca e inventar figurinhas novas com
o Gemini (no máximo 3 por dia, 90 min entre elas).

- O celular recebe "humor x" e escolhe os gestos soltos por ele.
- Toque na tela do celular conta como carinho/cutucada (/vida/cuidar).
- Respeita E-Stop e o modo presença (ninguém em casa = quieto).
- Painel: cartão Vida no Ao vivo com humor, barras, cuidados, ritmo e
  diário; liga por padrão.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-06 04:21:23 -03:00

548 lines
23 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
from historico import Historico
from vida import Vida
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/", "/historico/", "/vida/",
"/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", "historico", "vida", "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()
if hasattr(s, "historico"):
dados["pir"] = dict(s.historico.pir, presenca=dict(s.historico.presenca))
if hasattr(s, "vida"):
dados["vida"] = s.vida.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", "historico", "vida"):
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.historico = Historico(Path.cwd())
servidor.vida = vida = Vida(Path.cwd() / 'vida.json')
def registrar_conversa(papel, texto):
servidor.treinamento.registrar(papel, texto)
if papel == 'user':
vida.estimulo('conversa') # alguém conversou com ele: companhia
servidor.conversa.memoria.on_registrar = registrar_conversa
def celular_dormindo():
resumo = servidor.controle.resumo()
return bool(resumo['estado'].get('dormindo')) if resumo['online'] else None
def alguem_perto():
h = servidor.historico
recente = max(h.pir.get('ultimo') or 0, h.ultimo.get('movimento', 0))
return bool(h.pir.get('ativo')) or time.time() - recente < 120
vida.enfileirar = enfileirar
vida.treinamento = servidor.treinamento
vida.celular_dormindo = celular_dormindo
vida.celular_online = lambda: servidor.controle.resumo()['online']
vida.alguem_perto = alguem_perto
# E-Stop parado, ou modo presença com ninguém em casa (dormindo em silêncio): não faz nada
vida.bloqueado = lambda: ESTOP["ativo"] or servidor.historico.presenca.get("estado") == "ausente"
def presenca(tipo):
vida.estimulo(tipo)
# PIR viu alguém: se o rosto estiver dormindo (e talvez com a tela apagada), acorda
if tipo == "pir":
with servidor.controle.lock:
dormindo = bool(servidor.controle.estado.get("dormindo"))
if dormindo:
enfileirar("acordar")
def camera(ligada):
if servidor.controle.config.get("sensorCamera") != ligada:
servidor.controle.publicar("app_config", servidor.controle.atualizar({"sensorCamera": ligada}))
def alguem_chegou():
camera(True) # modo presença: o PIR liga a câmera do celular
enfileirar("acordar")
def ninguem_por_perto():
camera(False) # sem ninguém: câmera desligada, dorme e fica em silêncio
enfileirar("dormir")
servidor.historico.ao_detectar = presenca
servidor.historico.ao_presenca = alguem_chegou
servidor.historico.ao_ausencia = ninguem_por_perto
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