"""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 sinalizacao import Sinalizacao 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() 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 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/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': 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/health': enviar({'projeto': 'carinha-pc', 'versao': '0.7.0'}) elif rota == '/app/config': if metodo == 'POST': enviar({'ok': True, 'config': self.atualizar(dados)}) 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() enviar({'ok': True}) else: with self.lock: idade = time.monotonic() - self.status_at frame_idade = time.monotonic() - self.frame_at resultado = dict(online=idade < 5, idade=round(idade, 1) if self.status_at else None, celular_ip=self.celular_ip, servidor_ip=h.headers.get("Host", "").split(":")[0], recarga=self.recarga.copy(), estado=self.estado.copy(), preview=self.config['preview'], camera_recente=bool(self.frame) and frame_idade < 3 and self.config['preview'], frame_seq=self.frame_seq) enviar(resultado) 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.'})