"""WebRTC real entre o transmissor do app e o painel novo (web/dist), com câmera sintética. Requer `npm run build` em web/.""" import importlib.util from pathlib import Path import sys import tempfile import threading from playwright.sync_api import sync_playwright ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) spec = importlib.util.spec_from_file_location('kiosk_rtc', ROOT / 'kiosk-server.py') kiosk = importlib.util.module_from_spec(spec) spec.loader.exec_module(kiosk) class Handler(kiosk.Handler): def do_GET(self): assets = {'/test-sender': 'transmissao.html', '/qualidade-auto.js': 'qualidade-auto.js'} if self.path in assets: path = ROOT / 'android/app/assets' / assets[self.path] return self.enviar(path.read_bytes(), 'text/html' if self.path == '/test-sender' else 'text/javascript') return super().do_GET() def main(): with tempfile.TemporaryDirectory() as tmp: server = kiosk.Servidor(('127.0.0.1', 0), Handler) server.controle = kiosk.ControlePC(Path(tmp) / 'config.json') server.controle.atualizar({'preview': True,'rastrearRosto':False,'rastrearMaos':False}) kiosk.ligar_eventos(server) # telemetria e ajustes chegam ao painel por SSE thread = threading.Thread(target=server.serve_forever, daemon=True) thread.start() base = 'http://127.0.0.1:' + str(server.server_port) try: with sync_playwright() as p: browser = p.chromium.launch(headless=True, args=['--use-fake-device-for-media-stream=fps=30','--use-fake-ui-for-media-stream']) context = browser.new_context() sender = context.new_page() errors = [] sender.on('pageerror', lambda e: errors.append(str(e))) sender.add_init_script(""" window.CarinhaRTC = { enviar(json) { fetch('/rtc/send',{method:'POST',headers:{'Content-Type':'application/json'},body:json}).catch(console.error); }, movimento() {}, rastrear(json) { window.testAlvo=JSON.parse(json); }, estado(json) { fetch('/app/status',{method:'POST',headers:{'Content-Type':'application/json'},body:JSON.stringify({rosto:'carinha',video:JSON.parse(json)})}).catch(console.error); } }; """) sender.goto(base + '/test-sender') sender.evaluate("""() => { let id = '', cursor = 0, busy = false; window.testPoll = setInterval(async () => { if (busy) return; busy = true; try { const packet = await (await fetch('/rtc/poll?lado=app&desde='+cursor+'&id='+id)).json(); if ((packet.id||'') !== id) { cursor=0; id=packet.id||''; } packet.config = await (await fetch('/app/config')).json(); window.rtcReceber(packet); for (const msg of packet.mensagens) cursor=Math.max(cursor,msg.seq); } finally { busy=false; } },200); }""") sender.request.post(base + '/app/status', data={'rosto':'carinha'}) # Como o celular real, manda batimento a cada 1 s (senão o painel o considera offline). sender.evaluate("""() => setInterval(() => fetch('/app/status',{method:'POST', headers:{'Content-Type':'application/json'},body:JSON.stringify({rosto:'carinha'})}), 1000)""") panel = context.new_page() panel.on('pageerror', lambda e: errors.append(str(e))) panel.goto(base + '/') video = "document.querySelector('.camera video')" panel.wait_for_function(video + ".videoWidth > 0", timeout=20000) first = panel.evaluate(video + ".getVideoPlaybackQuality().totalVideoFrames") panel.wait_for_timeout(2200) count = panel.evaluate(video + ".getVideoPlaybackQuality().totalVideoFrames") - first assert count >= 45, 'Vídeo não está fluido: %s quadros em 2.2 s' % count assert panel.locator('.camera video').is_visible() qualidade = panel.locator('.camera .rodape .mono').inner_text() assert 'fps' in qualidade, qualidade assert not errors, errors print('Vídeo WebRTC real: %s quadros em 2.2 s; %s' % (count, qualidade)) # Canal de rastreamento (DataChannel) chega ao transmissor. sender.request.post(base+'/app/config',data={'rastrearMaos':True}) sender.wait_for_function('config.rastrearMaos === true') panel.wait_for_function('window.carinhaVideo?.rastreamentoConectado') panel.wait_for_timeout(1000) # sem vendor/ o rastreador falha e manda null uma vez panel.evaluate("carinhaVideo.rastrear({tipo:'mao',x:.8,y:.3,atividade:true})") sender.wait_for_function("window.testAlvo?.tipo === 'mao' && testAlvo.x === .8") panel.evaluate('carinhaVideo.rastrear(null)') sender.wait_for_function('window.testAlvo === null') sender.request.post(base+'/app/config',data={'rastrearMaos':False}) # Ajustes chegam ao celular; o painel acompanha por SSE. sender.request.post(base+'/app/config',data={'videoResolucao':'480'}) sender.wait_for_function("stream && stream.getVideoTracks()[0].getSettings().height === 480", timeout=10000) sender.request.post(base+'/app/config',data={'videoFPS':240}) sender.wait_for_function("aviso.includes('indisponível')", timeout=10000) assert sender.evaluate('stream.getVideoTracks()[0].readyState') == 'live' sender.request.post(base+'/app/config',data={'sensorCamera':False}) panel.wait_for_function(video + ".srcObject === null", timeout=10000) sender.wait_for_function('stream === null', timeout=10000) sender.request.post(base+'/app/config',data={'sensorCamera':True}) panel.wait_for_function(video + ".videoWidth > 0 && " + video + ".srcObject !== null", timeout=15000) assert not errors, errors # Desligar pela chave da tela Ao vivo. panel.get_by_role('switch', name='Transmitir câmera do celular').locator('xpath=..').click() panel.wait_for_timeout(1400) assert panel.evaluate(video + ".srcObject === null") preview = context.new_page() preview.on('pageerror', lambda e: errors.append(str(e))) preview.goto(base + '/camera.html?lado=user&resolucao=480&fps=30') preview.wait_for_function("document.getElementById('video').videoHeight === 480", timeout=10000) assert not errors, errors browser.close() finally: server.shutdown() server.server_close() thread.join() if __name__ == '__main__': main()