304 lines
13 KiB
Python
304 lines
13 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""
|
|
kiosk-server.py — servidor local do PC: painel, carinha e fila de ordens.
|
|
|
|
Alem de entregar os arquivos da pasta (padrao /root/carinha), ele expõe:
|
|
|
|
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
|
|
|
|
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:
|
|
|
|
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.
|
|
"""
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import threading
|
|
from http.server import SimpleHTTPRequestHandler, ThreadingHTTPServer
|
|
from urllib.parse import unquote
|
|
from pathlib import Path
|
|
from controle_pc import ControlePC
|
|
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()
|
|
FILA = []
|
|
|
|
# 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"]
|
|
|
|
# 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"
|
|
|
|
|
|
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:
|
|
with open(ARQUIVO_APARENCIA, "w", encoding="utf-8") as f:
|
|
json.dump(APARENCIA, f, ensure_ascii=False, indent=2)
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def aplicar_aparencia(dados):
|
|
"""Guarda rosto/cor e poe as ordens na fila para a pagina 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:
|
|
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:
|
|
pass
|
|
FILA.extend(ordens)
|
|
salvar_aparencia()
|
|
return ordens
|
|
|
|
|
|
class Handler(SimpleHTTPRequestHandler):
|
|
extensoes_map = {".html": "text/html; charset=utf-8",
|
|
".js": "text/javascript; charset=utf-8",
|
|
".css": "text/css; charset=utf-8",
|
|
".json": "application/json; charset=utf-8",
|
|
".svg": "image/svg+xml",
|
|
".png": "image/png",
|
|
".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")
|
|
self.end_headers()
|
|
self.wfile.write(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'):
|
|
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]
|
|
partes = [p for p in caminho.split("/") if p]
|
|
|
|
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))
|
|
|
|
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 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.
|
|
info = unquote("/".join(partes[1:]))
|
|
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}))
|
|
|
|
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', '/audio-robot.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'):
|
|
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)
|
|
|
|
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.')))
|
|
|
|
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}))
|
|
|
|
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 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")
|
|
|
|
|
|
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.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.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
|
|
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()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|