#!/usr/bin/env python3
"""Dokki Screen's capture helper for Linux hosts (the twin of `DokkiHelper screen-stream`).

    dokki-screen probe
    dokki-screen stream --source x11|portal|synthetic --mode mirror|extend --fps N --bitrate BPS
                        --encoder auto|vah264enc|vaapih264enc|nvh264enc|x264enc|openh264enc
                        --cursor 0|1 --input 0|1 --box WxH --max-long-side N [--size WxH]
                        [--display :0 --region X,Y,W,H] [--token-file PATH] [--crop X,Y,W,H] [--agent 1]
    dokki-screen input-x11 --display :0 --region X,Y,W,H

src/platform/linux-screen.js starts it; docs/screen-linux.md describes the whole Linux sender.

stdout (stream): records  type u8 | length u32 BE | payload
    1 CONFIG  u16 width | u16 height | avcC (AVCDecoderConfigurationRecord)
    2 FRAME   u64 captureUs (CLOCK_MONOTONIC, Node's hrtime) | u32 encodeUs | u8 flags (bit0 key) | AVCC AU
    3 STATUS  JSON: {"t":"ready",...} | {"t":"error","code","message"} | {"t":"locked"} | {"t":"local"} (agent)
    4 SNAPSHOT (agent) u32 id | u16 width | u16 height | u8 format (1 jpeg, 2 png) | image: exactly the
              dokki-screen-v1 SNAPSHOT (0x09) payload src/agent-screen.js reads
stdin: one command per line, integers only (anything else is ignored)
    K | R <bps> | Z <w> <h> | Q                       keyframe now, bitrate, client resized, quit
    p <u> <v> | b <1|2|3> <0|1> | k <evdev> <0|1> | s <h> <v> | x    input (see linux-screen.js)
    g <phase> <a>                                     a pinch (INPUT kind 8, protocol 4.2): magnification
                                                      change x 10000; phase 1 began, 0 changed, 2 ended,
                                                      3 cancelled. Turned here into Ctrl + wheel notches.
    z <notches>                                       Ctrl + vertical wheel, > 0 up (zoom in)
    S <id> <maxWidth> <1|2>                           (agent) the latest picture, scaled, as JPEG or PNG

--agent 1 (Agentes en la pantalla, docs/agent-screen.md): the same capture and input, but no encoder and
no video. The latest picture (RGBx, at most 10 a second) is kept; `S` encodes it (PIL, else GStreamer's
jpegenc/pngenc, else a PNG made here with zlib). On X11 the owner's own input (XInput2 raw events that do
not come from the XTEST devices, else XScreenSaver's idle time against our own last event) is reported as
{"t":"local"}, at most every 250 ms. Wayland cannot tell: no local reports there.

Wayland: xdg-desktop-portal ScreenCast (with RemoteDesktop when input is on, one dialog, a
restore token so it is asked once), or for --mode extend a virtual monitor: the portal's
VIRTUAL source type where the portal has it (KDE, GNOME 48; on GNOME sized by our caps), else
Mutter's own RecordVirtual on an older GNOME.
--crop (Dokki Beta's screen wall, basis points 0..10000 of the picture; portal mirror and the test
pattern only): a videocrop right after the source, so only that part is scaled and encoded, and
pointer input lands inside that part of the stream.
X11: GStreamer ximagesrc over the region Node chose (a monitor, or Node's xrandr virtual
monitor), input through XTest. The encoder is the first H.264 element that works, hardware
first; one that fails before its first frame is replaced by the next.
"""

import collections
import json
import math
import os
import signal
import struct
import sys
import threading
import time

ENCODERS = ["vah264enc", "vaapih264enc", "nvh264enc", "x264enc", "openh264enc"]
SOURCE_MONITOR, SOURCE_WINDOW, SOURCE_VIRTUAL = 1, 2, 4
CURSOR_HIDDEN, CURSOR_EMBEDDED, CURSOR_METADATA = 1, 2, 4
X_BUTTONS = {1: 1, 2: 2, 3: 3}  # left, middle, right (linux-screen.js: record bit0 -> 1, bit1 -> 3, bit2 -> 2)
EVDEV_BUTTONS = {1: 272, 2: 274, 3: 273}  # BTN_LEFT, BTN_MIDDLE (0x112), BTN_RIGHT
CTRL_KEYS = (29, 97)  # KEY_LEFTCTRL, KEY_RIGHTCTRL
GESTURE_CHANGED, GESTURE_BEGAN, GESTURE_ENDED, GESTURE_CANCELLED = 0, 1, 2, 3
# One wheel notch per 10 % of scale: the step Ctrl + wheel zooms by in browsers, GTK and LibreOffice.
NOTCH_LN = math.log(1.1)
# Mutter (GNOME 48) gives a remote session's first pointer event and first key press to the
# client as wl_pointer/wl_keyboard "enter" (a new input device), not as motion or a key: the
# phone's first tap or letter would be lost. A zero move and KEY_UNKNOWN (no keysym in any
# layout) are spent on that first, invisible to apps, right after the session starts.
PRIME_KEY = 240  # KEY_UNKNOWN
# evdev Ctrl, Shift, Alt, AltGr, Super (left/right): let go around typed text, pressed again after.
MODIFIER_KEYS = (29, 97, 42, 54, 56, 100, 125, 126)
SPARE_KEYCODES = 8  # the X11 ring of scratch keycodes for typed text
REC_CONFIG, REC_FRAME, REC_STATUS, REC_SNAPSHOT = 1, 2, 3, 4
SNAP_JPEG, SNAP_PNG = 1, 2
# A SNAPSHOT record fits src/platform/linux-screen-codec.js's record cap (4 MiB less a frame header and 1 KiB).
SNAP_MAX_IMAGE = 4 * 1024 * 1024 - 20 - 1024 - 9 - 1024
LOCAL_INTERVAL_S = 0.25
LOCAL_MOVE_PX = 15
START_TIMEOUT_S = 12
# A PipeWire stream that went quiet: pipewiresrc repeats the last picture every second
# (keepalive-time), so this long without one means the stream is stuck, not the screen still.
# xdg-desktop-portal-wlr 0.7 sometimes keeps every buffer after the first one ("out of
# buffers"): the stream is made again on the same portal session, a few times at most.
STALL_S = 5
STALL_RESTARTS = 3


# ---- pure helpers (native/linux/tests/test_dokki_screen.py) ----

