miracast-client/linux/miracast-cast

#!/usr/bin/env python3
"""Bildschirm auf einen wartenden Fernseher spiegeln (Miracast-Quelle).

    sudo ./miracast-cast check     Voraussetzungen prüfen
    sudo ./miracast-cast scan      Fernseher suchen
    sudo ./miracast-cast cast      verbinden und spiegeln

Ohne Argumente verhält sich das Programm wie "cast".
"""

import argparse
import logging
import os
import shutil
import signal
import sys
import threading
import time

sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

from wfd import p2p, rtsp, wpa                      # noqa: E402
from wfd.formats import VideoFormat                 # noqa: E402
from wfd.streamer import Streamer                   # noqa: E402

log = logging.getLogger("miracast")


def setup_logging(verbose: bool):
    logging.basicConfig(
        level=logging.DEBUG if verbose else logging.INFO,
        format="%(asctime)s  %(message)s",
        datefmt="%H:%M:%S",
    )


# --------------------------------------------------------------------- check

def command_check(args) -> int:
    print("Voraussetzungen für Miracast\n")
    ok = True

    wireless = wpa.wireless_interfaces()
    if wireless:
        print(f"  [ok]     WLAN-Gerät: {', '.join(wireless)}")
    else:
        print("  [FEHLT]  Kein WLAN-Gerät gefunden.")
        print("           Miracast läuft über Wi-Fi Direct - ohne WLAN-Karte geht es nicht.")
        ok = False

    sockets = wpa.find_interfaces()
    if sockets:
        print(f"  [ok]     wpa_supplicant erreichbar: {', '.join(sockets)}")
    else:
        print("  [FEHLT]  Kein Control-Socket von wpa_supplicant gefunden.")
        print("           Als root ausführen. Falls NetworkManager das Gerät verwaltet:")
        print("             nmcli device set <wlan> managed no")
        print("             wpa_supplicant -B -i <wlan> -c /etc/wpa_supplicant/wpa_supplicant.conf \\")
        print("                            -C /run/wpa_supplicant")
        ok = False

    if os.geteuid() == 0:
        print("  [ok]     Läuft mit Rootrechten")
    else:
        print("  [Hinweis] Ohne root fehlt meist der Zugriff auf wpa_supplicant")

    if shutil.which("ffmpeg"):
        print("  [ok]     ffmpeg vorhanden")
    else:
        print("  [FEHLT]  ffmpeg fehlt - ohne es gibt es kein Bild (apt install ffmpeg)")
        ok = False

    if any(shutil.which(tool) for tool in ("dhclient", "udhcpc", "dhcpcd")):
        print("  [ok]     DHCP-Client vorhanden")
    else:
        print("  [Hinweis] Kein DHCP-Client - in der Gruppe fehlt dann evtl. die Adresse")

    session = os.environ.get("XDG_SESSION_TYPE", "unbekannt")
    if os.environ.get("DISPLAY"):
        print(f"  [ok]     Bildquelle: X11 auf {os.environ['DISPLAY']} (Sitzung: {session})")
    else:
        print(f"  [Hinweis] Kein DISPLAY gesetzt (Sitzung: {session}).")
        print("           Bei Wayland mit '--source kms' arbeiten oder '--source test' zum Prüfen.")

    # Auf einer WFD-Quelle muss Port 7236 frei sein.
    import socket as socketlib
    probe = socketlib.socket(socketlib.AF_INET, socketlib.SOCK_STREAM)
    probe.setsockopt(socketlib.SOL_SOCKET, socketlib.SO_REUSEADDR, 1)
    try:
        probe.bind(("0.0.0.0", 7236))
        print("  [ok]     Port 7236 ist frei")
    except OSError as error:
        print(f"  [FEHLT]  Port 7236 belegt: {error}")
        ok = False
    finally:
        probe.close()

    print("\n" + ("Alles bereit." if ok else "Es fehlt noch etwas - siehe oben."))
    return 0 if ok else 1


# ---------------------------------------------------------------- Verbindung

