miracast-client/linux/wfd/p2p.py

"""Wi-Fi Direct und die Anmeldung als Wi-Fi-Display-Quelle.

Hier liegt der entscheidende Unterschied zur Android-Fassung: Über
wpa_supplicant lässt sich das WFD-Informationselement wirklich setzen. Damit
erkennt der Fernseher uns als Miracast-Quelle und baut die Steuerverbindung
von sich aus auf. Auf Android ist genau das Systemapps vorbehalten.
"""

import fcntl
import logging
import re
import shutil
import socket
import struct
import subprocess
import time
from dataclasses import dataclass
from typing import Dict, List, Optional

from .wpa import WpaClient, parse_key_values

log = logging.getLogger("wfd.p2p")

# WFD-Gerätetyp in den unteren zwei Bits des Informationsfeldes.
DEVICE_TYPE_SOURCE = 0b00
DEVICE_TYPE_PRIMARY_SINK = 0b01
SESSION_AVAILABLE = 0b01 << 4


@dataclass
class Peer:
    address: str
    name: str
    is_sink: bool
    session_available: bool
    control_port: int
    device_type: str = ""

    def describe(self) -> str:
        marks = []
        if self.is_sink:
            marks.append("Bildschirm")
        if self.session_available:
            marks.append("wartet auf Verbindung")
        suffix = f" ({', '.join(marks)})" if marks else ""
        return f"{self.name} [{self.address}]{suffix}"


@dataclass
class Group:
    interface: str
    role: str            # "GO" oder "client"
    go_address: str = ""
    local_address: str = ""


def parse_wfd_dev_info(value: str) -> Dict[str, object]:
    """Zerlegt wfd_dev_info, z.B. 0x00111c440032.

    Die ersten zwei Bytes sind das Informationsfeld, dann folgt der Port für
    die Steuerverbindung und der maximale Durchsatz.
    """
    text = value.strip()
    if text.startswith("0x"):
        text = text[2:]
    if len(text) < 4:
        return {}
    info = int(text[0:4], 16)
    port = int(text[4:8], 16) if len(text) >= 8 else 7236
    device_type = info & 0b11
    return {
        "info": info,
        "device_type": device_type,
        "is_sink": device_type in (DEVICE_TYPE_PRIMARY_SINK, 0b10, 0b11),
        "session_available": bool((info >> 4) & 0b11),
        "control_port": port or 7236,
    }


def interface_ip(interface: str) -> str:
    """Eigene IPv4-Adresse eines Interfaces, ohne externe Werkzeuge."""
    try:
        sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
        packed = struct.pack("256s", interface[:15].encode())
        return socket.inet_ntoa(fcntl.ioctl(sock.fileno(), 0x8915, packed)[20:24])
    except OSError:
        return ""


def default_gateway(interface: str) -> str:
    """Gegenstelle der Gruppe - beim Wi-Fi-Direct-Client also der Fernseher."""
    try:
        with open("/proc/net/route", encoding="ascii") as handle:
            for line in handle.readlines()[1:]:
                fields = line.split()
                if fields[0] != interface or fields[1] != "00000000":
                    continue
                return socket.inet_ntoa(struct.pack("<L", int(fields[2], 16)))
    except OSError:
        pass
    return ""


