carinha-control-panel/tests/test_rtc_browser.py

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