def open_client(interface: str = "") -> wpa.WpaClient:
    sockets = wpa.find_interfaces()
    if not sockets:
        raise SystemExit("Kein wpa_supplicant erreichbar. './miracast-cast check' zeigt, was fehlt.")
    if interface:
        matching = [path for path in sockets if os.path.basename(path) == interface]
        if not matching:
            raise SystemExit(f"Interface {interface} hat keinen Control-Socket. Vorhanden: {sockets}")
        return wpa.WpaClient(matching[0])
    # p2p-dev-* ist der Steuerkanal für Wi-Fi Direct und hat Vorrang.
    sockets.sort(key=lambda path: 0 if "p2p-dev" in path else 1)
    return wpa.WpaClient(sockets[0])


def discover(manager: p2p.P2pManager, seconds: int, only_sinks: bool = True):
    manager.find(seconds)
    deadline = time.monotonic() + seconds
    seen = {}
    while time.monotonic() < deadline:
        for peer in manager.peers():
            if only_sinks and not peer.is_sink:
                continue
            if peer.address not in seen:
                seen[peer.address] = peer
                print(f"  gefunden: {peer.describe()}")
        time.sleep(2)
    manager.stop_find()
    return list(seen.values())


def command_scan(args) -> int:
    client = open_client(args.interface)
    client.attach()
    manager = p2p.P2pManager(client)
    manager.enable_wfd_source()
    print(f"Suche {args.timeout} Sekunden nach Fernsehern ...")
    peers = discover(manager, args.timeout, only_sinks=not args.all)
    if not peers:
        print("Nichts gefunden. Steht am Fernseher der Screen-Mirroring-Bildschirm offen?")
    client.close()
    return 0 if peers else 1


def command_cast(args) -> int:
    client = open_client(args.interface)
    client.attach()
    manager = p2p.P2pManager(client)

    if not manager.enable_wfd_source(control_port=args.port):
        print("Warnung: Das WFD-Kennzeichen liess sich nicht setzen. "
              "Der Fernseher erkennt uns dann möglicherweise nicht als Quelle.")

    # Fernseher auswählen
    peer = None
    if args.peer:
        peer = p2p.Peer(address=args.peer, name=args.peer, is_sink=True,
                        session_available=True, control_port=args.port)
    else:
        print(f"Suche {args.timeout} Sekunden nach Fernsehern ...")
        peers = discover(manager, args.timeout, only_sinks=not args.all)
        if not peers:
            print("Kein Fernseher gefunden.")
            client.close()
            return 1
        if len(peers) == 1:
            peer = peers[0]
            print(f"Nehme {peer.describe()}")
        else:
            for index, candidate in enumerate(peers, 1):
                print(f"  {index}) {candidate.describe()}")
            choice = input("Welcher Fernseher? ").strip()
            peer = peers[int(choice) - 1] if choice.isdigit() else peers[0]

    streamer = Streamer(source=args.source, display=args.display, bitrate_mbit=args.bitrate)
    finished = threading.Event()
    state = {"format": None, "group": None}

    def on_play(host: str, port: int, fmt: VideoFormat):
        state["format"] = fmt
        print(f"\n>>> Übertragung läuft: {fmt} an {host}:{port}\n")
        streamer.start(host, port, fmt)

    def on_stopped(reason: str):
        print(f"\n>>> Ende: {reason}")
        finished.set()

    def on_status(text: str):
        print(f"    {text}")

    # Erst zuhören, dann verbinden: Der Fernseher meldet sich sofort, wenn
    # die Gruppe steht - da muss der Kanal schon offen sein.
    server = rtsp.SourceServer("0.0.0.0", args.max_height, on_play, on_stopped, on_status,
                               ports=(args.port, 8554))
    opened = server.start()
    if not opened:
        print("Konnte keinen Steuerkanal öffnen (Port 7236 belegt?).")
        client.close()
        return 1

    print(f"Verbinde mit {peer.describe()} ... am Fernseher ggf. bestätigen.")
    manager.connect(peer, go_intent=args.go_intent)

    group = manager.wait_for_group(timeout=args.connect_timeout)
    if not group:
        print("Die Direktverbindung kam nicht zustande.")
        server.stop(); client.close()
        return 1
    print(f"Gruppe steht auf {group.interface} (Rolle: {group.role})")

    address = p2p.obtain_address(group.interface)
    if address:
        group.local_address = address
        print(f"Eigene Adresse in der Gruppe: {address}")
        # Für die Präsentations-URL im Handshake
        if server.session:
            server.session.local_address = address
        gateway = p2p.default_gateway(group.interface)
        if gateway:
            group.go_address = gateway
            print(f"Fernseher vermutlich unter {gateway}")
            server.try_outgoing(gateway, peer.control_port)
    else:
        print("Keine Adresse erhalten - läuft ein DHCP-Client?")

    print("Warte darauf, dass der Fernseher die Sitzung aufbaut ...")

    def keep_alive():
        while not finished.is_set():
            if finished.wait(25):
                return
            if server.session:
                server.session.keep_alive()

    threading.Thread(target=keep_alive, daemon=True, name="keepalive").start()

    def shutdown(signum=None, frame=None):
        finished.set()

    signal.signal(signal.SIGINT, shutdown)
    signal.signal(signal.SIGTERM, shutdown)

    try:
        while not finished.is_set():
            finished.wait(1)
            if streamer.process and not streamer.is_running():
                print(">>> Der Bildstrom ist beendet.")
                break
    finally:
        print("Räume auf ...")
        streamer.stop()
        server.stop()
        manager.remove_group(state["group"].interface if state["group"] else "")
        manager.disable_wfd()
        client.close()
    return 0


