carinha-control-panel/controle_pc.py

"""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,
              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))


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 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()

    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:
                    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:
                        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]))
                    enviar({'ok': True})
                else:
                    enviar(self.resumo(h.headers.get("Host", "").split(":")[0]))
        except (ValueError, TypeError) as e:
            enviar({'erro': str(e)}, 400)
        except OSError as e:
            enviar({'erro': str(e)}, 503)
        return True