"""Configuração persistente e telemetria do celular; sinalização do vídeo WebRTC.""" import ipaddress import json import math from pathlib import Path 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, iris=False, subirBoca=0, distSobrancelha=16, videoRTC=True, apagarTela=True, apagarTelaMin=5, somAcordar=3, noiteSilenciosa=True, noiteInicio=23, noiteFim=7, 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), subirBoca=(-20, 60), distSobrancelha=(0, 60), apagarTelaMin=(1, 120), somAcordar=(1, 20), noiteInicio=(0, 23), noiteFim=(0, 23)) 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 def madrugada(config, agora=None): """Dentro do horário da madrugada silenciosa (ex.: 23h às 7h, atravessando a meia-noite)?""" if not config.get('noiteSilenciosa', True): return False ini, fim = int(config.get('noiteInicio', 23)), int(config.get('noiteFim', 7)) hora = time.localtime(agora if agora is not None else time.time()).tm_hour if ini == fim: return False return ini <= hora < fim if ini < fim else (hora >= ini or hora < fim) def ip_informado(dados): """IP que o próprio app informa (aparelho.ip); o endereço da conexão pode ser de um proxy.""" aparelho = dados.get('aparelho') ip = aparelho.get('ip') if isinstance(aparelho, dict) else None try: return str(ipaddress.IPv4Address(ip)) if isinstance(ip, str) else None except ValueError: return None 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.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() return novo.copy() console = staticmethod(lambda origem, texto, motivo='', nivel='info', agora=None: None) # o main troca pelo console do painel 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']) 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', '/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')) if tamanho <= 0 or tamanho > 16384: self.registrar_recusa(rota, 'corpo de %d bytes' % tamanho) enviar({'erro': 'Corpo vazio ou grande demais'}, 413) return True corpo = h.rfile.read(tamanho) if len(corpo) != tamanho: raise ValueError('Corpo incompleto') dados = 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: antes = self.estado self.estado = dados self.celular_ip = ip_informado(dados) or 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])) self.mudancas(antes, dados) enviar({'ok': True}) else: enviar(self.resumo(h.headers.get("Host", "").split(":")[0])) except (ValueError, TypeError) as e: self.registrar_recusa(rota, str(e)) enviar({'erro': str(e)}, 400) except OSError as e: enviar({'erro': str(e)}, 503) return True def mudancas(self, antes, agora): """O que mudou no celular desde o último status: vai para o console do painel.""" if not antes: self.console('celular', 'Celular conectado', 'rosto %s' % agora.get('rosto', '?')) return if bool(antes.get('dormindo')) != bool(agora.get('dormindo')): self.console('celular', 'O rosto dormiu' if agora.get('dormindo') else 'O rosto acordou') if antes.get('atividade') != agora.get('atividade'): if agora.get('atividade'): self.console('celular', 'Começou a atividade: %s' % agora['atividade']) else: self.console('celular', 'Terminou a atividade: %s' % antes['atividade']) if bool(antes.get('tela_apagada')) != bool(agora.get('tela_apagada')): self.console('celular', 'Apagou a tela (dormindo há um tempo)' if agora.get('tela_apagada') else 'Acendeu a tela') def registrar_recusa(self, rota, motivo): """Pedido do celular recusado: aparece no journal (no máximo 1 linha por minuto).""" agora = time.monotonic() if agora - getattr(self, '_recusa_em', -60) >= 60: self._recusa_em = agora print('app: recusado', rota, '-', motivo[:160], flush=True)