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