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