julioandClaude Opus 5.5
c4b53a8976
- web/ com Vite + Svelte 5 + TypeScript; servidor entrega web/dist em "/" (telas antigas seguem em /dashboard, /painel, /braco...) - design system em tokens CSS, tema escuro e claro, fontes locais - estado unico: foto de /estado + um EventSource em /eventos, sem polling - shell com navegacao lateral/inferior, selos de status reais e E-Stop fixo - command palette (Ctrl+K), gestos favoritos com atalhos 1-9, g+letra navega - Ao vivo: camera WebRTC e rastreamento portados para TS, rosto, aparencia, sensores do celular, IA local e feed de acontecimentos - Ajustes: campos do celular agrupados com busca, salvamento otimista e restaurar padrao (GET /app/padrao); motor da conversa, Gemini e token - carinha.html ganha modo ?painel=1 dirigido por postMessage (sem polling) - teste Playwright de ponta a ponta do painel novo Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
211 lines
10 KiB
Python
211 lines
10 KiB
Python
"""Configuração persistente, telemetria e prévia da câmera do celular."""
|
|
import ipaddress
|
|
import json
|
|
import math
|
|
import os
|
|
from pathlib import Path
|
|
import subprocess
|
|
import threading
|
|
import time
|
|
from comum import origem_valida
|
|
from sinalizacao import Sinalizacao
|
|
|
|
|
|
def versao():
|
|
try:
|
|
return json.loads((Path(__file__).with_name('version.json')).read_text(encoding='utf-8'))['versao']
|
|
except (OSError, ValueError, KeyError):
|
|
return 'desconhecida'
|
|
|
|
PADRAO = dict(preview=False, auto=True, audio=True, volume=0.35,
|
|
ouvir=True, reagir=True, somlim=0.35, somganho=0.4,
|
|
somaltolim=0.55, sompausa=20, somcooldown=5,
|
|
movlim=0.05, movforte=0.12, movcooldown=6, movolhar=40,
|
|
economia=True, brilho=0.7, brilhoRepouso=0.08, repousoTela=30,
|
|
boca=True, sobrancelhas=False, videoRTC=True,
|
|
sensorMic=True, sensorCamera=True, micSono=True, cameraSono=False,
|
|
cameraLado='user', videoResolucao='auto', videoFPS=0,
|
|
rastrearRosto=True, rastrearMaos=True, seguirRastreamento=True, mostrarRastreamento=True)
|
|
ESCOLHAS = dict(cameraLado=('user', 'environment'),
|
|
videoResolucao=('auto', 'max', '480', '720', '1080', '2160', '4320'),
|
|
videoFPS=(0, -1, 15, 24, 30, 60, 120, 240))
|
|
LIMITES = dict(volume=(0, 1), somlim=(0, 1), somganho=(0, 1),
|
|
somaltolim=(0, 1), sompausa=(1, 3600), somcooldown=(0, 120),
|
|
movlim=(0, 1), movforte=(0, 1), movcooldown=(0, 120), movolhar=(0, 100),
|
|
brilho=(0.01, 1), brilhoRepouso=(0.01, 1), repousoTela=(5, 3600))
|
|
TOOLS = Path(os.environ.get('CARINHA_TOOLS',
|
|
'C:/Users/JullioPC/Documents/Open/orange-pi-zero3/android/tools'))
|
|
|
|
|
|
def validar(dados):
|
|
if not isinstance(dados, dict):
|
|
raise ValueError('Esperava um objeto JSON')
|
|
for chave, valor in dados.items():
|
|
if chave not in PADRAO:
|
|
raise ValueError('Ajuste desconhecido: ' + chave)
|
|
if chave in ESCOLHAS:
|
|
if isinstance(valor, bool) or valor not in ESCOLHAS[chave]:
|
|
raise ValueError('Opção inválida: ' + chave)
|
|
elif isinstance(PADRAO[chave], bool):
|
|
if not isinstance(valor, bool):
|
|
raise ValueError(chave + ' precisa ser verdadeiro ou falso')
|
|
else:
|
|
baixo, alto = LIMITES[chave]
|
|
if isinstance(valor, bool) or not isinstance(valor, (int, float)) or not math.isfinite(valor) or not baixo <= valor <= alto:
|
|
raise ValueError('%s deve ficar entre %s e %s' % (chave, baixo, alto))
|
|
return dados
|
|
|
|
|
|
class ControlePC:
|
|
def __init__(self, arquivo):
|
|
self.arquivo = Path(arquivo)
|
|
self.lock = threading.Lock()
|
|
self.config = PADRAO.copy()
|
|
self.estado = {}
|
|
self.status_at = 0
|
|
self.celular_ip = None
|
|
self.recarga = {"id": None, "confirmada": False}
|
|
self.frame = b''
|
|
self.frame_at = 0
|
|
self.frame_seq = 0
|
|
self.rtc = Sinalizacao()
|
|
self.publicar = lambda tipo, dados: None
|
|
try:
|
|
self.config.update(validar(json.loads(self.arquivo.read_text(encoding='utf-8'))))
|
|
except (OSError, ValueError):
|
|
pass
|
|
|
|
def atualizar(self, dados):
|
|
validar(dados)
|
|
with self.lock:
|
|
novo = dict(self.config, **dados)
|
|
tmp = self.arquivo.with_suffix('.tmp')
|
|
tmp.write_text(json.dumps(novo, ensure_ascii=False, indent=2), encoding='utf-8')
|
|
tmp.replace(self.arquivo)
|
|
self.config = novo
|
|
if not novo['preview']:
|
|
self.rtc.encerrar()
|
|
self.frame = b''
|
|
self.frame_at = 0
|
|
return novo.copy()
|
|
|
|
def resumo(self, host=''):
|
|
"""Estado do celular para o painel (GET /app/status, /estado e evento 'telemetria')."""
|
|
with self.lock:
|
|
agora = time.monotonic()
|
|
idade = agora - self.status_at
|
|
return dict(online=idade < 5, idade=round(idade, 1) if self.status_at else None,
|
|
celular_ip=self.celular_ip, servidor_ip=host, recarga=self.recarga.copy(),
|
|
estado=self.estado.copy(), preview=self.config['preview'],
|
|
camera_recente=bool(self.frame) and agora - self.frame_at < 3 and self.config['preview'],
|
|
frame_seq=self.frame_seq)
|
|
|
|
def rotas(self, h, metodo):
|
|
if self.rtc.atender(h, metodo, self.config['preview']):
|
|
return True
|
|
rota = h.path.split('?', 1)[0].rstrip('/')
|
|
if rota not in ('/app/config', '/app/padrao', '/app/status', '/camera/frame', '/app/connect', '/app/health'):
|
|
return False
|
|
|
|
def enviar(dados, codigo=200):
|
|
h.enviar(json.dumps(dados, ensure_ascii=False), codigo=codigo)
|
|
|
|
try:
|
|
if metodo == 'POST':
|
|
if not origem_valida(h):
|
|
raise ValueError('Origem inválida')
|
|
if rota == '/app/config' and h.headers.get('Content-Type', '').split(';')[0] != 'application/json':
|
|
raise ValueError('Use JSON')
|
|
tamanho = int(h.headers.get('Content-Length', '0'))
|
|
limite = 1_000_000 if rota == '/camera/frame' else 16384
|
|
if tamanho <= 0 or tamanho > limite:
|
|
enviar({'erro': 'Corpo vazio ou grande demais'}, 413)
|
|
return True
|
|
corpo = h.rfile.read(tamanho)
|
|
if len(corpo) != tamanho:
|
|
raise ValueError('Corpo incompleto')
|
|
dados = None if rota == '/camera/frame' else json.loads(corpo)
|
|
else:
|
|
dados = None
|
|
|
|
if rota == '/app/padrao':
|
|
enviar(PADRAO)
|
|
elif rota == '/app/health':
|
|
enviar({'projeto': 'carinha-pc', 'versao': versao()})
|
|
elif rota == '/app/config':
|
|
if metodo == 'POST':
|
|
config = self.atualizar(dados)
|
|
self.publicar('app_config', config)
|
|
enviar({'ok': True, 'config': config})
|
|
else:
|
|
with self.lock:
|
|
config = self.config.copy()
|
|
enviar(config)
|
|
elif rota == '/app/status':
|
|
if metodo == 'POST':
|
|
if not isinstance(dados, dict):
|
|
raise ValueError('Estado deve ser objeto JSON')
|
|
with self.lock:
|
|
self.estado = dados
|
|
self.celular_ip = h.client_address[0]
|
|
if self.recarga["id"] and str(dados.get("tela_atualizada", "")) == self.recarga["id"]:
|
|
self.recarga["confirmada"] = True
|
|
self.status_at = time.monotonic()
|
|
self.publicar('telemetria', self.resumo(h.headers.get("Host", "").split(":")[0]))
|
|
enviar({'ok': True})
|
|
else:
|
|
enviar(self.resumo(h.headers.get("Host", "").split(":")[0]))
|
|
elif rota == '/camera/frame':
|
|
with self.lock:
|
|
if metodo == 'POST':
|
|
if not self.config['preview']:
|
|
enviar({'erro': 'Transmissão desligada'}, 409)
|
|
elif h.headers.get('Content-Type', '').split(';')[0] != 'image/jpeg' or not corpo.startswith(b'\xff\xd8') or not corpo.endswith(b'\xff\xd9'):
|
|
enviar({'erro': 'Esperava uma imagem JPEG'}, 400)
|
|
else:
|
|
self.frame = corpo
|
|
self.frame_at = time.monotonic()
|
|
self.frame_seq += 1
|
|
enviar({'ok': True})
|
|
else:
|
|
recente = self.config['preview'] and self.frame and time.monotonic() - self.frame_at < 3
|
|
h.enviar(self.frame if recente else b'', 'image/jpeg', 200 if recente else 204)
|
|
elif rota == '/app/connect':
|
|
if metodo != 'POST':
|
|
enviar({'erro': 'Use POST'}, 405)
|
|
elif h.client_address[0] not in ('127.0.0.1', '::1'):
|
|
enviar({'erro': 'Abra o painel em localhost no PC para conectar'}, 403)
|
|
else:
|
|
self.conectar(h, dados, enviar)
|
|
except (ValueError, TypeError) as e:
|
|
enviar({'erro': str(e)}, 400)
|
|
except OSError as e:
|
|
enviar({'erro': str(e)}, 503)
|
|
return True
|
|
|
|
def conectar(self, h, dados, enviar):
|
|
ip = str(ipaddress.IPv4Address(dados.get('ip', '192.168.1.16')))
|
|
porta = dados.get('porta', 5555)
|
|
if isinstance(porta, bool) or not isinstance(porta, int) or not 1 <= porta <= 65535:
|
|
raise ValueError('Porta ADB inválida')
|
|
pc = str(ipaddress.IPv4Address(dados.get('pc', '192.168.1.15')))
|
|
adb = str(TOOLS / 'android-sdk/platform-tools/adb.exe')
|
|
serial = '%s:%s' % (ip, porta)
|
|
url = 'http://%s:%s' % (pc, h.server.server_port)
|
|
comandos = [[adb, 'connect', serial], [adb, '-s', serial, 'shell', 'input', 'keyevent', 'KEYCODE_WAKEUP'],
|
|
[adb, '-s', serial, 'shell', 'wm', 'dismiss-keyguard'],
|
|
[adb, '-s', serial, 'shell', 'am', 'start', '-n', 'br.carinha.rosto/.MainActivity', '-e', 'servidor', url]]
|
|
saidas = []
|
|
for cmd in comandos:
|
|
try:
|
|
r = subprocess.run(cmd, capture_output=True, text=True, timeout=15,
|
|
creationflags=subprocess.CREATE_NO_WINDOW if os.name == 'nt' else 0)
|
|
except subprocess.TimeoutExpired:
|
|
enviar({'erro': 'ADB não respondeu. Confira a depuração Wi-Fi no celular.'}, 504)
|
|
return
|
|
saidas.append((r.stdout + r.stderr).strip())
|
|
if r.returncode or 'cannot connect' in r.stdout.lower() or 'failed' in r.stdout.lower() or 'error:' in r.stdout.lower():
|
|
enviar({'erro': '\n'.join(saidas)}, 502)
|
|
return
|
|
enviar({'ok': True, 'servidor': url, 'mensagem': 'App aberto. Aguarde a confirmação de conexão no painel.'})
|