113 lines
6.5 KiB
Python
113 lines
6.5 KiB
Python
"""Real WebRTC sender and panel in Chromium, using a synthetic camera."""
|
|
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})
|
|
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'})
|
|
panel = context.new_page()
|
|
panel.on('pageerror', lambda e: errors.append(str(e)))
|
|
panel.goto(base + '/painel')
|
|
panel.wait_for_function("document.getElementById('cameraVideo').videoWidth > 0", timeout=20000)
|
|
first = panel.evaluate("document.getElementById('cameraVideo').getVideoPlaybackQuality().totalVideoFrames")
|
|
panel.wait_for_timeout(2200)
|
|
count = panel.evaluate("document.getElementById('cameraVideo').getVideoPlaybackQuality().totalVideoFrames") - first
|
|
assert count >= 45, 'Vídeo não está fluido: %s quadros em 2.2 s' % count
|
|
assert panel.locator('#cameraVideo').is_visible()
|
|
assert 'fps' in panel.locator('#videoQualidade').inner_text()
|
|
assert not errors, errors
|
|
print('Vídeo WebRTC real: %s quadros em 2.2 s; %s' % (count, panel.locator('#videoQualidade').inner_text()))
|
|
sender.request.post(base+'/app/config',data={'rastrearMaos':True})
|
|
sender.wait_for_function('config.rastrearMaos === true')
|
|
panel.wait_for_function('CarinhaVideo.rastreamentoConectado()')
|
|
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})
|
|
panel.locator('#ajuste-videoResolucao').select_option('480')
|
|
sender.wait_for_function("stream && stream.getVideoTracks()[0].getSettings().height === 480", timeout=10000)
|
|
panel.locator('#ajuste-videoFPS').select_option('240')
|
|
sender.wait_for_function("aviso.includes('indisponível')", timeout=10000)
|
|
assert sender.evaluate('stream.getVideoTracks()[0].readyState') == 'live'
|
|
panel.locator('#ajuste-sensorCamera').uncheck()
|
|
panel.wait_for_function("document.getElementById('cameraVideo').srcObject === null", timeout=10000)
|
|
sender.wait_for_function('stream === null', timeout=10000)
|
|
panel.locator('#ajuste-sensorCamera').check()
|
|
panel.wait_for_function("document.getElementById('cameraVideo').videoWidth > 0 && document.getElementById('cameraVideo').srcObject !== null", timeout=15000)
|
|
assert not errors, errors
|
|
panel.get_by_role('button', name='Desligar câmera').click()
|
|
panel.wait_for_timeout(1400)
|
|
assert panel.evaluate("document.getElementById('cameraVideo').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()
|