class P2pManager:
    def __init__(self, client: WpaClient):
        self.client = client

    # ------------------------------------------------- Als Quelle anmelden

    def enable_wfd_source(self, control_port: int = 7236, max_throughput: int = 50) -> bool:
        """Kündigt uns als Miracast-Quelle an, die eine Sitzung annehmen kann."""
        if not self.client.ok("SET wifi_display 1"):
            log.warning("wifi_display liess sich nicht einschalten")
            return False
        info = DEVICE_TYPE_SOURCE | SESSION_AVAILABLE
        payload = f"{info:04x}{control_port:04x}{max_throughput:04x}"
        # Subelement 0 ist die Geräteinformation, davor steht ihre Länge (6 Byte).
        if not self.client.ok(f"WFD_SUBELEM_SET 0 0006{payload}"):
            log.warning("WFD-Kennzeichen liess sich nicht setzen")
            return False
        log.info("Als Wi-Fi-Display-Quelle angemeldet (Port %s)", control_port)
        return True

    def disable_wfd(self):
        self.client.ok("SET wifi_display 0")

    # ------------------------------------------------------------- Suchen

    def find(self, seconds: int = 30):
        self.client.ok(f"P2P_FIND {seconds}")

    def stop_find(self):
        self.client.ok("P2P_STOP_FIND")

    def peers(self) -> List[Peer]:
        raw = self.client.request("P2P_PEERS")
        result = []
        for address in raw.split():
            if not re.fullmatch(r"(?:[0-9a-fA-F]{2}:){5}[0-9a-fA-F]{2}", address):
                continue
            details = parse_key_values(self.client.request(f"P2P_PEER {address}"))
            wfd = parse_wfd_dev_info(details.get("wfd_dev_info", ""))
            result.append(Peer(
                address=address,
                name=details.get("device_name", address),
                is_sink=bool(wfd.get("is_sink", False)),
                session_available=bool(wfd.get("session_available", False)),
                control_port=int(wfd.get("control_port", 7236)),
                device_type=details.get("pri_dev_type", ""),
            ))
        return result

    # ---------------------------------------------------------- Verbinden

    def connect(self, peer: Peer, go_intent: int = 0, method: str = "pbc") -> str:
        """Stellt die Direktverbindung her. go_intent=0 heißt: der Fernseher
        soll die Gruppe führen - dann bekommen wir von ihm eine Adresse."""
        command = f"P2P_CONNECT {peer.address} {method} go_intent={go_intent}"
        answer = self.client.request(command, timeout=10)
        log.info("P2P_CONNECT -> %s", answer)
        return answer

    def wait_for_group(self, timeout: float = 60.0) -> Optional[Group]:
        """Wartet auf den Gruppenstart und liefert Interface und Rolle."""
        event = self.client.wait_for_event("P2P-GROUP-STARTED", timeout=timeout)
        if not event:
            return None
        # Beispiel: P2P-GROUP-STARTED p2p-wlan0-0 client ssid="DIRECT-xy" ...
        parts = event.split()
        if len(parts) < 3:
            return None
        interface = parts[1]
        role = "GO" if parts[2].upper() == "GO" else "client"
        return Group(interface=interface, role=role)

    def remove_group(self, interface: str = ""):
        self.client.ok(f"P2P_GROUP_REMOVE {interface or '*'}")


def obtain_address(interface: str, timeout: int = 20) -> str:
    """Holt per DHCP eine Adresse in der Gruppe.

    wpa_supplicant kümmert sich nicht um IP-Adressen; als Gruppen-Gast müssen
    wir selbst fragen. Wir probieren die üblichen Clients der Reihe nach.
    """
    existing = interface_ip(interface)
    if existing:
        return existing

    candidates = [
        (["dhclient", "-1", "-v", interface], 25),
        (["udhcpc", "-i", interface, "-q", "-n", "-t", "6"], 20),
        (["dhcpcd", "-4", "-t", str(timeout), interface], timeout + 5),
    ]
    for command, limit in candidates:
        if not shutil.which(command[0]):
            continue
        log.info("Hole Adresse mit %s", command[0])
        try:
            subprocess.run(command, timeout=limit, check=False,
                           stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
        except (subprocess.TimeoutExpired, OSError) as error:
            log.debug("%s erfolglos: %s", command[0], error)
        address = interface_ip(interface)
        if address:
            return address

    # Letzter Ausweg: die übliche Aufteilung einer Wi-Fi-Direct-Gruppe.
    deadline = time.monotonic() + 5
    while time.monotonic() < deadline:
        address = interface_ip(interface)
        if address:
            return address
        time.sleep(0.5)
    return ""