def main() -> int:
    parser = argparse.ArgumentParser(
        description="Bildschirm auf einen Miracast-Fernseher spiegeln",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog=__doc__)
    parser.add_argument("-v", "--verbose", action="store_true", help="ausführliche Ausgabe")
    parser.add_argument("-i", "--interface", default="", help="WLAN-Interface (sonst automatisch)")
    sub = parser.add_subparsers(dest="command")

    sub.add_parser("check", help="Voraussetzungen prüfen")

    scan = sub.add_parser("scan", help="Fernseher suchen")
    scan.add_argument("-t", "--timeout", type=int, default=20, help="Suchdauer in Sekunden")
    scan.add_argument("--all", action="store_true", help="auch Geräte zeigen, die keine Bildschirme sind")

    cast = sub.add_parser("cast", help="verbinden und spiegeln")
    cast.add_argument("--peer", default="", help="MAC-Adresse des Fernsehers")
    cast.add_argument("-t", "--timeout", type=int, default=20, help="Suchdauer in Sekunden")
    cast.add_argument("--all", action="store_true", help="auch Nicht-Bildschirme anbieten")
    cast.add_argument("--source", default="auto", choices=["auto", "x11", "kms", "test"],
                      help="Bildquelle; 'test' schickt ein Testbild")
    cast.add_argument("--display", default=None, help="X11-Display, z.B. :0")
    cast.add_argument("--bitrate", type=int, default=8, help="Bitrate in Mbit/s")
    cast.add_argument("--max-height", type=int, default=1080, help="größte erlaubte Bildhöhe")
    cast.add_argument("--port", type=int, default=7236, help="Port für die Steuerverbindung")
    cast.add_argument("--go-intent", type=int, default=0,
                      help="0 = der Fernseher führt die Gruppe (Standard)")
    cast.add_argument("--connect-timeout", type=int, default=60,
                      help="Wartezeit auf die Direktverbindung")

    args = parser.parse_args()
    setup_logging(args.verbose)

    if args.command == "check":
        return command_check(args)
    if args.command == "scan":
        return command_scan(args)
    if args.command in (None, "cast"):
        if args.command is None:
            args = parser.parse_args(["cast"])
            setup_logging(False)
        return command_cast(args)
    parser.print_help()
    return 1


if __name__ == "__main__":
    sys.exit(main())