223 lines
9.0 KiB
Python
223 lines
9.0 KiB
Python
"""Der volle Durchlauf gegen einen nachgebauten Miracast-Fernseher.
|
|
|
|
Der simulierte Fernseher verhält sich wie ein echtes Gerät: Er baut die
|
|
Steuerverbindung zu uns auf (so schreibt es der Standard vor), führt den
|
|
Handshake M1 bis M7 und empfängt anschließend den RTP-Strom. Was hier
|
|
ankommt, wird mit ffprobe auf Dekodierbarkeit geprüft.
|
|
|
|
Nicht abgedeckt: die Wi-Fi-Direct-Verbindung selbst - dafür braucht es
|
|
WLAN-Hardware.
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import socket
|
|
import subprocess
|
|
import sys
|
|
import tempfile
|
|
import threading
|
|
import time
|
|
import unittest
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from wfd import rtsp
|
|
from wfd.formats import VideoFormat
|
|
from wfd.streamer import Streamer
|
|
|
|
CONTROL_PORT = 17236
|
|
SINK_FORMATS = "00 00 02 04 0001DEFF 155557FF 00000FFF 00 0000 0000 11 none none"
|
|
|
|
|
|
class SimulatedTv(threading.Thread):
|
|
"""Spielt einen Miracast-Empfänger nach."""
|
|
|
|
def __init__(self, control_port, rtp_socket):
|
|
super().__init__(daemon=True, name="simulierter-fernseher")
|
|
self.control_port = control_port
|
|
self.rtp_socket = rtp_socket
|
|
self.transcript = []
|
|
self.error = None
|
|
self.played = threading.Event()
|
|
|
|
def run(self):
|
|
try:
|
|
self._talk()
|
|
except Exception as error: # noqa: BLE001 - im Test wollen wir alles sehen
|
|
self.error = error
|
|
|
|
def _talk(self):
|
|
sock = socket.create_connection(("127.0.0.1", self.control_port), timeout=10)
|
|
stream = sock.makefile("rwb")
|
|
|
|
requests, responses = [], []
|
|
lock = threading.Condition()
|
|
|
|
def reader():
|
|
while True:
|
|
message = rtsp.read_message(stream)
|
|
if message is None:
|
|
break
|
|
with lock:
|
|
(requests if message.is_request else responses).append(message)
|
|
lock.notify_all()
|
|
|
|
threading.Thread(target=reader, daemon=True).start()
|
|
|
|
def take(bucket, predicate=lambda m: True, timeout=15):
|
|
deadline = time.monotonic() + timeout
|
|
with lock:
|
|
while time.monotonic() < deadline:
|
|
for index, message in enumerate(bucket):
|
|
if predicate(message):
|
|
return bucket.pop(index)
|
|
lock.wait(0.2)
|
|
raise AssertionError("Nachricht blieb aus")
|
|
|
|
def write(data):
|
|
stream.write(data)
|
|
stream.flush()
|
|
|
|
# M1: Die Quelle meldet sich von sich aus.
|
|
m1 = take(requests, lambda m: m.method == "OPTIONS")
|
|
assert m1.headers.get("require") == "org.wfa.wfd1.0", m1.headers
|
|
self.transcript.append("M1 OPTIONS")
|
|
write(rtsp.build_response(m1.cseq, {
|
|
"Public": "org.wfa.wfd1.0, SETUP, TEARDOWN, PLAY, PAUSE, GET_PARAMETER, SET_PARAMETER"}))
|
|
|
|
# M2: Wir fragen zurück.
|
|
write(rtsp.build_request("OPTIONS", "*", 100, {"Require": "org.wfa.wfd1.0"}))
|
|
m2 = take(responses)
|
|
assert m2.status_code == 200, m2.status_code
|
|
self.transcript.append("M2 beantwortet")
|
|
|
|
# M3: Die Quelle fragt unsere Fähigkeiten ab.
|
|
m3 = take(requests, lambda m: m.method == "GET_PARAMETER")
|
|
assert "wfd_video_formats" in m3.body
|
|
self.transcript.append("M3 GET_PARAMETER")
|
|
write(rtsp.build_response(m3.cseq, body=(
|
|
f"wfd_video_formats: {SINK_FORMATS}\r\n"
|
|
"wfd_audio_codecs: AAC 00000007 00\r\n"
|
|
f"wfd_client_rtp_ports: RTP/AVP/UDP;unicast {self.rtp_socket.getsockname()[1]} 0 mode=play\r\n"
|
|
"wfd_content_protection: none\r\n")))
|
|
|
|
# M4: Sie legt genau ein Format fest.
|
|
m4 = take(requests, lambda m: m.method == "SET_PARAMETER" and "wfd_video_formats" in m.body)
|
|
announced = m4.param("wfd_video_formats")
|
|
cea = int(announced.split()[4], 16)
|
|
assert bin(cea).count("1") == 1, f"Es muss genau ein Format sein: {announced}"
|
|
assert cea == 1 << 5, f"Erwartet 1280x720p30, bekam Bit {cea.bit_length() - 1}"
|
|
assert "wfd_presentation_URL" in m4.body
|
|
self.transcript.append("M4 Format festgelegt")
|
|
write(rtsp.build_response(m4.cseq))
|
|
|
|
# M5: Sie bittet uns, die Sitzung einzurichten.
|
|
m5 = take(requests, lambda m: m.method == "SET_PARAMETER" and "wfd_trigger_method" in m.body)
|
|
assert m5.param("wfd_trigger_method") == "SETUP"
|
|
self.transcript.append("M5 Trigger SETUP")
|
|
write(rtsp.build_response(m5.cseq))
|
|
|
|
# M6: Wir richten die Sitzung ein und nennen unseren Empfangsport.
|
|
port = self.rtp_socket.getsockname()[1]
|
|
write(rtsp.build_request("SETUP", "rtsp://127.0.0.1/wfd1.0/streamid=0", 101,
|
|
{"Transport": f"RTP/AVP/UDP;unicast;client_port={port}"}))
|
|
m6 = take(responses)
|
|
assert m6.status_code == 200
|
|
assert m6.headers.get("session"), "Session-Kopfzeile fehlt"
|
|
self.transcript.append("M6 Sitzung eingerichtet")
|
|
|
|
# M7: Wiedergabe anfordern.
|
|
write(rtsp.build_request("PLAY", "rtsp://127.0.0.1/wfd1.0/streamid=0", 102,
|
|
{"Session": m6.session}))
|
|
m7 = take(responses)
|
|
assert m7.status_code == 200
|
|
self.transcript.append("M7 Wiedergabe")
|
|
self.played.set()
|
|
|
|
time.sleep(8)
|
|
|
|
|
|
class EndToEndTest(unittest.TestCase):
|
|
|
|
@unittest.skipUnless(shutil.which("ffmpeg"), "ffmpeg wird gebraucht")
|
|
def test_handshake_then_real_video(self):
|
|
rtp_socket = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
|
|
rtp_socket.bind(("127.0.0.1", 0))
|
|
rtp_socket.settimeout(12)
|
|
|
|
streamer = Streamer(source="test", bitrate_mbit=4)
|
|
played = {"format": None, "host": None, "port": None}
|
|
stopped = []
|
|
|
|
def on_play(host, port, fmt):
|
|
played.update(format=fmt, host=host, port=port)
|
|
streamer.start(host, port, fmt)
|
|
|
|
server = rtsp.SourceServer("127.0.0.1", 1080, on_play,
|
|
lambda reason: stopped.append(reason),
|
|
ports=(CONTROL_PORT,))
|
|
opened = server.start()
|
|
self.assertEqual([CONTROL_PORT], opened, "Der Steuerkanal muss offen sein")
|
|
|
|
tv = SimulatedTv(CONTROL_PORT, rtp_socket)
|
|
tv.start()
|
|
|
|
try:
|
|
self.assertTrue(tv.played.wait(25),
|
|
f"Handshake unvollständig: {tv.transcript} / {tv.error}")
|
|
self.assertIsNone(tv.error, f"Fernseher meldet: {tv.error}")
|
|
self.assertEqual(["M1 OPTIONS", "M2 beantwortet", "M3 GET_PARAMETER",
|
|
"M4 Format festgelegt", "M5 Trigger SETUP",
|
|
"M6 Sitzung eingerichtet", "M7 Wiedergabe"], tv.transcript)
|
|
self.assertEqual((1280, 720, 30),
|
|
(played["format"].width, played["format"].height,
|
|
played["format"].fps))
|
|
|
|
# Jetzt muss echtes Bild ankommen.
|
|
payload = bytearray()
|
|
sequences = []
|
|
gaps = 0
|
|
deadline = time.monotonic() + 12
|
|
while len(payload) < 250_000 and time.monotonic() < deadline:
|
|
try:
|
|
data, _ = rtp_socket.recvfrom(4096)
|
|
except socket.timeout:
|
|
break
|
|
self.assertEqual(2, data[0] >> 6, "RTP-Version muss 2 sein")
|
|
self.assertEqual(33, data[1] & 0x7F, "Nutzlasttyp muss 33 (MP2T) sein")
|
|
self.assertEqual(0x47, data[12], "Nutzlast muss MPEG-TS sein")
|
|
self.assertEqual(0, (len(data) - 12) % 188)
|
|
sequence = (data[2] << 8) | data[3]
|
|
if sequences and sequence != (sequences[-1] + 1) & 0xFFFF:
|
|
gaps += 1
|
|
sequences.append(sequence)
|
|
payload += data[12:]
|
|
|
|
self.assertGreater(len(payload), 200_000, "Zu wenig Bilddaten angekommen")
|
|
self.assertEqual(0, gaps, "Die RTP-Sequenz muss lückenlos sein")
|
|
|
|
# Der letzte Frame ist angeschnitten - für die Prüfung abrunden.
|
|
whole = len(payload) - (len(payload) % 188)
|
|
target = os.path.join(tempfile.gettempdir(), "wfd-endtoend.ts")
|
|
with open(target, "wb") as handle:
|
|
handle.write(payload[:whole - 188 * 40])
|
|
|
|
probe = subprocess.run(
|
|
["ffprobe", "-v", "error", "-show_entries",
|
|
"stream=codec_name,width,height", "-of", "default=noprint_wrappers=1", target],
|
|
capture_output=True, text=True, timeout=30)
|
|
self.assertIn("codec_name=h264", probe.stdout, probe.stderr)
|
|
self.assertIn("width=1280", probe.stdout)
|
|
self.assertIn("height=720", probe.stdout)
|
|
print(f"\n Handshake: {' -> '.join(tv.transcript)}")
|
|
print(f" Bild: {len(payload)} Byte in {len(sequences)} RTP-Paketen, "
|
|
f"{gaps} Lücken, dekodiert als {probe.stdout.strip().splitlines()}")
|
|
finally:
|
|
streamer.stop()
|
|
server.stop()
|
|
rtp_socket.close()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|