def even(n):
    return max(16, int(n) // 2 * 2)


def fit_size(src_w, src_h, box_w, box_h, max_long=2752):
    scale = min(box_w / src_w, box_h / src_h, max_long / max(src_w, src_h), 1.0)
    return even(src_w * scale), even(src_h * scale)


def parse_wh(text):
    w, h = text.split("x")
    w, h = int(w), int(h)
    if not (16 <= w <= 8192 and 16 <= h <= 8192):
        raise ValueError("size out of range")
    return w, h


def parse_region(text):
    x, y, w, h = (int(v) for v in text.split(","))
    if not (-32768 <= x <= 32767 and -32768 <= y <= 32767 and 1 <= w <= 16384 and 1 <= h <= 16384):
        raise ValueError("region out of range")
    return x, y, w, h


def parse_crop(text):
    """--crop X,Y,W,H in basis points: integers, each side >= 100, inside 0..10000 (src/platform/screen.js cropArg)."""
    parts = text.split(",")
    if len(parts) != 4 or not all(p.isdigit() and p.isascii() and len(p) <= 5 for p in parts):
        raise ValueError("bad --crop")
    x, y, w, h = (int(p) for p in parts)
    if w < 100 or h < 100 or x + w > 10000 or y + h > 10000:
        raise ValueError("bad --crop")
    return x, y, w, h


def crop_pixels(src_w, src_h, crop):
    """A crop of a src_w x src_h picture in whole pixels, even sizes of at least 2, inside it (the
    numbers of linux-screen.js cropRegion and the Windows sender's ScreenCrop.Pixels). No crop: all of it."""
    if not crop or src_w < 2 or src_h < 2:
        return 0, 0, src_w, src_h
    x, y, w, h = crop

    def down2(v):
        return max(2, int(v // 2) * 2)
    cw = min(down2(src_w * w / 10000), down2(src_w))
    ch = min(down2(src_h * h / 10000), down2(src_h))
    cx = min(int(src_w * x // 10000), src_w - cw)
    cy = min(int(src_h * y // 10000), src_h - ch)
    return cx, cy, cw, ch


def encoder_description(name, fps, bitrate):
    """The GStreamer encoder (and the raw format it wants), low latency, no B-frames."""
    kbps = max(100, int(bitrate // 1000))
    gop = int(fps) * 10
    if name == "x264enc":
        # Sliced threads (zerolatency's own choice): frame threads would hold one picture per thread
        # (~1.5 x the cores, 14 on an 8-core) before the first output. A compositor sends pictures
        # only when the screen changes, so a still GNOME/KDE desktop would never start, and every
        # change would show that many changes late.
        return "I420", ("x264enc name=enc tune=zerolatency speed-preset=ultrafast bframes=0 b-adapt=false rc-lookahead=0 "
                        "key-int-max=%d bitrate=%d vbv-buf-capacity=250 byte-stream=false aud=false sliced-threads=true" % (gop, kbps))
    if name == "openh264enc":
        return "I420", "openh264enc name=enc usage-type=screen complexity=low rate-control=bitrate gop-size=%d bitrate=%d" % (gop, kbps * 1000)
    if name == "vah264enc":
        return "NV12", "vah264enc name=enc rate-control=cbr b-frames=0 key-int-max=%d bitrate=%d target-usage=7" % (gop, kbps)
    if name == "vaapih264enc":
        return "NV12", "vaapih264enc name=enc rate-control=cbr max-bframes=0 keyframe-period=%d bitrate=%d tune=low-power" % (gop, kbps)
    if name == "nvh264enc":
        return "NV12", "nvh264enc name=enc rc-mode=cbr bframes=0 zerolatency=true preset=low-latency-hq gop-size=%d bitrate=%d" % (gop, kbps)
    raise ValueError("unknown encoder")


def source_description(source, fps, cursor, display=None, region=None, fd=None, node=None, size=None):
    if source == "x11":
        x, y, w, h = region
        return ("ximagesrc display-name=%s startx=%d starty=%d endx=%d endy=%d use-damage=false show-pointer=%s "
                "! video/x-raw,framerate=%d/1" % (display, max(0, x), max(0, y), x + w - 1, y + h - 1, "true" if cursor else "false", fps))
    if source == "synthetic":
        return "videotestsrc is-live=true pattern=smpte ! video/x-raw,framerate=%d/1,width=1920,height=1080" % fps
    if source == "pipewire":
        parts = ["pipewiresrc", "name=pwsrc", "path=%d" % int(node), "do-timestamp=true", "keepalive-time=1000", "always-copy=true"]
        if fd is not None:
            parts.insert(1, "fd=%d" % int(fd))
        # No framerate in the caps: compositors send frames as the screen changes (variable rate),
        # and a fixed or max framerate a producer does not advertise fails PipeWire's negotiation.
        if not size:
            return " ".join(parts)
        # A virtual monitor takes the size the consumer asks for.
        return " ".join(parts) + " ! video/x-raw,width=%d,height=%d" % size
    raise ValueError("unknown source")


# Encoders that may only write Annex-B: they need h264parse (gstreamer1.0-plugins-bad) to become
# AVCC. Only x264enc writes AVC with codec_data itself (byte-stream=false).
BYTE_STREAM_ONLY = {"vah264enc", "vaapih264enc", "nvh264enc", "openh264enc"}


def usable_encoders(found, has_parse):
    return [e for e in ENCODERS if e in found and (has_parse or e not in BYTE_STREAM_ONLY)]


def pipeline_description(src, encoder, fps, bitrate, size, parse=True, crop=False):
    fmt, enc = encoder_description(encoder, fps, bitrate)
    w, h = size
    # AVCC access units with the avcC in the caps (codec_data): h264parse when there is one,
    # else straight from an encoder that can write AVC itself (x264enc, vah264enc, nvh264enc).
    tail = "h264parse config-interval=-1 ! " if parse else ""
    if crop:
        # The screen wall: the part this device shows, cut before the scaler (Streamer.apply_crop sets it).
        src += " ! videocrop name=crop"
    return ("%s ! queue max-size-buffers=1 leaky=downstream ! videoscale ! videoconvert "
            "! capsfilter name=size caps=video/x-raw,format=%s,width=%d,height=%d,pixel-aspect-ratio=1/1 "
            "! %s ! %svideo/x-h264,stream-format=avc,alignment=au "
            "! appsink name=sink emit-signals=true sync=false max-buffers=4 drop=false" % (src, fmt, w, h, enc, tail))


def record(kind, payload):
    return struct.pack(">BI", kind, len(payload)) + payload


def config_record(width, height, avcc):
    return record(REC_CONFIG, struct.pack(">HH", width, height) + bytes(avcc))


def frame_record(capture_us, encode_us, keyframe, data):
    return record(REC_FRAME, struct.pack(">QIB", max(0, int(capture_us)), max(0, min(0xFFFFFFFF, int(encode_us))), 1 if keyframe else 0) + bytes(data))


def status_record(obj):
    return record(REC_STATUS, json.dumps(obj, separators=(",", ":")).encode())


def text_codepoint(text):
    """`t HEX`'s code point: 1..6 lowercase or uppercase hex digits, a typable character, else None."""
    if not 1 <= len(text) <= 6 or any(c not in "0123456789abcdefABCDEF" for c in text):
        return None
    cp = int(text, 16)
    if cp < 0x20 or 0x7F <= cp <= 0x9F or 0xD800 <= cp <= 0xDFFF or cp > 0x10FFFF:
        return None  # controls (C0, DEL, C1), surrogate halves, past Unicode
    return cp


def text_keysym(cp):
    """The X keysym that types a code point: Latin-1's own (ASCII and U+00A0..U+00FF), else 0x01000000 | cp."""
    if 0x20 <= cp <= 0x7E or 0xA0 <= cp <= 0xFF:
        return cp
    return 0x01000000 | cp


def spare_keycodes(keysyms, first, per, count=SPARE_KEYCODES):
    """Keycodes with no keysym at all, from an XGetKeyboardMapping table (`per` keysyms per keycode from
    `first`), the highest first, at most `count`; none free: the highest keycode (its mapping is put back)."""
    total = len(keysyms) // per if per else 0
    free = [first + i for i in range(total - 1, -1, -1) if not any(keysyms[i * per:(i + 1) * per])]
    return free[:count] if free else ([first + total - 1] if total else [])


def parse_command(line):
    """One stdin line -> a tuple, or None. Integers only, bounded."""
    parts = line.strip().split()
    if not parts or len(parts) > 4:
        return None
    op, rest = parts[0], parts[1:]
    if op == "t":
        cp = text_codepoint(rest[0]) if len(rest) == 1 else None
        return ("t", cp) if cp is not None else None
    try:
        nums = [int(v) for v in rest]
    except ValueError:
        return None
    shapes = {"K": 0, "Q": 0, "x": 0, "R": 1, "Z": 2, "p": 2, "b": 2, "k": 2, "s": 2, "S": 3, "g": 2, "z": 1}
    if op not in shapes or len(nums) != shapes[op]:
        return None
    if op == "R" and not 100_000 <= nums[0] <= 100_000_000:
        return None
    if op == "Z" and not all(64 <= n <= 8192 for n in nums):
        return None
    if op == "p" and not all(0 <= n <= 65535 for n in nums):
        return None
    if op == "b" and (nums[0] not in X_BUTTONS or nums[1] not in (0, 1)):
        return None
    if op == "k" and (not 1 <= nums[0] <= 767 or nums[1] not in (0, 1)):
        return None
    if op == "s" and not all(-50 <= n <= 50 for n in nums):
        return None
    if op == "g" and (nums[0] not in (0, 1, 2, 3) or not -10000 <= nums[1] <= 10000):
        return None
    if op == "z" and not -50 <= nums[0] <= 50:
        return None
    if op == "S" and not (0 <= nums[0] <= 0xFFFFFFFF and 16 <= nums[1] <= 8192 and nums[2] in (SNAP_JPEG, SNAP_PNG)):
        return None
    return (op, *nums)


def to_pixels(u, v, w, h):
    """Normalized 0..65535 -> a pixel in a w x h area (the far edge is the last pixel)."""
    return round(u * (w - 1) / 65535), round(v * (h - 1) / 65535)


class MagnifyNotches:
    """A pinch -> whole Ctrl + wheel notches. XTest, the RemoteDesktop portal and Mutter have no gesture
    events, so a pinch becomes what a mouse user does to zoom. `a` is each record's scale change x 10000,
    so the logs add up to ln(total scale): a notch per ln(1.1), the remainder carried to the next record so
    a slow pinch still zooms; each gesture starts from zero (no part-notch outlives its gesture)."""

    def __init__(self):
        self.carry = 0.0

    def reset(self):
        self.carry = 0.0

    def push(self, phase, a):
        if phase in (GESTURE_BEGAN, GESTURE_CANCELLED):
            self.reset()
        if phase == GESTURE_CANCELLED or not a:
            if phase == GESTURE_ENDED:
                self.reset()
            return 0
        total = self.carry + math.log(1 + max(-9999, min(10000, a)) / 10000) / NOTCH_LN  # -10000 would be ln(0)
        # Toward zero (a pinch turning back first pays off the carry); the nudge keeps 1.1^2 from being 1.999...
        notches = int(total + math.copysign(1e-9, total))
        self.carry = 0.0 if phase == GESTURE_ENDED else total - notches
        return max(-50, min(50, notches))


def zoom_needs_ctrl(keys):
    """Ctrl is pressed for a zoom only when the session does not hold one already: letting it go after
    the wheel would otherwise lift the phone's own Ctrl."""
    return not any(k in keys for k in CTRL_KEYS)


def gestures_caps(input_ok):
    """What `helper.ready` says this sender can post (protocol 4.2): a pinch as Ctrl + wheel and the middle
    button. No rotate (XTest, the portal and Mutter have no gesture events; only a uinput virtual touchpad
    that libinput reads as a pinch could) and no scroll phases (an X11 or evdev wheel has none)."""
    return {"magnify": bool(input_ok), "rotate": False, "phases": False, "middle": bool(input_ok)}


def zoom_command(magnify, cmd):
    """A `g` line -> its `z` notches (None while they do not add up to one); `x` also ends the pinch."""
    if cmd[0] == "x":
        magnify.reset()
    if cmd[0] != "g":
        return cmd
    notches = magnify.push(cmd[1], cmd[2])
    return ("z", notches) if notches else None


def request_path(unique_name, token):
    sender = unique_name.lstrip(":").replace(".", "_")
    return "/org/freedesktop/portal/desktop/request/%s/%s" % (sender, token)


# ---- agent mode, pure (docs/agent-screen.md) ----

def agent_pipeline_description(src, size, crop=False):
    """The agent's capture: no encoder, RGBx pictures (4 bytes a pixel, rows never padded) for snapshots."""
    w, h = size
    if crop:
        src += " ! videocrop name=crop"
    return ("%s ! queue max-size-buffers=1 leaky=downstream ! videoscale ! videoconvert "
            "! capsfilter name=size caps=video/x-raw,format=RGBx,width=%d,height=%d,pixel-aspect-ratio=1/1 "
            "! appsink name=sink emit-signals=true sync=false max-buffers=1 drop=true" % (src, w, h))


def snapshot_size(src_w, src_h, max_width):
    """The picture scaled to max_width: aspect kept, never upscaled, at least 1 px (the Windows sender's numbers)."""
    if src_w <= 0 or src_h <= 0:
        return 0, 0
    w = max(1, min(src_w, max_width, 65535))
    h = max(1, int(src_h * w / src_w + 0.5))
    if h > 65535:
        w, h = max(1, w * 65535 // h), 65535
    return w, h


def snapshot_record(snap_id, width, height, fmt, image):
    return record(REC_SNAPSHOT, struct.pack(">IHHB", snap_id, width, height, fmt) + bytes(image))


def rgbx_to_rgb(row):
    """One RGBx row -> RGB (extended slices: no per-pixel Python loop)."""
    out = bytearray(len(row) // 4 * 3)
    out[0::3] = row[0::4]
    out[1::3] = row[1::4]
    out[2::3] = row[2::4]
    return out


def scale_rgbx_nearest(data, w, h, tw, th):
    """RGBx w x h -> RGBx tw x th, nearest pixel (the last fallback: no PIL, no GStreamer encoder)."""
    import array
    if (tw, th) == (w, h):
        return bytes(data[:w * h * 4])
    xs = [min(w - 1, (2 * x + 1) * w // (2 * tw)) for x in range(tw)]
    out = bytearray()
    view = memoryview(data)
    for y in range(th):
        sy = min(h - 1, (2 * y + 1) * h // (2 * th))
        row = array.array("I")
        row.frombytes(view[sy * w * 4:(sy + 1) * w * 4])
        out += array.array("I", [row[x] for x in xs]).tobytes()
    return bytes(out)


def png_rgbx(data, w, h):
    """A PNG (8-bit RGB, filter 0 rows, zlib) of an RGBx picture, in pure Python."""
    import zlib

    def chunk(kind, body):
        return struct.pack(">I", len(body)) + kind + body + struct.pack(">I", zlib.crc32(kind + body) & 0xFFFFFFFF)
    raw = bytearray()
    for y in range(h):
        raw.append(0)
        raw += rgbx_to_rgb(data[y * w * 4:(y + 1) * w * 4])
    return (b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", struct.pack(">IIBBBBB", w, h, 8, 2, 0, 0, 0))
            + chunk(b"IDAT", zlib.compress(bytes(raw), 6)) + chunk(b"IEND", b""))


def encode_pil(data, w, h, tw, th, fmt):
    import io
    from PIL import Image
    image = Image.frombuffer("RGB", (w, h), bytes(data[:w * h * 4]), "raw", "RGBX", 0, 1)
    if (tw, th) != (w, h):
        image = image.resize((tw, th), Image.BILINEAR, reducing_gap=2.0)
    buf = io.BytesIO()
    if fmt == SNAP_JPEG:
        image.save(buf, "JPEG", quality=80)
    else:
        image.save(buf, "PNG", compress_level=6)
    return fmt, buf.getvalue()


def encode_snapshot(data, w, h, max_width, fmt, gst_encode=None, use_pil=True):
    """(format, width, height, image) of an RGBx picture: PIL, else GStreamer (`gst_encode`), else a PNG made
    here (a JPEG asked for then comes back as PNG: the format byte says which). Smaller while over the cap."""
    tw, th = snapshot_size(w, h, max_width)
    for _ in range(5):
        if tw < 16:
            break
        made = None
        if use_pil:
            try:
                made = encode_pil(data, w, h, tw, th, fmt)
            except Exception:
                use_pil = False
        if made is None and gst_encode is not None:
            try:
                made = gst_encode(data, w, h, tw, th, fmt)
            except Exception as error:
                sys.stderr.write("snapshot: gstreamer: %s\n" % error)
                gst_encode = None
        if made is None:
            made = SNAP_PNG, png_rgbx(scale_rgbx_nearest(data, w, h, tw, th), tw, th)
        if len(made[1]) <= SNAP_MAX_IMAGE:
            return made[0], tw, th, made[1]
        tw, th = snapshot_size(w, h, tw * 3 // 4)
    return None


class LocalMotion:
    """Pointer moves count only past 15 px of the owner's own travel; travel older than 0.5 s is forgotten."""

    def __init__(self, threshold=LOCAL_MOVE_PX, quiet=0.5):
        self.threshold, self.quiet = threshold, quiet
        self.last = None
        self.travelled = 0.0
        self.last_own = None

    def move(self, x, y, injected, now):
        previous, self.last = self.last, (x, y)
        if injected or previous is None:
            return False
        if self.last_own is None or now - self.last_own > self.quiet:
            self.travelled = 0.0
        self.last_own = now
        self.travelled += ((x - previous[0]) ** 2 + (y - previous[1]) ** 2) ** 0.5
        if self.travelled <= self.threshold:
            return False
        self.travelled = 0.0
        return True


class LocalThrottle:
    """{"t":"local"} at most every 250 ms."""

    def __init__(self, interval=LOCAL_INTERVAL_S):
        self.interval, self.last = interval, None

    def take(self, now):
        if self.last is not None and now - self.last < self.interval:
            return False
        self.last = now
        return True


def xtest_devices(rows):
    """The XInput2 device ids of XTest's virtual devices ("Virtual core XTEST pointer/keyboard"): our own events."""
    return {dev for dev, name in rows if b"XTEST" in (name or b"")}


def xi_mask(events):
    """An XIEventMask's bytes for these XI2 event types."""
    out = bytearray(max(events) // 8 + 1)
    for e in events:
        out[e // 8] |= 1 << (e % 8)
    return bytes(out)


def local_from_idle(idle_ms, since_post_ms, margin_ms=150, recent_ms=250):
    """XScreenSaver fallback: input happened lately (idle) and after our own last event (+ margin)."""
    return idle_ms < recent_ms and idle_ms + margin_ms < since_post_ms


# ---- output ----

class Out:
    """stdout is the record stream: one writer, whole records only."""

    def __init__(self):
        self.lock = threading.Lock()
        self.fd = sys.stdout.buffer

    def write(self, data):
        with self.lock:
            try:
                self.fd.write(data)
                self.fd.flush()
            except (BrokenPipeError, OSError):
                os._exit(0)

    def status(self, obj):
        self.write(status_record(obj))

    def error(self, code, message):
        self.status({"t": "error", "code": code, "message": str(message)[:120]})


def monotonic_us():
    return time.clock_gettime_ns(time.CLOCK_MONOTONIC) // 1000


def die_with_parent():
    """The helper never outlives Node (PR_SET_PDEATHSIG)."""
    try:
        import ctypes
        libc = ctypes.CDLL(None)
        libc.prctl(1, signal.SIGTERM)
        if os.getppid() == 1:
            os._exit(0)
    except Exception:
        pass


# ---- input: X11 XTest ----

class XTest:
    def __init__(self, display, region):
        import ctypes
        self.c = ctypes
        self.x = ctypes.CDLL("libX11.so.6")
        self.t = ctypes.CDLL("libXtst.so.6")
        self.x.XOpenDisplay.restype = ctypes.c_void_p
        self.x.XOpenDisplay.argtypes = [ctypes.c_char_p]
        for fn in ("XFlush", "XCloseDisplay"):
            getattr(self.x, fn).argtypes = [ctypes.c_void_p]
        self.t.XTestFakeMotionEvent.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int, ctypes.c_int, ctypes.c_ulong]
        self.t.XTestFakeButtonEvent.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_int, ctypes.c_ulong]
        self.t.XTestFakeKeyEvent.argtypes = [ctypes.c_void_p, ctypes.c_uint, ctypes.c_int, ctypes.c_ulong]
        self.x.XSync.argtypes = [ctypes.c_void_p, ctypes.c_int]
        self.x.XDisplayKeycodes.argtypes = [ctypes.c_void_p, ctypes.POINTER(ctypes.c_int), ctypes.POINTER(ctypes.c_int)]
        self.x.XGetKeyboardMapping.restype = ctypes.POINTER(ctypes.c_ulong)
        self.x.XGetKeyboardMapping.argtypes = [ctypes.c_void_p, ctypes.c_ubyte, ctypes.c_int, ctypes.POINTER(ctypes.c_int)]
        self.x.XChangeKeyboardMapping.argtypes = [ctypes.c_void_p, ctypes.c_int, ctypes.c_int, ctypes.POINTER(ctypes.c_ulong), ctypes.c_int]
        self.x.XKeysymToKeycode.restype = ctypes.c_ubyte
        self.x.XKeysymToKeycode.argtypes = [ctypes.c_void_p, ctypes.c_ulong]
        self.x.XFree.argtypes = [ctypes.c_void_p]
        self.dpy = self.x.XOpenDisplay(display.encode())
        if not self.dpy:
            raise RuntimeError("cannot open the X display")
        self.region = region
        self.last_post = 0.0  # monotonic time of our own last event (the agent's XScreenSaver fallback)
        self.buttons, self.keys = set(), set()
        self.ring = None      # spare keycodes for typed text, found on the first character
        self.ring_next = 0
        self.bound = {}       # keycode -> its original keysyms, while text holds it

    def handle(self, cmd):
        op = cmd[0]
        self.last_post = time.monotonic()
        if op == "p":
            x, y, w, h = self.region
            px, py = to_pixels(cmd[1], cmd[2], w, h)
            self.t.XTestFakeMotionEvent(self.dpy, -1, x + px, y + py, 0)
        elif op == "b":
            down = cmd[2] == 1
            (self.buttons.add if down else self.buttons.discard)(cmd[1])
            self.t.XTestFakeButtonEvent(self.dpy, X_BUTTONS[cmd[1]], 1 if down else 0, 0)
        elif op == "k":
            down = cmd[2] == 1
            (self.keys.add if down else self.keys.discard)(cmd[1])
            self.t.XTestFakeKeyEvent(self.dpy, cmd[1] + 8, 1 if down else 0, 0)  # evdev + 8 = X keycode
        elif op == "s":
            for steps, pos, neg in ((cmd[2], 4, 5), (cmd[1], 6, 7)):
                button = pos if steps > 0 else neg
                for _ in range(abs(steps)):
                    self.t.XTestFakeButtonEvent(self.dpy, button, 1, 0)
                    self.t.XTestFakeButtonEvent(self.dpy, button, 0, 0)
        elif op == "z" and cmd[1]:
            # Never left held: the Ctrl pressed here is let go in this same call.
            ctrl = zoom_needs_ctrl(self.keys)
            if ctrl:
                self.t.XTestFakeKeyEvent(self.dpy, CTRL_KEYS[0] + 8, 1, 0)
            button = 4 if cmd[1] > 0 else 5
            for _ in range(abs(cmd[1])):
                self.t.XTestFakeButtonEvent(self.dpy, button, 1, 0)
                self.t.XTestFakeButtonEvent(self.dpy, button, 0, 0)
            if ctrl:
                self.t.XTestFakeKeyEvent(self.dpy, CTRL_KEYS[0] + 8, 0, 0)
        elif op == "t":
            self.type_char(cmd[1])
        elif op == "x":
            self.release_all()
            return
        self.x.XFlush(self.dpy)

    def _mapping(self, first, count):
        """(keysyms, per) of `count` keycodes from `first`."""
        c = self.c
        per = c.c_int(0)
        table = self.x.XGetKeyboardMapping(self.dpy, first, count, c.byref(per))
        if not table:
            return [], 0
        try:
            return [table[i] for i in range(count * per.value)], per.value
        finally:
            self.x.XFree(table)

    def _bind(self, keycode, keysyms):
        arr = (self.c.c_ulong * len(keysyms))(*keysyms)
        self.x.XChangeKeyboardMapping(self.dpy, keycode, len(keysyms), arr, 1)
        self.x.XSync(self.dpy, 0)

    def type_char(self, cp):
        """A character whatever the layout (xdotool's way): the keycode already typing its keysym at level 0,
        else one of a ring of spare keycodes bound to it. The ring is reused in turn and only put back at the
        end, so an app still reading the previous key never sees it remapped under it. The modifiers this
        session holds are let go around it (text never becomes a shortcut) and pressed again after."""
        keysym = text_keysym(cp)
        keycode = self.x.XKeysymToKeycode(self.dpy, keysym)
        if keycode:
            syms, _per = self._mapping(keycode, 1)
            if not syms or syms[0] != keysym:
                keycode = 0
        if not keycode:
            if self.ring is None:
                lo, hi = self.c.c_int(0), self.c.c_int(0)
                self.x.XDisplayKeycodes(self.dpy, self.c.byref(lo), self.c.byref(hi))
                syms, per = self._mapping(lo.value, hi.value - lo.value + 1)
                self.ring = spare_keycodes(syms, lo.value, per)
            if not self.ring:
                return
            keycode = self.ring[self.ring_next % len(self.ring)]
            self.ring_next += 1
            if keycode not in self.bound:
                self.bound[keycode] = self._mapping(keycode, 1)[0]
            self._bind(keycode, [keysym, keysym])
        mods = [k for k in MODIFIER_KEYS if k in self.keys]
        for k in mods:
            self.t.XTestFakeKeyEvent(self.dpy, k + 8, 0, 0)
        self.t.XTestFakeKeyEvent(self.dpy, keycode, 1, 0)
        self.t.XTestFakeKeyEvent(self.dpy, keycode, 0, 0)
        for k in mods:
            self.t.XTestFakeKeyEvent(self.dpy, k + 8, 1, 0)
        self.x.XSync(self.dpy, 0)
        self.last_post = time.monotonic()

    def unbind_all(self):
        for keycode, syms in list(self.bound.items()):
            try:
                self._bind(keycode, syms or [0])
            except Exception:
                pass
        self.bound.clear()

    def close(self):
        self.release_all()
        if self.dpy:
            self.unbind_all()
            self.x.XCloseDisplay(self.dpy)
            self.dpy = None

    def release_all(self):
        if not self.dpy:
            return
        for b in list(self.buttons):
            self.t.XTestFakeButtonEvent(self.dpy, X_BUTTONS[b], 0, 0)
        for k in list(self.keys):
            self.t.XTestFakeKeyEvent(self.dpy, k + 8, 0, 0)
        self.buttons.clear()
        self.keys.clear()
        self.x.XFlush(self.dpy)


class LocalWatch:
    """Agent mode on X11 (Argon's review point 7): the owner's own input. XInput2 raw events on the root
    window, skipping those from XTest's devices (ours, and any other injector's), so the owner is never
    blinded during the agent's drags or typing; without XInput2, XScreenSaver's idle time against our own
    last event (best effort). Its own thread and X connection."""

    GENERIC_EVENT = 35
    XI_RAW_KEY_PRESS, XI_RAW_BUTTON_PRESS, XI_RAW_MOTION = 13, 15, 17

    def __init__(self, display, emit, last_post):
        self.display, self.emit, self.last_post = display, emit, last_post
        self.motion, self.throttle = LocalMotion(), LocalThrottle()
        self.running = True
        self.kind = None

    def report(self):
        if self.throttle.take(time.monotonic()):
            self.emit()

    def start(self):
        """"xi2", "idle" or None (nothing can tell here)."""
        import ctypes
        c = ctypes
        try:
            x = c.CDLL("libX11.so.6")
            x.XOpenDisplay.restype = c.c_void_p
            x.XOpenDisplay.argtypes = [c.c_char_p]
            x.XDefaultRootWindow.restype = c.c_ulong
            x.XDefaultRootWindow.argtypes = [c.c_void_p]
            dpy = x.XOpenDisplay(self.display.encode())
            if not dpy:
                return None
        except Exception as error:
            sys.stderr.write("local: %s\n" % error)
            return None
        self.c, self.x, self.dpy = c, x, dpy
        self.root = x.XDefaultRootWindow(dpy)
        for kind, setup, loop in (("xi2", self.setup_xi2, self.loop_xi2), ("idle", self.setup_idle, self.loop_idle)):
            try:
                if setup():
                    self.kind = kind
                    threading.Thread(target=self.run, args=(loop,), daemon=True).start()
                    return kind
            except Exception as error:
                sys.stderr.write("local %s: %s\n" % (kind, error))
        return None

    def run(self, loop):
        try:
            loop()
        except Exception as error:
            sys.stderr.write("local: stopped (%s)\n" % error)

    # -- XInput2 raw events --

    def setup_xi2(self):
        c, x = self.c, self.x
        xi = c.CDLL("libXi.so.6")

        class XIDeviceInfo(c.Structure):
            _fields_ = [("deviceid", c.c_int), ("name", c.c_char_p), ("use", c.c_int), ("attachment", c.c_int),
                        ("enabled", c.c_int), ("num_classes", c.c_int), ("classes", c.c_void_p)]

        class XIEventMask(c.Structure):
            _fields_ = [("deviceid", c.c_int), ("mask_len", c.c_int), ("mask", c.POINTER(c.c_ubyte))]
        x.XQueryExtension.argtypes = [c.c_void_p, c.c_char_p, c.POINTER(c.c_int), c.POINTER(c.c_int), c.POINTER(c.c_int)]
        opcode, first_event, first_error = c.c_int(), c.c_int(), c.c_int()
        if not x.XQueryExtension(self.dpy, b"XInputExtension", c.byref(opcode), c.byref(first_event), c.byref(first_error)):
            return False
        xi.XIQueryVersion.argtypes = [c.c_void_p, c.POINTER(c.c_int), c.POINTER(c.c_int)]
        major, minor = c.c_int(2), c.c_int(0)
        if xi.XIQueryVersion(self.dpy, c.byref(major), c.byref(minor)) != 0:
            return False
        xi.XIQueryDevice.restype = c.POINTER(XIDeviceInfo)
        xi.XIQueryDevice.argtypes = [c.c_void_p, c.c_int, c.POINTER(c.c_int)]
        xi.XIFreeDeviceInfo.argtypes = [c.POINTER(XIDeviceInfo)]
        count = c.c_int()
        devices = xi.XIQueryDevice(self.dpy, 0, c.byref(count))  # XIAllDevices
        if not devices:
            return False
        try:
            self.xtest_ids = xtest_devices([(devices[i].deviceid, devices[i].name) for i in range(count.value)])
        finally:
            xi.XIFreeDeviceInfo(devices)
        if not self.xtest_ids:
            return False  # cannot tell our own events apart: the idle fallback
        bits = xi_mask([self.XI_RAW_KEY_PRESS, self.XI_RAW_BUTTON_PRESS, self.XI_RAW_MOTION])
        mask_bytes = (c.c_ubyte * len(bits))(*bits)
        mask = XIEventMask(1, len(bits), c.cast(mask_bytes, c.POINTER(c.c_ubyte)))  # XIAllMasterDevices
        xi.XISelectEvents.argtypes = [c.c_void_p, c.c_ulong, c.POINTER(XIEventMask), c.c_int]
        if xi.XISelectEvents(self.dpy, self.root, c.byref(mask), 1) != 0:
            return False
        x.XFlush.argtypes = [c.c_void_p]
        x.XFlush(self.dpy)
        self.xi, self.opcode, self._mask_keep = xi, opcode.value, mask_bytes
        return True

    def pointer(self):
        c = self.c
        root, child = c.c_ulong(), c.c_ulong()
        rx, ry, wx, wy, buttons = c.c_int(), c.c_int(), c.c_int(), c.c_int(), c.c_uint()
        if not self.x.XQueryPointer(self.dpy, self.root, c.byref(root), c.byref(child), c.byref(rx), c.byref(ry), c.byref(wx), c.byref(wy), c.byref(buttons)):
            return None
        return rx.value, ry.value

    def loop_xi2(self):
        import select
        c, x = self.c, self.x

        class XGenericEventCookie(c.Structure):
            _fields_ = [("type", c.c_int), ("serial", c.c_ulong), ("send_event", c.c_int), ("display", c.c_void_p),
                        ("extension", c.c_int), ("evtype", c.c_int), ("cookie", c.c_uint), ("data", c.c_void_p)]

        class XIValuatorState(c.Structure):
            _fields_ = [("mask_len", c.c_int), ("mask", c.POINTER(c.c_ubyte)), ("values", c.POINTER(c.c_double))]

        class XIRawEvent(c.Structure):
            _fields_ = [("type", c.c_int), ("serial", c.c_ulong), ("send_event", c.c_int), ("display", c.c_void_p),
                        ("extension", c.c_int), ("evtype", c.c_int), ("time", c.c_ulong), ("deviceid", c.c_int),
                        ("sourceid", c.c_int), ("detail", c.c_int), ("flags", c.c_int), ("valuators", XIValuatorState),
                        ("raw_values", c.POINTER(c.c_double))]
        x.XPending.argtypes = [c.c_void_p]
        x.XConnectionNumber.argtypes = [c.c_void_p]
        x.XNextEvent.argtypes = [c.c_void_p, c.c_void_p]
        x.XGetEventData.argtypes = [c.c_void_p, c.POINTER(XGenericEventCookie)]
        x.XFreeEventData.argtypes = [c.c_void_p, c.POINTER(XGenericEventCookie)]
        x.XQueryPointer.argtypes = [c.c_void_p, c.c_ulong] + [c.POINTER(c.c_ulong)] * 2 + [c.POINTER(c.c_int)] * 4 + [c.POINTER(c.c_uint)]
        fd = x.XConnectionNumber(self.dpy)
        event = (c.c_long * 24)()  # XEvent: a union of 24 longs
        last_query = 0.0
        while self.running:
            if x.XPending(self.dpy) == 0:
                select.select([fd], [], [], 0.5)
                continue
            x.XNextEvent(self.dpy, event)
            cookie = c.cast(event, c.POINTER(XGenericEventCookie)).contents
            if cookie.type != self.GENERIC_EVENT or cookie.extension != self.opcode:
                continue
            if not x.XGetEventData(self.dpy, c.byref(cookie)):
                continue
            try:
                raw = c.cast(cookie.data, c.POINTER(XIRawEvent)).contents
                evtype, own = cookie.evtype, raw.sourceid not in self.xtest_ids
            finally:
                x.XFreeEventData(self.dpy, c.byref(cookie))
            now = time.monotonic()
            if evtype == self.XI_RAW_MOTION:
                # Where the pointer is (XTest's moves too, so the owner's next delta never jumps); the
                # owner's moves are sampled at most every 20 ms (a 1000 Hz mouse would flood the X server).
                if own and now - last_query < 0.02:
                    continue
                last_query = now
                at = self.pointer()
                if at and self.motion.move(at[0], at[1], not own, now):
                    self.report()
            elif own:
                self.report()

    # -- XScreenSaver idle (no XInput2) --

    def setup_idle(self):
        c = self.c
        xss = c.CDLL("libXss.so.1")

        class XScreenSaverInfo(c.Structure):
            _fields_ = [("window", c.c_ulong), ("state", c.c_int), ("kind", c.c_int), ("til_or_since", c.c_ulong),
                        ("idle", c.c_ulong), ("eventMask", c.c_ulong)]
        xss.XScreenSaverAllocInfo.restype = c.POINTER(XScreenSaverInfo)
        xss.XScreenSaverQueryInfo.argtypes = [c.c_void_p, c.c_ulong, c.POINTER(XScreenSaverInfo)]
        info = xss.XScreenSaverAllocInfo()
        if not info or not xss.XScreenSaverQueryInfo(self.dpy, self.root, info):
            return False
        self.xss, self.info = xss, info
        return True

    def loop_idle(self):
        while self.running:
            time.sleep(0.1)
            if not self.xss.XScreenSaverQueryInfo(self.dpy, self.root, self.info):
                continue
            since_post = (time.monotonic() - self.last_post()) * 1000
            if local_from_idle(self.info.contents.idle, since_post):
                self.report()

    def stop(self):
        self.running = False


def _exit_on_signal(_sig, _frame):
    raise SystemExit(0)  # runs the finally blocks: every held key and button is released


def input_x11_main(args):
    die_with_parent()
    for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
        signal.signal(sig, _exit_on_signal)
    try:
        xt = XTest(args["display"], parse_region(args["region"]))
    except Exception as error:  # no libXtst, no display
        sys.stderr.write("input-x11: %s\n" % error)
        return 3
    magnify = MagnifyNotches()
    try:
        for line in sys.stdin:
            cmd = parse_command(line)
            if not cmd:
                continue
            if cmd[0] == "Q":
                break
            cmd = zoom_command(magnify, cmd)
            if cmd and cmd[0] in "pbksxtz":
                xt.handle(cmd)
    finally:
        xt.close()
    return 0


# ---- GStreamer + portal / Mutter ----

def load_gi():
    import gi
    gi.require_version("Gst", "1.0")
    gi.require_version("GstVideo", "1.0")
    from gi.repository import GLib, Gio, Gst, GstVideo  # noqa: F401
    Gst.init(None)
    return GLib, Gio, Gst, GstVideo


def probe_main():
    out = {"gst": False, "encoders": [], "ximagesrc": False, "pipewiresrc": False, "portal": {}, "mutter": False, "xtest": False}
    try:
        GLib, Gio, Gst, _ = load_gi()
        out["gst"] = True
        out["h264parse"] = bool(Gst.ElementFactory.find("h264parse"))
        out["encoders"] = usable_encoders([e for e in ENCODERS if Gst.ElementFactory.find(e)], out["h264parse"])
        out["ximagesrc"] = bool(Gst.ElementFactory.find("ximagesrc"))
        out["pipewiresrc"] = bool(Gst.ElementFactory.find("pipewiresrc"))
        out["videotestsrc"] = bool(Gst.ElementFactory.find("videotestsrc"))
        if not (Gst.ElementFactory.find("appsink") and Gst.ElementFactory.find("videoconvert") and Gst.ElementFactory.find("videoscale")):
            out["gst"] = False  # gstreamer1.0-plugins-base
        try:
            bus = Gio.bus_get_sync(Gio.BusType.SESSION, None)

            def prop(dest, path, iface, name):
                try:
                    v = bus.call_sync(dest, path, "org.freedesktop.DBus.Properties", "Get", GLib.Variant("(ss)", (iface, name)),
                                      GLib.VariantType("(v)"), Gio.DBusCallFlags.NONE, 2000, None)
                    return v.unpack()[0]
                except Exception:
                    return None
            p = ("org.freedesktop.portal.Desktop", "/org/freedesktop/portal/desktop")
            sc = prop(*p, "org.freedesktop.portal.ScreenCast", "version")
            if sc:
                out["portal"] = {
                    "screencast": sc,
                    "sourceTypes": prop(*p, "org.freedesktop.portal.ScreenCast", "AvailableSourceTypes") or 0,
                    "cursorModes": prop(*p, "org.freedesktop.portal.ScreenCast", "AvailableCursorModes") or 0,
                    "remoteDesktop": prop(*p, "org.freedesktop.portal.RemoteDesktop", "version") or 0,
                }
            names = bus.call_sync("org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", "ListNames", None,
                                  GLib.VariantType("(as)"), Gio.DBusCallFlags.NONE, 2000, None).unpack()[0]
            out["mutter"] = "org.gnome.Mutter.ScreenCast" in names and "org.gnome.Mutter.RemoteDesktop" in names
        except Exception:
            pass
    except Exception as error:
        out["error"] = str(error)[:200]
    try:
        import ctypes.util
        out["xtest"] = bool(ctypes.util.find_library("Xtst"))
    except Exception:
        pass
    sys.stdout.write(json.dumps(out) + "\n")
    return 0


class Portal:
    """xdg-desktop-portal ScreenCast (+ RemoteDesktop): Request/Response, one session."""

    DEST, PATH = "org.freedesktop.portal.Desktop", "/org/freedesktop/portal/desktop"

    def __init__(self, GLib, Gio, bus, loop):
        self.GLib, self.Gio, self.bus, self.loop = GLib, Gio, bus, loop
        self.session = None
        self.counter = 0
        self.remote = False
        # Input calls in order, one at a time: xdg-desktop-portal handles each method call on its
        # own thread, so back-to-back Notify* calls reach the compositor reordered (GNOME typed
        # "hI" for Shift+h, i). The next is sent when the previous one has answered.
        self.queue = collections.deque()
        self.inflight = False
        self.flushing = False

    def _token(self):
        self.counter += 1
        return "dokki%d_%d" % (os.getpid(), self.counter)

    def call(self, iface, method, args_fmt, args, timeout=120):
        """A portal method that answers through a Request's Response signal; blocks (a nested loop)."""
        GLib, Gio = self.GLib, self.Gio
        token = self._token()
        path = request_path(self.bus.get_unique_name(), token)
        result = {}
        ctx = GLib.MainLoop()

        def on_response(_c, _s, _p, _i, _sig, params):
            code, results = params.unpack()
            result["code"], result["results"] = code, results
            ctx.quit()
        sub = self.bus.signal_subscribe(self.DEST, "org.freedesktop.portal.Request", "Response", path, None,
                                        Gio.DBusSignalFlags.NO_MATCH_RULE, on_response)
        try:
            options = args[-1]
            options["handle_token"] = GLib.Variant("s", token)
            self.bus.call_sync(self.DEST, self.PATH, iface, method, GLib.Variant(args_fmt, tuple(args)), None,
                               Gio.DBusCallFlags.NONE, -1, None)
            fired = {}

            def give_up():
                fired["yes"] = True
                ctx.quit()
                return False
            timer = GLib.timeout_add_seconds(timeout, give_up)
            ctx.run()
            if not fired:
                GLib.source_remove(timer)
        finally:
            self.bus.signal_unsubscribe(sub)
        return result.get("code", 2), result.get("results", {})

    def start(self, want_input, source_types, cursor_modes, rd_version, token_file):
        GLib = self.GLib
        V = GLib.Variant
        restore = None
        if token_file and source_types & SOURCE_VIRTUAL:
            token_file += "-virtual"  # a token is bound to what was shared: one per kind
        if token_file:
            try:
                with open(token_file) as f:
                    restore = f.read().strip()[:256] or None
            except OSError:
                restore = None
        self.remote = bool(want_input and rd_version)
        iface = "org.freedesktop.portal.RemoteDesktop" if self.remote else "org.freedesktop.portal.ScreenCast"
        code, res = self.call(iface, "CreateSession", "(a{sv})", [{"session_handle_token": V("s", self._token())}])
        if code != 0:
            raise portal_failure(code, "screen sharing was not allowed on this computer", "create a session")
        self.session = res["session_handle"]
        persist = {}
        if restore:
            persist["restore_token"] = V("s", restore)
        persist["persist_mode"] = V("u", 2)
        if self.remote:
            opts = {"types": V("u", 3)}  # keyboard | pointer
            if rd_version >= 2:
                opts.update(persist)
            code, _ = self.call(iface, "SelectDevices", "(oa{sv})", [self.session, opts])
            if code != 0:
                raise portal_failure(code, "input was not allowed", "choose the input devices")
        sopts = {"types": V("u", source_types), "multiple": V("b", False)}
        if cursor_modes & CURSOR_EMBEDDED:
            sopts["cursor_mode"] = V("u", CURSOR_EMBEDDED)
        if not self.remote:
            sopts.update(persist)
        code, _ = self.call("org.freedesktop.portal.ScreenCast", "SelectSources", "(oa{sv})", [self.session, sopts])
        if code != 0:
            raise portal_failure(code, "no screen was chosen", "choose a screen")
        code, res = self.call(iface, "Start", "(osa{sv})", [self.session, "", {}])
        if code != 0:
            raise portal_failure(code, "screen sharing was not allowed on this computer", "start the screen stream")
        streams = res.get("streams") or []
        if not streams:
            raise PortalError("portal", "the portal gave no stream")
        if token_file and res.get("restore_token"):
            try:
                os.makedirs(os.path.dirname(token_file), mode=0o700, exist_ok=True)
                fd = os.open(token_file, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
                with os.fdopen(fd, "w") as f:
                    f.write(res["restore_token"])
            except OSError:
                pass
        if self.remote and not (int(res.get("devices") or 0) & 1 and int(res.get("devices") or 0) & 2):
            self.remote = False  # the owner shared the screen but not the keyboard/pointer
        node, props = streams[0]
        size = props.get("size") or (0, 0)
        fd_list = self.Gio.UnixFDList.new()
        ret, out_fds = self.bus.call_with_unix_fd_list_sync(self.DEST, self.PATH, "org.freedesktop.portal.ScreenCast", "OpenPipeWireRemote",
                                                            V("(oa{sv})", (self.session, {})), GLib.VariantType("(h)"),
                                                            self.Gio.DBusCallFlags.NONE, -1, fd_list, None)
        fd = out_fds.get(ret.unpack()[0])
        source_type = int(props.get("source_type") or SOURCE_MONITOR)
        return node, fd, size, source_type

    def input(self, cmd, node, area, held):
        """`area`: (x, y, w, h) of the stream the phone shows, in stream pixels (all of it, or the wall's crop)."""
        if not self.remote or not self.session or self.flushing:
            return
        rd = "org.freedesktop.portal.RemoteDesktop"
        op = cmd[0]
        try:
            if op == "p":
                ox, oy, w, h = area
                x, y = to_pixels(cmd[1], cmd[2], max(1, w), max(1, h))
                x, y = x + ox, y + oy
                self._notify(rd, "NotifyPointerMotionAbsolute", "(oa{sv}udd)", (self.session, {}, node, float(x), float(y)))
            elif op == "b":
                (held["buttons"].add if cmd[2] else held["buttons"].discard)(cmd[1])
                self._notify(rd, "NotifyPointerButton", "(oa{sv}iu)", (self.session, {}, EVDEV_BUTTONS[cmd[1]], cmd[2]))
            elif op == "k":
                (held["keys"].add if cmd[2] else held["keys"].discard)(cmd[1])
                self._notify(rd, "NotifyKeyboardKeycode", "(oa{sv}iu)", (self.session, {}, cmd[1], cmd[2]))
            elif op == "t":
                # Typed text by keysym (the compositor finds or makes a key for it), bare: held modifiers up around it.
                mods = [k for k in MODIFIER_KEYS if k in held["keys"]]
                for k in mods:
                    self._notify(rd, "NotifyKeyboardKeycode", "(oa{sv}iu)", (self.session, {}, k, 0))
                for state in (1, 0):
                    self._notify(rd, "NotifyKeyboardKeysym", "(oa{sv}iu)", (self.session, {}, text_keysym(cmd[1]), state))
                for k in mods:
                    self._notify(rd, "NotifyKeyboardKeycode", "(oa{sv}iu)", (self.session, {}, k, 1))
            elif op == "s":
                if cmd[2]:
                    self._notify(rd, "NotifyPointerAxisDiscrete", "(oa{sv}ui)", (self.session, {}, 0, -cmd[2]))  # + is down here
                if cmd[1]:
                    self._notify(rd, "NotifyPointerAxisDiscrete", "(oa{sv}ui)", (self.session, {}, 1, -cmd[1]))
            elif op == "z" and cmd[1]:
                # Ctrl + wheel, the Ctrl let go in this same call (flush() never drops a key up).
                ctrl = zoom_needs_ctrl(held["keys"])
                if ctrl:
                    self._notify(rd, "NotifyKeyboardKeycode", "(oa{sv}iu)", (self.session, {}, CTRL_KEYS[0], 1))
                self._notify(rd, "NotifyPointerAxisDiscrete", "(oa{sv}ui)", (self.session, {}, 0, -cmd[1]))
                if ctrl:
                    self._notify(rd, "NotifyKeyboardKeycode", "(oa{sv}iu)", (self.session, {}, CTRL_KEYS[0], 0))
        except Exception as error:
            sys.stderr.write("portal input: %s\n" % error)

    def release(self, held):
        for b in list(held["buttons"]):
            self.input(("b", b, 0), 0, (1, 1), held)
        for k in list(held["keys"]):
            self.input(("k", k, 0), 0, (1, 1), held)

    def prime(self):
        """See PRIME_KEY: the compositor's virtual pointer and keyboard take their first event now."""
        if not self.remote or not self.session:
            return
        rd = "org.freedesktop.portal.RemoteDesktop"
        try:
            self._notify(rd, "NotifyPointerMotion", "(oa{sv}dd)", (self.session, {}, 0.0, 0.0))
            for state in (1, 0):
                self._notify(rd, "NotifyKeyboardKeycode", "(oa{sv}iu)", (self.session, {}, PRIME_KEY, state))
        except Exception as error:
            sys.stderr.write("portal prime: %s\n" % error)

    QUEUE_MAX = 256

    def _notify(self, iface, method, fmt, args):
        item = (iface, method, fmt, args)
        last = self.queue[-1] if self.queue else None
        if method == "NotifyPointerMotionAbsolute" and self._unmoved(args):
            return  # the same place as a move still waiting, with only wheel steps since (a scroll)
        if method == "NotifyPointerMotionAbsolute" and last and last[1] == method:
            self.queue[-1] = item  # back-to-back absolute moves: only the newest still matters
        elif method == "NotifyPointerAxisDiscrete" and last and last[1] == method and last[3][2] == args[2]:
            # Back-to-back wheel steps on one axis add up (bounded like the phone's own records).
            steps = max(-50, min(50, last[3][3] + args[3]))
            self.queue[-1] = (iface, method, fmt, (*args[:3], steps))
        elif method in ("NotifyPointerAxisDiscrete", "NotifyPointerMotion") and len(self.queue) >= self.QUEUE_MAX:
            pass  # a wheel step or a relative nudge can go; a move before a click, a button or a key never
        else:
            self.queue.append(item)
        self._pump()

    def _unmoved(self, args):
        """A queued (unsent) absolute move to this place, followed by nothing but wheel steps."""
        for _iface, method, _fmt, queued in reversed(self.queue):
            if method == "NotifyPointerMotionAbsolute":
                return queued[2:] == args[2:]
            if method != "NotifyPointerAxisDiscrete":
                return False
        return False

    def _pump(self):
        if self.inflight or not self.queue:
            return
        iface, method, fmt, args = self.queue.popleft()
        self.inflight = True

        def done(bus, result, _data=None):
            try:
                bus.call_finish(result)
            except Exception as error:
                sys.stderr.write("portal input %s: %s\n" % (method, error))
            self.inflight = False
            self._pump()
        self.bus.call(self.DEST, self.PATH, iface, method, self.GLib.Variant(fmt, args), None, self.Gio.DBusCallFlags.NONE, 1000, None, done, None)

    def flush(self):
        """At the end (the main loop may not run again): whatever is queued, in order, now.

        The call in flight is let finish first (up to 0.5 s, running the main context), so a
        release sent now cannot overtake the press before it. Input arriving meanwhile is dropped.
        """
        self.flushing = True
        context = self.GLib.MainContext.default()
        end = time.monotonic() + 0.5
        while (self.inflight or self.queue) and time.monotonic() < end:
            if not context.iteration(False):
                time.sleep(0.005)
        while self.queue:
            iface, method, fmt, args = self.queue.popleft()
            if method not in ("NotifyPointerButton", "NotifyKeyboardKeycode") and not (method == "NotifyKeyboardKeysym" and args[3] == 0):
                continue  # out of time: only what would otherwise stay held goes out
            try:
                self.bus.call_sync(self.DEST, self.PATH, iface, method, self.GLib.Variant(fmt, args), None, self.Gio.DBusCallFlags.NONE, 500, None)
            except Exception as error:
                sys.stderr.write("portal input %s: %s\n" % (method, error))

    def close(self):
        if self.session:
            self.flush()  # the releases stop() just queued go out before the session closes
            try:
                self.bus.call_sync(self.DEST, self.session, "org.freedesktop.portal.Session", "Close", None, None, self.Gio.DBusCallFlags.NONE, 1000, None)
            except Exception:
                pass
            self.session = None


class Mutter:
    """GNOME: a virtual monitor (RecordVirtual) with its own RemoteDesktop session for input."""

    def __init__(self, GLib, Gio, bus):
        self.GLib, self.Gio, self.bus = GLib, Gio, bus
        self.rd = self.sc = self.stream = None

    def _call(self, dest, path, iface, method, args=None, reply=None):
        return self.bus.call_sync(dest, path, iface, method, args, self.GLib.VariantType(reply) if reply else None,
                                  self.Gio.DBusCallFlags.NONE, 5000, None)

    def start(self, cursor):
        V = self.GLib.Variant
        self.rd = self._call("org.gnome.Mutter.RemoteDesktop", "/org/gnome/Mutter/RemoteDesktop", "org.gnome.Mutter.RemoteDesktop", "CreateSession", None, "(o)").unpack()[0]
        session_id = self._call("org.gnome.Mutter.RemoteDesktop", self.rd, "org.freedesktop.DBus.Properties", "Get",
                                V("(ss)", ("org.gnome.Mutter.RemoteDesktop.Session", "SessionId")), "(v)").unpack()[0]
        self.sc = self._call("org.gnome.Mutter.ScreenCast", "/org/gnome/Mutter/ScreenCast", "org.gnome.Mutter.ScreenCast", "CreateSession",
                             V("(a{sv})", ({"remote-desktop-session-id": V("s", session_id)},)), "(o)").unpack()[0]
        props = {"cursor-mode": V("u", 1 if cursor else 0), "is-platform": V("b", True)}
        self.stream = self._call("org.gnome.Mutter.ScreenCast", self.sc, "org.gnome.Mutter.ScreenCast.Session", "RecordVirtual",
                                 V("(a{sv})", (props,)), "(o)").unpack()[0]
        node = {}
        ctx = self.GLib.MainLoop()

        def added(_c, _s, _p, _i, _sig, params):
            node["id"] = params.unpack()[0]
            ctx.quit()
        sub = self.bus.signal_subscribe("org.gnome.Mutter.ScreenCast", "org.gnome.Mutter.ScreenCast.Stream", "PipeWireStreamAdded",
                                        self.stream, None, self.Gio.DBusSignalFlags.NONE, added)
        try:
            self._call("org.gnome.Mutter.RemoteDesktop", self.rd, "org.gnome.Mutter.RemoteDesktop.Session", "Start")
            fired = {}

            def give_up():
                fired["yes"] = True
                ctx.quit()
                return False
            timer = self.GLib.timeout_add_seconds(10, give_up)
            ctx.run()
            if not fired:
                self.GLib.source_remove(timer)
        finally:
            self.bus.signal_unsubscribe(sub)
        if "id" not in node:
            raise PortalError("unavailable", "GNOME did not create the virtual monitor")
        return node["id"]

    def input(self, cmd, size, held):
        if not self.rd:
            return
        V = self.GLib.Variant
        iface = "org.gnome.Mutter.RemoteDesktop.Session"
        op = cmd[0]

        def call(method, fmt, args):
            self.bus.call("org.gnome.Mutter.RemoteDesktop", self.rd, iface, method, V(fmt, args), None, self.Gio.DBusCallFlags.NONE, 1000, None, None, None)
        try:
            if op == "p":
                x, y = to_pixels(cmd[1], cmd[2], max(1, size[0]), max(1, size[1]))
                call("NotifyPointerMotionAbsolute", "(sdd)", (self.stream, float(x), float(y)))
            elif op == "b":
                (held["buttons"].add if cmd[2] else held["buttons"].discard)(cmd[1])
                call("NotifyPointerButton", "(ib)", (EVDEV_BUTTONS[cmd[1]], bool(cmd[2])))
            elif op == "k":
                (held["keys"].add if cmd[2] else held["keys"].discard)(cmd[1])
                call("NotifyKeyboardKeycode", "(ub)", (cmd[1], bool(cmd[2])))
            elif op == "t":
                mods = [k for k in MODIFIER_KEYS if k in held["keys"]]
                for k in mods:
                    call("NotifyKeyboardKeycode", "(ub)", (k, False))
                for state in (True, False):
                    call("NotifyKeyboardKeysym", "(ub)", (text_keysym(cmd[1]), state))
                for k in mods:
                    call("NotifyKeyboardKeycode", "(ub)", (k, True))
            elif op == "s":
                if cmd[2]:
                    call("NotifyPointerAxisDiscrete", "(ui)", (0, -cmd[2]))
                if cmd[1]:
                    call("NotifyPointerAxisDiscrete", "(ui)", (1, -cmd[1]))
            elif op == "z" and cmd[1]:
                ctrl = zoom_needs_ctrl(held["keys"])
                if ctrl:
                    call("NotifyKeyboardKeycode", "(ub)", (CTRL_KEYS[0], True))
                call("NotifyPointerAxisDiscrete", "(ui)", (0, -cmd[1]))
                if ctrl:
                    call("NotifyKeyboardKeycode", "(ub)", (CTRL_KEYS[0], False))
        except Exception as error:
            sys.stderr.write("mutter input: %s\n" % error)

    def release(self, held, size):
        for b in list(held["buttons"]):
            self.input(("b", b, 0), size, held)
        for k in list(held["keys"]):
            self.input(("k", k, 0), size, held)

    def prime(self):
        """See PRIME_KEY."""
        if not self.rd:
            return
        V = self.GLib.Variant
        iface = "org.gnome.Mutter.RemoteDesktop.Session"
        try:
            self.bus.call("org.gnome.Mutter.RemoteDesktop", self.rd, iface, "NotifyPointerMotionRelative", V("(dd)", (0.0, 0.0)), None, self.Gio.DBusCallFlags.NONE, 1000, None, None, None)
            for state in (True, False):
                self.bus.call("org.gnome.Mutter.RemoteDesktop", self.rd, iface, "NotifyKeyboardKeycode", V("(ub)", (PRIME_KEY, state)), None, self.Gio.DBusCallFlags.NONE, 1000, None, None, None)
        except Exception as error:
            sys.stderr.write("mutter prime: %s\n" % error)

    def close(self):
        if self.rd:
            try:
                self._call("org.gnome.Mutter.RemoteDesktop", self.rd, "org.gnome.Mutter.RemoteDesktop.Session", "Stop")
            except Exception:
                pass
            self.rd = None


class PortalError(Exception):
    def __init__(self, code, message):
        super().__init__(message)
        self.code = code


def portal_failure(response, denied, step):
    """A portal Request's non-zero response: 1 is the owner saying no, anything else the portal failing.

    Only a "no" is a permission refusal; a backend that fails (2: xdg-desktop-portal-wlr without
    GPU buffers, a compositor that refuses) must not read as the owner's refusal on the phone.
    """
    if response == 1:
        return PortalError("permission", denied)
    return PortalError("portal", "the screen-sharing portal could not %s (response %s): see the xdg-desktop-portal backend's log" % (step, response))


class Streamer:
    def __init__(self, args, out):
        self.args, self.out = args, out
        self.GLib, self.Gio, self.Gst, self.GstVideo = load_gi()
        self.loop = self.GLib.MainLoop()
        self.pipeline = None
        self.enc = self.capsfilter = None
        self.code = 0
        self.encoders = []
        self.encoder_index = 0
        self.produced = False
        self.announced = False
        self.sink = None
        self.last_sample = 0.0
        self.restarts = 0
        self.watchdog = None
        self.size = None
        self.src_size = None
        self.config = None
        self.portal = self.mutter = None
        self.node = self.fd = None
        self.size_known = True
        self.input_size = (1, 1)
        self.held = {"buttons": set(), "keys": set()}
        self.magnify = MagnifyNotches()
        self.xtest = None
        self.mode, self.virtual, self.mode_note = args["mode"], False, None
        self.bitrate = int(args["bitrate"])
        self.fps = int(args["fps"])
        self.box = parse_wh(args["box"])
        self.max_long = int(args.get("max-long-side") or 2752)
        self.crop = parse_crop(args["crop"]) if args.get("crop") else None
        self.part = None  # the crop in source pixels, once the source's size is known
        self.closing = False
        # Agentes en la pantalla: no encoder; the latest picture for snapshots, local input reports.
        self.agent = args.get("agent") == "1"
        self.frame = None  # (rgbx bytes, width, height), at most every 100 ms
        self.frame_at = 0.0
        self.frame_lock = threading.Condition()
        self.snapshots = None
        self.local = None

    # -- source --

    def open_source(self):
        a = self.args
        source = a["source"]
        cursor = a.get("cursor") == "1"
        want_input = a.get("input") == "1"
        if source == "x11":
            region = parse_region(a["region"])
            self.src_size = (region[2], region[3])
            self.size = parse_wh(a["size"]) if a.get("size") else fit_size(region[2], region[3], *self.box, self.max_long)
            self.src = ("x11", dict(display=a["display"], region=region))
            if want_input:
                try:
                    self.xtest = XTest(a["display"], region)
                except Exception as error:
                    sys.stderr.write("xtest: %s\n" % error)
            return
        if source == "synthetic":
            self.set_source_size(1920, 1080)
            self.size = parse_wh(a["size"]) if a.get("size") else fit_size(*self.part[2:], *self.box, self.max_long)
            self.src = ("synthetic", {})
            return
        # Wayland
        bus = self.Gio.bus_get_sync(self.Gio.BusType.SESSION, None)
        info = self.probe_portal(bus)
        types = int(info.get("sourceTypes") or 0)
        if self.mode == "extend":
            if types & SOURCE_VIRTUAL:
                self.start_portal(bus, info, SOURCE_VIRTUAL, want_input)
                self.virtual = True
                if info.get("mutter"):
                    # GNOME's portal makes the virtual monitor with Mutter's RecordVirtual, which
                    # takes the size the consumer's caps ask for: the phone's, not Mutter's default
                    # 1280x720 (found on GNOME 48, where the portal offers VIRTUAL).
                    size = parse_wh(a["size"]) if a.get("size") else fit_size(*self.box, *self.box, self.max_long)
                    self.size = self.src_size = self.input_size = size
                    self.size_known = True
                    self.src = ("pipewire", dict(self.src[1], size=size))
                return
            if info.get("mutter"):
                self.mutter = Mutter(self.GLib, self.Gio, bus)
                size = parse_wh(a["size"]) if a.get("size") else fit_size(*self.box, *self.box, self.max_long)
                try:
                    self.node = self.mutter.start(cursor)
                    self.size = self.src_size = self.input_size = size
                    self.src = ("pipewire", dict(node=self.node, size=size))
                    self.virtual = True
                    return
                except Exception as error:  # remote desktop off, an older Mutter: mirror through the portal
                    sys.stderr.write("mutter: %s\n" % error)
                    self.mutter.close()
                    self.mutter = None
            self.mode, self.mode_note = "mirror", "extend-unavailable"
        self.start_portal(bus, info, SOURCE_MONITOR, want_input)

    def probe_portal(self, bus):
        GLib, Gio = self.GLib, self.Gio

        def prop(iface, name):
            try:
                v = bus.call_sync(Portal.DEST, Portal.PATH, "org.freedesktop.DBus.Properties", "Get", GLib.Variant("(ss)", (iface, name)),
                                  GLib.VariantType("(v)"), Gio.DBusCallFlags.NONE, 2000, None)
                return v.unpack()[0]
            except Exception:
                return 0
        names = []
        try:
            names = bus.call_sync("org.freedesktop.DBus", "/org/freedesktop/DBus", "org.freedesktop.DBus", "ListNames", None,
                                  GLib.VariantType("(as)"), Gio.DBusCallFlags.NONE, 2000, None).unpack()[0]
        except Exception:
            pass
        return {
            "sourceTypes": prop("org.freedesktop.portal.ScreenCast", "AvailableSourceTypes"),
            "cursorModes": prop("org.freedesktop.portal.ScreenCast", "AvailableCursorModes"),
            "remoteDesktop": prop("org.freedesktop.portal.RemoteDesktop", "version"),
            "mutter": "org.gnome.Mutter.ScreenCast" in names and "org.gnome.Mutter.RemoteDesktop" in names,
        }

    def start_portal(self, bus, info, source_type, want_input):
        self.portal = Portal(self.GLib, self.Gio, bus, self.loop)
        self.out.status({"t": "waiting"})  # the compositor may show its dialog now
        node, fd, size, _ = self.portal.start(want_input, source_type, int(info.get("cursorModes") or 0), int(info.get("remoteDesktop") or 0), self.args.get("token-file"))
        self.node, self.fd = node, fd
        # "size" is optional in the portal spec (xdg-desktop-portal-wlr leaves it out): then the
        # stream's own caps tell it, see watch_source_size.
        self.size_known = bool(size and size[0] and size[1])
        w, h = size if self.size_known else (1920, 1080)
        self.set_source_size(w, h)
        # A portal's virtual monitor (KDE) has the size the compositor gave it: fitted like a monitor.
        self.size = fit_size(*self.part[2:], *self.box, self.max_long)
        self.src = ("pipewire", dict(node=node, fd=fd))

    def set_source_size(self, w, h):
        """The source's size (portal, test pattern): the wall's crop in its pixels, and where input lands."""
        self.src_size = self.input_size = (w, h)
        self.part = crop_pixels(w, h, self.crop)

    def input_area(self):
        if self.part:
            return self.part
        return (0, 0, *self.input_size)

    def apply_crop(self):
        """videocrop takes what is cut from each side; set before the source's caps reach it."""
        element = self.pipeline.get_by_name("crop") if (self.pipeline is not None and self.crop) else None
        if element is None or not self.part:
            return
        x, y, w, h = self.part
        sw, sh = self.src_size
        for name, value in (("left", x), ("top", y), ("right", sw - x - w), ("bottom", sh - y - h)):
            element.set_property(name, int(value))

    def source_text(self):
        kind, opts = self.src
        cursor = self.args.get("cursor") == "1"
        if kind == "pipewire" and opts.get("fd") is not None:
            opts = dict(opts, fd=os.dup(opts["fd"]))  # pipewiresrc closes the fd it is given
        return source_description(kind, self.fps, cursor, **opts)

    # -- pipeline --

    def build(self):
        Gst = self.Gst
        name = self.encoders[self.encoder_index]
        if self.agent:
            text = agent_pipeline_description(self.source_text(), self.size, crop=bool(self.crop))
        else:
            text = pipeline_description(self.source_text(), name, self.fps, self.bitrate, self.size, parse=bool(Gst.ElementFactory.find("h264parse")), crop=bool(self.crop))
        sys.stderr.write("pipeline: %s\n" % text)
        try:
            self.pipeline = Gst.parse_launch(text)
        except Exception as error:
            self.pipeline = None
            self.GLib.idle_add(self.next_encoder, str(error)[:80], None)
            return
        clock = Gst.SystemClock.obtain()
        clock.set_property("clock-type", Gst.ClockType.MONOTONIC)
        self.pipeline.use_clock(clock)
        self.enc = self.pipeline.get_by_name("enc")
        self.capsfilter = self.pipeline.get_by_name("size")
        # A portal stream of unknown size: videocrop stays open (0 on every side) until its first caps,
        # or a crop of the 1920x1080 stand-in could refuse a smaller stream while it negotiates.
        if getattr(self, "size_known", True):
            self.apply_crop()
        pipeline = self.pipeline
        sink = self.sink = self.pipeline.get_by_name("sink")
        sink.connect("new-sample", self.on_sample)
        bus = self.pipeline.get_bus()
        bus.add_signal_watch()
        # Each handler knows its pipeline: a message from one check_stall replaced is ignored.
        bus.connect("message::error", self.on_error, pipeline)
        bus.connect("message::eos", lambda *_: pipeline is self.pipeline and self.stop(0))
        self.config = None
        self.produced = False

        def play():
            # pipewiresrc blocks its state change until PipeWire has negotiated a format: never on
            # the main loop, which must stay free for stdin commands and SIGTERM.
            if pipeline.set_state(Gst.State.PLAYING) == Gst.StateChangeReturn.FAILURE and pipeline is self.pipeline:
                self.GLib.idle_add(self.failed, "could not start %s" % name, pipeline)
        if not getattr(self, "size_known", True):
            self.watch_source_size()
        threading.Thread(target=play, daemon=True).start()
        self.GLib.timeout_add_seconds(START_TIMEOUT_S, self.check_started, pipeline)
        if self.watchdog is None and self.src and self.src[0] == "pipewire":
            self.watchdog = self.GLib.timeout_add_seconds(1, self.check_stall)

    def check_stall(self):
        """See STALL_S. Runs every second on the main loop while the helper lives."""
        if self.closing:
            self.watchdog = None
            return False
        # Only once the session has streamed (before that, check_started and the encoder fallback decide).
        if not self.announced or self.pipeline is None or time.monotonic() - self.last_sample < STALL_S:
            return True
        if self.restarts >= STALL_RESTARTS:
            return True  # (still nothing after a few new streams: leave it, the hub's liveness decides)
        self.restarts += 1
        sys.stderr.write("stream stalled (no picture in %d s): making it again (%d)\n" % (STALL_S, self.restarts))
        old, self.pipeline, self.sink = self.pipeline, None, None  # before teardown: its last samples are dropped
        self.teardown(old)
        self.last_sample = time.monotonic()
        self.build()
        return True

    def teardown(self, pipeline):
        try:
            pipeline.get_bus().remove_signal_watch()
        except Exception:
            pass
        pipeline.set_state(self.Gst.State.NULL)

    def failed(self, why, pipeline):
        """A pipeline failed: before the session streamed, the next encoder; after (a stream
        check_stall made again), the session ends with the reason."""
        if pipeline is not self.pipeline or self.closing:
            return False
        if not self.announced:
            return self.next_encoder(why, pipeline)
        self.out.error("capture", "the capture stopped: %s" % why)
        self.stop(1)
        return False

    def check_started(self, pipeline):
        if pipeline is self.pipeline and not self.produced and not self.announced and not self.closing:
            self.out.error("capture", "the screen stream did not start (no frame in %d s)" % START_TIMEOUT_S)
            self.stop(1)
        return False

    def watch_source_size(self):
        """A portal stream without a size: fit and map input from its first caps."""
        src = self.pipeline.get_by_name("pwsrc")
        if src is None:
            return

        def on_caps(_pad, info):
            event = info.get_event()
            if event.type == self.Gst.EventType.CAPS:
                st = event.parse_caps().get_structure(0)
                ok_w, w = st.get_int("width")
                ok_h, h = st.get_int("height")
                if ok_w and ok_h and w > 0 and h > 0 and (not self.size_known or (w, h) != self.src_size):
                    self.size_known = True
                    self.set_source_size(w, h)
                    # Now, before this CAPS event reaches videocrop and the capsfilter: the first frame is cut and sized right.
                    self.apply_crop()
                    self.apply_size(fit_size(*self.part[2:], *self.box, self.max_long))
            return self.Gst.PadProbeReturn.OK
        src.get_static_pad("src").add_probe(self.Gst.PadProbeType.EVENT_DOWNSTREAM, on_caps)

    def next_encoder(self, why, pipeline=None):
        # One failure can be reported twice (the state change and the bus): only the first counts.
        if pipeline is not None and pipeline is not self.pipeline:
            return False
        if self.closing:
            return False
        if self.pipeline:
            old, self.pipeline, self.sink = self.pipeline, None, None
            self.teardown(old)
        if self.agent:
            self.out.error("capture", "the screen could not be captured (%s)" % why)
            self.stop(1)
        elif self.encoder_index + 1 < len(self.encoders):
            sys.stderr.write("encoder %s failed (%s)\n" % (self.encoders[self.encoder_index], why))
            self.encoder_index += 1
            self.build()
        else:
            self.out.error("encoder", "no H.264 encoder worked here (%s)" % why)
            self.stop(1)
        return False

    def on_error(self, _bus, message, pipeline=None):
        err, _dbg = message.parse_error()
        if pipeline is not self.pipeline:
            return  # (from a pipeline already replaced)
        self.GLib.idle_add(self.failed, str(err)[:80], pipeline)

    def on_sample(self, sink):
        if self.agent:
            return self.on_agent_sample(sink)
        Gst = self.Gst
        pipeline = self.pipeline
        sample = sink.emit("pull-sample")
        if sample is None or pipeline is None or sink is not self.sink:
            return Gst.FlowReturn.OK  # (a picture from a stream check_stall just replaced)
        caps = sample.get_caps()
        st = caps.get_structure(0)
        ok, codec_data = st.get_value("codec_data") is not None, st.get_value("codec_data")
        width, height = st.get_value("width") or self.size[0], st.get_value("height") or self.size[1]
        buf = sample.get_buffer()
        ok_map, info = buf.map(Gst.MapFlags.READ)
        if not ok_map:
            return Gst.FlowReturn.OK
        try:
            payload = bytes(info.data)
        finally:
            buf.unmap(info)
        now = monotonic_us()
        capture = now
        if buf.pts != Gst.CLOCK_TIME_NONE:
            base = pipeline.get_base_time()
            capture = min(now, (base + buf.pts) // 1000)
        key = not buf.has_flags(Gst.BufferFlags.DELTA_UNIT)
        if sink is not self.sink:
            return Gst.FlowReturn.OK  # (replaced while this picture was being read)
        if ok:
            data = codec_data.extract_dup(0, codec_data.get_size())
            if data != self.config:
                self.config = data
                self.out.write(config_record(width, height, data))
        self.last_sample = time.monotonic()
        if self.restarts and self.produced:
            self.restarts = 0
        self.produced = True
        if not self.announced:
            self.announced = True
            self.out.status({"t": "ready", "width": width, "height": height, "mode": self.mode, "virtual": self.virtual,
                             "encoder": self.encoders[self.encoder_index], "input": self.input_ok(), "gestures": gestures_caps(self.input_ok()),
                             **({"modeNote": self.mode_note} if self.mode_note else {})})
        self.out.write(frame_record(capture, now - capture, key, payload))
        return Gst.FlowReturn.OK

    # -- agent --

    def on_agent_sample(self, sink):
        """Keeps the latest picture (a copy at most every 100 ms: a held PipeWire buffer could starve the stream)."""
        Gst = self.Gst
        sample = sink.emit("pull-sample")
        if sample is None or self.pipeline is None or sink is not self.sink:
            return Gst.FlowReturn.OK
        now = time.monotonic()
        self.last_sample = now
        if self.frame is not None and now - self.frame_at < 0.1:
            return Gst.FlowReturn.OK
        st = sample.get_caps().get_structure(0)
        width, height = st.get_value("width") or self.size[0], st.get_value("height") or self.size[1]
        buf = sample.get_buffer()
        ok_map, info = buf.map(Gst.MapFlags.READ)
        if not ok_map:
            return Gst.FlowReturn.OK
        try:
            data = bytes(info.data)
        finally:
            buf.unmap(info)
        if len(data) < width * height * 4:
            return Gst.FlowReturn.OK
        with self.frame_lock:
            self.frame, self.frame_at = (data, width, height), now
            self.frame_lock.notify_all()
        if self.restarts and self.produced:
            self.restarts = 0
        self.produced = True
        if not self.announced:
            self.announced = True
            self.out.status({"t": "ready", "agent": True, "width": width, "height": height, "mode": self.mode, "virtual": self.virtual,
                             "input": self.input_ok(), "gestures": gestures_caps(self.input_ok()),
                             "snapshot": True, "local": self.local is not None, "hit": False})
        return Gst.FlowReturn.OK

    def start_agent(self):
        import queue
        self.snapshots = queue.Queue(maxsize=8)
        threading.Thread(target=self.snapshot_loop, daemon=True).start()
        if self.src and self.src[0] == "x11":
            watch = LocalWatch(self.src[1]["display"], lambda: self.out.status({"t": "local"}),
                               lambda: self.xtest.last_post if self.xtest else 0.0)
            if watch.start():
                self.local = watch
                sys.stderr.write("local input: %s\n" % watch.kind)

    def snapshot(self, snap_id, max_width, fmt):
        if self.snapshots is not None:
            try:
                self.snapshots.put_nowait((snap_id, max_width, fmt, time.monotonic()))
            except Exception:
                pass  # (Node waits for each one: a full queue is a confused caller)

    def gst_encode(self, data, w, h, tw, th, fmt):
        """GStreamer's jpegenc / pngenc (gstreamer1.0-plugins-good) on one RGBx picture."""
        Gst = self.Gst
        enc = "jpegenc quality=80" if fmt == SNAP_JPEG else "pngenc compression-level=6"
        if not Gst.ElementFactory.find(enc.split()[0]):
            return None
        pipeline = Gst.parse_launch(
            "appsrc name=src format=time caps=\"video/x-raw,format=RGBx,width=%d,height=%d,framerate=0/1\" "
            "! videoscale ! videoconvert ! video/x-raw,width=%d,height=%d,pixel-aspect-ratio=1/1 ! %s "
            "! appsink name=sink sync=false" % (w, h, tw, th, enc))
        src, sink = pipeline.get_by_name("src"), pipeline.get_by_name("sink")
        pipeline.set_state(Gst.State.PLAYING)
        try:
            buf = Gst.Buffer.new_wrapped(bytes(data[:w * h * 4]))
            buf.pts, buf.duration = 0, Gst.SECOND
            src.emit("push-buffer", buf)
            src.emit("end-of-stream")
            sample = sink.emit("try-pull-sample", 5 * Gst.SECOND)
            if sample is None:
                return None
            out = sample.get_buffer()
            return fmt, out.extract_dup(0, out.get_size())
        finally:
            pipeline.set_state(Gst.State.NULL)

    def snapshot_loop(self):
        while True:
            snap_id, max_width, fmt, asked = self.snapshots.get()
            with self.frame_lock:
                # Nothing captured yet: wait a while (Node gives up after its own timeout).
                while self.frame is None and time.monotonic() - asked < 5 and not self.closing:
                    self.frame_lock.wait(0.25)
                frame = self.frame
            if frame is None or self.closing:
                continue
            data, w, h = frame
            try:
                made = encode_snapshot(data, w, h, max_width, fmt, gst_encode=self.gst_encode)
            except Exception as error:
                sys.stderr.write("snapshot: %s\n" % error)
                made = None
            if made is None:
                sys.stderr.write("snapshot: no image under the cap\n")
                continue
            out_fmt, tw, th, image = made
            self.out.write(snapshot_record(snap_id, tw, th, out_fmt, image))

    # -- commands --

    def keyframe(self):
        if self.enc is None:
            return
        event = self.GstVideo.video_event_new_upstream_force_key_unit(self.Gst.CLOCK_TIME_NONE, True, 0)
        self.enc.get_static_pad("src").send_event(event)

    def set_bitrate(self, bps):
        self.bitrate = bps
        if self.enc is None:
            return
        name = self.encoders[self.encoder_index]
        try:
            self.enc.set_property("bitrate", bps if name == "openh264enc" else max(100, bps // 1000))
        except Exception as error:
            sys.stderr.write("bitrate: %s\n" % error)

    def resize(self, w, h):
        self.box = (w, h)
        # Mutter's virtual monitor is sized by our caps: it keeps its size. Everything else (a
        # monitor, the portal's virtual output, an X11 region) is fitted into the client again.
        if self.mutter is not None or not self.src_size or (self.src and self.src[1].get("size")):
            return False  # (GNOME's portal virtual monitor, sized by our caps, keeps it too)
        self.apply_size(fit_size(*(self.part[2:] if self.part else self.src_size), w, h, self.max_long))
        return False

    def apply_size(self, size):
        if size == self.size or self.capsfilter is None:
            return
        self.size = size
        fmt = "RGBx" if self.agent else encoder_description(self.encoders[self.encoder_index], self.fps, self.bitrate)[0]
        self.capsfilter.set_property("caps", self.Gst.Caps.from_string("video/x-raw,format=%s,width=%d,height=%d,pixel-aspect-ratio=1/1" % (fmt, *size)))

    def command(self, cmd):
        if self.closing:
            return False  # (still queued on the loop while stop() flushes the portal)
        op = cmd[0]
        if op == "K":
            self.keyframe()
        elif op == "R":
            self.set_bitrate(cmd[1])
        elif op == "Z":
            self.resize(cmd[1], cmd[2])
        elif op == "Q":
            self.stop(0)
        elif op == "x":
            self.release_input()
        elif op == "S":
            self.snapshot(cmd[1], cmd[2], cmd[3])
        elif op == "g":
            zoom = zoom_command(self.magnify, cmd)
            if zoom:
                self.command(zoom)
        elif self.xtest:
            self.xtest.handle(cmd)
        elif self.mutter:
            self.mutter.input(cmd, self.input_size, self.held)
        elif self.portal:
            self.portal.input(cmd, self.node, self.input_area(), self.held)
        return False

    def input_ok(self):
        return bool(self.xtest or (self.portal and self.portal.remote) or self.mutter)

    def release_input(self):
        self.magnify.reset()
        if self.xtest:
            self.xtest.release_all()
        elif self.mutter:
            self.mutter.release(self.held, self.input_size)
        elif self.portal:
            self.portal.release(self.held)

    def read_stdin(self):
        for line in sys.stdin:
            cmd = parse_command(line)
            if cmd:
                self.GLib.idle_add(self.command, cmd)
        self.GLib.idle_add(self.stop, 0)  # Node went away

    def stop(self, code=0):
        if self.closing:
            return False
        self.closing = True
        self.code = code
        # Whatever happens below (a pipeline stuck in a state change), the helper is gone soon.
        hard = threading.Timer(3.0, lambda: os._exit(code))
        hard.daemon = True
        hard.start()
        try:
            self.release_input()
        except Exception:
            pass
        if self.local:
            self.local.stop()
        with self.frame_lock:
            self.frame_lock.notify_all()
        # The compositor's session first: it also ends the PipeWire stream, which unblocks a
        # pipewiresrc still waiting in its state change.
        for s in (self.portal, self.mutter):
            if s:
                s.close()
        if self.fd is not None:
            try:
                os.close(self.fd)
            except OSError:
                pass
            self.fd = None
        if self.pipeline:
            self.pipeline.set_state(self.Gst.State.NULL)
            self.pipeline = None
        if self.xtest:
            self.xtest.close()
        if not self.loop.is_running():
            # Stopped during setup (a nested portal wait): nothing else is left to run.
            sys.stdout.flush()
            os._exit(code)
        self.loop.quit()
        return False

    def run(self, available):
        forced = self.args.get("encoder", "auto")
        self.encoders = ["agent"] if self.agent else [forced] if forced != "auto" else available
        if not self.encoders:
            self.out.error("unavailable", "no H.264 encoder in GStreamer (install gstreamer1.0-plugins-ugly or a VA-API driver)")
            return 1
        self.code = 0
        # Before the portal: a stop (SIGTERM, Q, EOF) during its dialog still closes the session.
        # The portal's waits run the default main context, where these sources live.
        for sig in (signal.SIGTERM, signal.SIGINT, signal.SIGHUP):
            self.GLib.unix_signal_add(self.GLib.PRIORITY_HIGH, sig, lambda *_: (self.stop(0), False)[1])
        threading.Thread(target=self.read_stdin, daemon=True).start()
        try:
            self.open_source()
        except PortalError as error:
            self.out.error(error.code, str(error))
            self.stop(7 if error.code == "permission" else 1)
            return self.code
        except Exception as error:
            self.out.error("capture", "the screen could not be captured: %s" % str(error)[:80])
            self.stop(1)
            return 1
        if self.closing:
            return self.code
        for remote in (self.mutter, self.portal):
            if remote:
                remote.prime()
        if self.agent:
            self.start_agent()
        self.build()
        self.loop.run()
        return self.code


def valid_display(text):
    """An X display name like ":0", ":99.0" or "host:1": never anything a shell or XOpenDisplay could misread."""
    import re
    return bool(text) and re.fullmatch(r"[A-Za-z0-9_.-]*:\d{1,4}(\.\d{1,2})?", text) is not None


def parse_args(argv):
    out = {}
    i = 0
    while i < len(argv):
        key = argv[i]
        if not key.startswith("--") or i + 1 >= len(argv):
            raise ValueError("bad argument %r" % key)
        out[key[2:]] = argv[i + 1]
        i += 2
    return out


def validate_stream_args(a):
    if a.get("source") not in ("x11", "portal", "synthetic"):
        raise ValueError("bad --source")
    if a.get("mode", "mirror") not in ("mirror", "extend"):
        raise ValueError("bad --mode")
    a.setdefault("mode", "mirror")
    if not 1 <= int(a.get("fps", "60")) <= 240:
        raise ValueError("bad --fps")
    a.setdefault("fps", "60")
    if not 100_000 <= int(a.get("bitrate", "10000000")) <= 100_000_000:
        raise ValueError("bad --bitrate")
    a.setdefault("bitrate", "10000000")
    if a.get("encoder", "auto") not in ["auto", *ENCODERS]:
        raise ValueError("bad --encoder")
    a.setdefault("encoder", "auto")
    parse_wh(a.get("box", "1920x1080"))
    a.setdefault("box", "1920x1080")
    if a.get("size"):
        parse_wh(a["size"])
    if a["source"] == "x11":
        parse_region(a["region"])
        if not valid_display(a.get("display")):
            raise ValueError("bad --display")
    if a.get("crop") is not None:
        # X11 crops in Node (the grab region); a crop is a part of a mirrored picture, never a virtual monitor.
        if a["source"] == "x11" or a["mode"] != "mirror":
            raise ValueError("--crop only with a mirrored portal stream or the test pattern")
        parse_crop(a["crop"])
    if a.get("agent", "0") not in ("0", "1"):
        raise ValueError("bad --agent")
    if a.get("agent") == "1" and a["mode"] != "mirror":
        raise ValueError("--agent only mirrors")
    return a


def main(argv):
    if not argv:
        sys.stderr.write(__doc__)
        return 2
    verb, rest = argv[0], argv[1:]
    if verb == "probe":
        return probe_main()
    try:
        args = parse_args(rest)
    except ValueError as error:
        sys.stderr.write("%s\n" % error)
        return 2
    if verb == "input-x11":
        try:
            parse_region(args["region"])
            if not valid_display(args["display"]):
                raise ValueError("bad --display")
        except (KeyError, ValueError):
            sys.stderr.write("bad --region or --display\n")
            return 2
        return input_x11_main(args)
    if verb == "stream":
        die_with_parent()
        out = Out()
        try:
            validate_stream_args(args)
        except (KeyError, ValueError) as error:
            out.error("unavailable", str(error))
            return 2
        try:
            _, _, Gst, _ = load_gi()
        except Exception:
            out.error("unavailable", "Dokki Screen needs GStreamer for Python (python3-gi, gir1.2-gstreamer-1.0)")
            return 1
        if args.get("crop") and not Gst.ElementFactory.find("videocrop"):
            out.error("unavailable", "the screen wall needs GStreamer's videocrop (gstreamer1.0-plugins-good)")
            return 1
        # An agent sender encodes no video: no H.264 encoder needed.
        available = [] if args.get("agent") == "1" else usable_encoders([e for e in ENCODERS if Gst.ElementFactory.find(e)], bool(Gst.ElementFactory.find("h264parse")))
        return Streamer(args, out).run(available)
    sys.stderr.write(__doc__)
    return 2


if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))
