MediaPipelinePixel

def __init__(
    self,
    async_event_loop: asyncio.AbstractEventLoop,
    encoder: str,
    framerate: float = 30,
    video_bitrate: int = 8000,
    audio_bitrate: int = 128000,
    width: int = 1920,
    height: int = 1080,
    audio_channels: int = 2,
    audio_enabled: bool = True,
    audio_device_name: str = "output.monitor",
    crf: int = 23,
    rc_mode: RateControlMode = RateControlMode.CBR,
    video_fullcolor: bool = False,
    video_10bit: bool = False,
    use_cpu: bool = False,
    cpu_encode_threading: bool = True,
    video_streaming_mode: bool = True,
    use_paint_over_quality: bool = True,
    video_paintover_crf: int = 18,
    video_paintover_burst_frames: int = 5,
    display_id: str = "primary",
    capture_region: Optional[Tuple[int, int]] = None,
) -> None:
    self.async_event_loop = async_event_loop
    self.display_id = display_id or "primary"
    self.capture_region: Optional[Tuple[int, int]] = capture_region
    self.audio_channels = audio_channels
    self.encoder = encoder
    self.framerate = framerate
    self.video_bitrate = video_bitrate
    self.rc_mode = rc_mode
    self.video_crf = crf
    self.video_fullcolor = video_fullcolor
    self.video_10bit = video_10bit
    self.use_cpu = use_cpu
    self.cpu_encode_threading = cpu_encode_threading
    self.video_streaming_mode = video_streaming_mode
    self.use_paint_over_quality = use_paint_over_quality
    self.video_paintover_crf = video_paintover_crf
    self.video_paintover_burst_frames = video_paintover_burst_frames
    self.audio_bitrate = audio_bitrate
    self.last_resize_success = True
    self.width = width
    self.height = height
    self.scale = 1.0
    self.audio_enabled = audio_enabled
    self.audio_device_name = audio_device_name
    self.capture_cursor = False
    # A requested IDR not yet captured: a request landing meanwhile is
    # satisfied by it; one landing after it was captured is not.
    self.idr_pending = False
    self.produce_data: Callable[..., None] = lambda buf, pts, kind, keyframe=True, timing=None, dependency=None, codec=None, anchor=False: logger.warning(
        "unhandled produce_data"
    )
    self.on_pipeline_started: Callable[[], None] = lambda: None
    # A capture that could not build its codec's encoder streams H.264 instead;
    # told the encoder it now runs, once frames flow.
    self.on_encoder_demoted: Callable[[str], None] = lambda encoder: None
    # What the capture streams and how, published on every change (`stream_stats`).
    self.on_stream_info: Callable[[str, Dict[str, Any]], Awaitable[None]] = _discard_stream_info
    self.stream_watch = stream_stats.StreamWatch(
        self.display_id, lambda did, info: self.on_stream_info(did, info))
    self.on_cursor_data: Callable[[dict], None] = lambda data: None
    self.get_cursor_size_cap: Callable[[], int] = lambda: 0

    self.capture_module: Any = None
    self.pcmflux_module: Any = None
    self._is_screen_capturing = False
    self._is_pcmflux_capturing = False
    self._running = False
    self.async_lock = asyncio.Lock()
    self._video_pts_anchor: Optional[int] = None
    # Whether the live capture has delivered a frame, for the first-frame check.
    self._framed = False
    self._last_video_pts = -1
    self._audio_capture_epoch = 0
    self._audio_cb_epoch = -1
    self._audio_pts_offset = 0
    self._audio_last_pts = -1
    self._audio_last_wall = 0.0
    self._audio_frame_samples = 480
    self._audio_routing_task: Optional[asyncio.Task] = None
    self._audio_control: Optional[AudioControl] = None
    self._audio_recover_last_attempt = 0.0
    self.stereo_companion = False

def set_stereo_companion(self, wanted: bool) -> None:
    """Have a surround capture encode its stereo companion, live, for the
    peers whose engine decodes no multiopus; stored first, so a capture
    started later takes it too. Nothing for a stereo capture."""
    self.stereo_companion = wanted
    if self.pcmflux_module is not None and self.audio_channels > 2:
        self.pcmflux_module.set_stereo_companion(wanted)

async def set_pointer_visible(self, visible: bool) -> None:
    """Toggle pixelflux cursor capture, live (the capture thread re-reads
    the flag per grab); stored first so a toggle sent during a pause or
    before a secondary display starts shapes that next start."""
    if self.capture_cursor == visible:
        return

    self.capture_cursor = visible
    if not self._is_screen_capturing or self.capture_module is None:
        return
    try:
        self.capture_module.update_tunables(self.generate_capture_settings())
        logger.debug(f"Set pointer visibility to: {visible}")
    except Exception as e:
        logger.error(f"Error setting pointer visibility: {e}", exc_info=True)

async def update_rate_control_mode(self, mode: RateControlMode) -> None:
    """Set rate control mode for the video encoder.

    Args:
        mode: Either RateControlMode.CBR or RateControlMode.CRF. A mode
            switch is structural, so it restarts the capture.
    """
    if mode not in [RateControlMode.CBR, RateControlMode.CRF]:
        logger.error(f"Invalid rate control mode: {mode}")
        return

    if mode == self.rc_mode:
        return

    self.rc_mode = mode
    if not self._is_screen_capturing or self.capture_module is None:
        return

    try:
        await self.restart_screen_capture()
        logger.info(f"Updated rate control mode to: {self.rc_mode}")
    except Exception as e:
        logger.info(f"Error updating rate control mode {e}", exc_info=True)

async def set_crf(self, crf: int) -> None:
    """Set the video encoder's target CRF, applied live in CRF mode.

    Every encoder re-reads the CRF per frame, so no restart. The value is
    stored whatever the active mode, since the settings snapshot always
    carries a CRF: one chosen under CBR is what a later switch to CRF
    encodes at. A failed live update reverts the stored value to what the
    encoder kept, so the equality guard lets the same value be retried.

    Args:
        crf: Constant-rate-factor value; lower is higher quality.
    """
    if self.video_crf == crf:
        return

    old_crf = self.video_crf
    self.video_crf = crf
    if self.rc_mode != RateControlMode.CRF:
        return
    if not self._is_screen_capturing or self.capture_module is None:
        return
    try:
        self.capture_module.update_tunables(self.generate_capture_settings())
        logger.info(f"Updated CRF live: {old_crf} -> {crf}")
    except Exception as e:
        self.video_crf = old_crf
        logger.info(f"Error updating CRF {e}", exc_info=True)

async def set_use_cpu(self, use_cpu: bool) -> None:
    """Switch h264enc between software (the pixelflux build's encoder) and
    hardware encoding.

    Unlike CRF/bitrate this is structural (a different encoder instance), so
    it restarts the capture instead of going through the live update path.
    """
    if self.use_cpu == use_cpu:
        return
    self.use_cpu = use_cpu
    if not self._is_screen_capturing or self.capture_module is None:
        return
    logger.info(f"use_cpu -> {use_cpu}; restarting screen capture")
    await self.restart_screen_capture()

async def set_cpu_encode_threading(self, threading: bool) -> None:
    """Toggle a software encoder's threads. Structural (the encoder is reopened), so restart
    capture (WS parity)."""
    if self.cpu_encode_threading == threading:
        return
    self.cpu_encode_threading = threading
    if not self._is_screen_capturing or self.capture_module is None:
        return
    logger.info(f"cpu_encode_threading -> {threading}; restarting screen capture")
    await self.restart_screen_capture()

async def _apply_tunables_live(self, what: str) -> None:
    """Push current capture settings to the running module with no restart."""
    if not self._is_screen_capturing or self.capture_module is None:
        return
    try:
        self.capture_module.update_tunables(self.generate_capture_settings())
        logger.info(f"Updated {what} live")
    except Exception as e:
        logger.info(f"Error updating {what}: {e}", exc_info=True)

async def set_video_fullcolor(self, fullcolor: bool) -> None:
    """Toggle 4:4:4 color. Structural (pixel format), so restart capture (WS parity)."""
    if self.video_fullcolor == fullcolor:
        return
    self.video_fullcolor = fullcolor
    if not self._is_screen_capturing or self.capture_module is None:
        return
    logger.info(f"video_fullcolor -> {fullcolor}; restarting screen capture")
    await self.restart_screen_capture()

async def set_video_10bit(self, ten_bit: bool) -> None:
    """Toggle 10-bit samples. Structural (pixel format), so restart capture (WS parity)."""
    if self.video_10bit == ten_bit:
        return
    self.video_10bit = ten_bit
    if not self._is_screen_capturing or self.capture_module is None:
        return
    logger.info(f"video_10bit -> {ten_bit}; restarting screen capture")
    await self.restart_screen_capture()

async def set_encoder(self, encoder: str) -> None:
    """Switch the WebRTC video encoder. Structural (a different encoder
    instance), so restart capture — same as use_cpu (WS parity)."""
    if self.encoder == encoder:
        return
    self.encoder = encoder
    if not self._is_screen_capturing or self.capture_module is None:
        return
    logger.info(f"encoder -> {encoder}; restarting screen capture")
    await self.restart_screen_capture()

async def set_video_streaming_mode(self, enabled: bool) -> None:
    """Toggle Turbo (stream every frame vs damage-gated). Live tunable."""
    if self.video_streaming_mode == enabled:
        return
    self.video_streaming_mode = enabled
    await self._apply_tunables_live(f"streaming mode -> {enabled}")

async def set_use_paint_over_quality(self, enabled: bool) -> None:
    if self.use_paint_over_quality == enabled:
        return
    self.use_paint_over_quality = enabled
    await self._apply_tunables_live(f"paint-over -> {enabled}")

async def set_video_paintover_crf(self, crf: int) -> None:
    if self.video_paintover_crf == crf:
        return
    self.video_paintover_crf = crf
    await self._apply_tunables_live(f"paint-over CRF -> {crf}")

async def set_video_paintover_burst_frames(self, frames: int) -> None:
    if self.video_paintover_burst_frames == frames:
        return
    self.video_paintover_burst_frames = frames
    await self._apply_tunables_live(f"paint-over burst -> {frames}")

async def set_video_bitrate(self, bitrate: int) -> None:
    """Set the video encoder's target bitrate, applied live in CBR mode.

    The value is stored whatever the active mode, since the settings
    snapshot always carries a bitrate: one chosen under CRF is what a
    later switch to CBR encodes at. A failed live update reverts the
    stored value: congestion control steers from `video_bitrate` as the
    rate currently being encoded, and the revert also lets the equality
    guard retry the same value.

    Args:
        bitrate: Target bitrate in kbps.
    """
    if bitrate <= 0 or self.video_bitrate == bitrate:
        return

    old_bitrate = self.video_bitrate
    self.video_bitrate = bitrate
    if self.rc_mode == RateControlMode.CRF:
        return
    if not self._is_screen_capturing or self.capture_module is None:
        return
    try:
        self.capture_module.update_video_bitrate(int(bitrate))
        logger.info(
            f"Updated video bitrate: {old_bitrate} -> {bitrate} kbps"
        )
    except Exception as e:
        self.video_bitrate = old_bitrate
        logger.info(f"Error updating video bitrate {e}", exc_info=True)

async def set_audio_bitrate(self, bitrate: int) -> None:
    """Set the Opus encoder's target bitrate, applied live.

    A failed live update restarts the audio pipeline (websockets parity):
    the stored value is what the fresh capture starts at, so the encoder
    never keeps running at a rate the pipeline no longer reports.

    Args:
        bitrate: Target bitrate in bps.
    """
    if bitrate <= 0 or self.audio_bitrate == bitrate:
        return

    old_bitrate = self.audio_bitrate
    self.audio_bitrate = bitrate
    if not self._is_pcmflux_capturing or self.pcmflux_module is None:
        return
    try:
        self.pcmflux_module.update_audio_bitrate(bitrate)
        logger.info(
            f"Updated audio bitrate: {old_bitrate // 1000} -> {bitrate // 1000} kbps"
        )
    except Exception as e:
        logger.warning(
            f"Live audio bitrate update failed ({e}); restarting audio pipeline."
        )
        await self._stop_audio_pipeline()
        await self._start_audio_pipeline()

async def set_framerate(self, framerate: float) -> None:
    """Set the pixelflux capture rate, applied live."""
    async with self.async_lock:
        if framerate <= 0 or self.framerate == framerate:
            return

        self.framerate = framerate
        if not self._is_screen_capturing or self.capture_module is None:
            return
        self.capture_module.update_framerate(float(self.framerate))
        logger.info(f"Updated framerate to {fps_label(self.framerate)} fps")

def shed_framerate(self, fps: Optional[float]) -> None:
    """Run the capture at `fps` under the configured rate, or back at the configured rate (None), leaving
    the configured rate itself (`framerate`, which the display's modes follow) as it is."""
    if not self._is_screen_capturing or self.capture_module is None:
        return
    self.capture_module.update_framerate(float(fps if fps else self.framerate))

async def dynamic_idr_frame(self) -> None:
    """Request an IDR frame from pixelflux; `idr_pending` holds until it is captured."""
    if not self._is_screen_capturing or self.capture_module is None:
        return
    try:
        self.idr_pending = True
        self.capture_module.request_idr_frame()
        logger.debug("IDR frame requested successfully")
    except Exception as e:
        logger.error(f"Error requesting IDR frame: {e}", exc_info=True)

def invalidate_reference(self, frame_id: int, withheld: bool = False) -> None:
    """A consumer lost `frame_id`, or its sender withheld it (`withheld`): the frames after
    it stop predicting from it."""
    if self._is_screen_capturing and self.capture_module is not None:
        try:
            self.capture_module.invalidate_reference(frame_id & 0xFFFF, withheld=withheld)
        except TypeError:
            # A pixelflux that has no withheld report takes it as a loss.
            self.capture_module.invalidate_reference(frame_id & 0xFFFF)

def set_link_rate(self, kbps: int) -> None:
    """Tell the encoder what the path carries for a still screen's cleanup (pixelflux's
    `update_link_rate`), 0 where it is not known."""
    if self._is_screen_capturing and self.capture_module is not None and hasattr(
            self.capture_module, "update_link_rate"):
        self.capture_module.update_link_rate(int(kbps))

def acknowledge_reference(self, frame_id: int, held: bool = True) -> None:
    """Every consumer holds `frame_id`, or where not `held` was sent it (`CommonFrames`)."""
    if self._is_screen_capturing and self.capture_module is not None:
        self.capture_module.acknowledge_reference(frame_id & 0xFFFF, held)

def generate_capture_settings(self) -> Any:
    """Build the pixelflux CaptureSettings snapshot for the current state.

    A secondary display's `capture_region` pins its exact geometry
    (auto-adjust would balloon the region to the whole root); the primary
    auto-sizes from (0, 0). Both backends keep pixelflux's per-stripe
    header: its picture-type byte says whether the encoder produced a
    keyframe, which the video bridge needs to keep the wire decodable
    across a drop. The header is sliced off the zero-copy view before RTP
    framing.

    Returns:
        A populated `pixelflux.CaptureSettings` (annotated as Any because
        the import is optional).
    """
    cs = CaptureSettings()
    cs.capture_width = self.width
    cs.capture_height = self.height
    if self.capture_region is not None:
        cs.capture_x = int(self.capture_region[0])
        cs.capture_y = int(self.capture_region[1])
        cs.auto_adjust_screen_capture_size = False
    else:
        cs.capture_x = 0
        cs.capture_y = 0
        cs.auto_adjust_screen_capture_size = True
    cs.codec = codec_for_encoder(self.encoder)
    cs.omit_stripe_headers = False
    apply_common_capture_settings(
        cs, app_settings,
        is_wayland=bool(app_settings.wayland[0]),
        display_name=self.display_id,
        scale=self.scale,
        framerate=self.framerate,
        encoder=self.encoder,
        use_cpu=self.use_cpu,
        cbr=self.rc_mode == RateControlMode.CBR,
        bitrate_kbps=self.video_bitrate,
        crf=self.video_crf,
        paintover_crf=self.video_paintover_crf,
        paintover_burst=self.video_paintover_burst_frames,
        fullcolor=self.video_fullcolor,
        ten_bit=self.video_10bit,
        streaming=self.video_streaming_mode,
        use_paint_over_quality=self.use_paint_over_quality,
        capture_cursor=self.capture_cursor,
        cursor_size_cap_hint=int(self.get_cursor_size_cap() or 0),
        cpu_encode_threading=self.cpu_encode_threading,
    )
    return cs

def _screen_capture_callback(self, frame: Any) -> None:
    """Deliver one encoded video frame; runs on the pixelflux capture thread.

    The frame owns its native buffer and goes downstream as a zero-copy
    memoryview sliced past the header, with its keyframe flag and codec
    read off the header's type byte and, for a frame whose encoder tracks
    its references, the frame's id and the id it predicts from: only
    whole-frame sessions track them, and a frame is judged by that rather
    than by the pipeline's size, which a resize changes while frames of
    the old size and the new are still arriving;
    `produce_data` wraps it in an EncodedPacket and keeps a reference so
    the frame stays alive. pts (90 kHz) is the frame's capture instant, as
    pixelflux stamped it on CLOCK_MONOTONIC, from the pipeline-scoped
    anchor rather than `frame.frame_id`: the u16 counter wraps, restarts at
    0 on every capture restart, and its implied step changes on live fps
    raises, all backward RTP jumps on a live sender. The capture instant
    rather than the delivery keeps the encode's varying duration out of
    the RTP clock, so a receiver mapping it through the sender reports
    reads capture times; a path that stamps no capture falls back to the
    delivery. Ties bump one tick so pts is strictly increasing. Only one
    capture thread exists at a time (stop joins before a new start), so
    this state needs no lock; and `produce_data` is synchronous, so
    `call_soon_threadsafe` delivers it with no per-frame Future, matching
    the websockets path. The frame's capture and encode instants travel
    with it for the sender's timing extensions and reports, the capture
    one being the instant its pts was taken at.
    """
    self._framed = True
    try:
        view = memoryview(frame)
        if len(view) > STRIPE_HEADER_LEN:
            keyframe = view[0] != 0x04 or (view[1] & 0x0F) == 0x01
            if keyframe:
                self.idr_pending = False
            data_bytes = view[STRIPE_HEADER_LEN:]
            captured = frame.capture_ns
            if captured <= 0:
                captured = time.monotonic_ns()
            if self._video_pts_anchor is None:
                self._video_pts_anchor = captured
            pts = (captured - self._video_pts_anchor) * 9 // 100_000
            if pts <= self._last_video_pts:
                pts = self._last_video_pts + 1
            self._last_video_pts = pts
            timing = (captured, frame.encode_start_ns, frame.encode_end_ns)
            reference = frame.reference_frame_id
            dependency = None
            if reference != -2:
                dependency = (frame.frame_id & 0xFFFF, None if reference == -1 else reference)
            codec = WIRE_VIDEO_MIMES.get(view[1] >> 4) if view[0] == 0x04 else None
            anchor = view[0] == 0x04 and bool(view[1] & FRAME_ANCHOR)
            self.async_event_loop.call_soon_threadsafe(
                functools.partial(self.produce_data, data_bytes, pts, "video", keyframe,
                                  timing=timing, dependency=dependency, codec=codec, anchor=anchor)
            )

    except Exception as e:
        logger.error(f"Error in capture callback: {e}", exc_info=False)

def _pixelflux_cursor_handler(
    self, msg_type: str, data_bytes: Optional[bytes], hot_x: int, hot_y: int
) -> None:
    """Translate pixelflux cursor events (either backend) into client cursor
    messages (websockets parity). Runs on the capture-side thread; delivery
    hops to the asyncio loop."""
    try:
        size = int(getattr(app_settings, "cursor_size", -1) or -1)
        if size <= 0:
            size = 24
        payload = format_pixelflux_cursor(msg_type, data_bytes, hot_x, hot_y, size)
        if payload is None:
            return
        self.async_event_loop.call_soon_threadsafe(self.on_cursor_data, payload)
    except Exception as e:
        logger.error(f"Error handling pixelflux cursor: {e}")

async def start_screen_capture(self) -> None:
    """Start pixelflux screen capture at the current settings snapshot.

    pixelflux is the cursor source on both backends (compositor on
    Wayland, XFixes monitor on X11); an older X11-only build stashes the
    callback harmlessly and the input handler's monitor keeps delivering.
    The capture counts as running from before the start is awaited: its
    first frames reach the video bridge while the start is still under
    way, and what the bridge and the other live setters ask of it in reply
    (a lost frame, a keyframe, a rate) has to reach it rather than be
    dropped as meant for a capture that is not running.

    Raises:
        MediaPipelineError: When pixelflux is unavailable or the capture
            fails to start, so the caller never reports a live stream
            while nothing is captured.
    """
    if self._is_screen_capturing:
        return

    if ScreenCapture is None or CaptureSettings is None:
        raise MediaPipelineError(
            "pixelflux is unavailable (missing or ABI/version-skewed wheel); "
            "cannot start screen capture"
        )

    settings = self.generate_capture_settings()

    try:
        self.capture_module = ScreenCapture()
        self.capture_module.set_cursor_callback(self._pixelflux_cursor_handler)
        self._framed = False
        self._is_screen_capturing = True
        await asyncio.to_thread(
            self.capture_module.start_capture,
            self._screen_capture_callback,
            settings,
        )
        logger.info("Started screen capture module")
        module = self.capture_module
        asyncio.get_running_loop().call_later(
            FIRST_FRAME_WAIT_S,
            lambda: self._is_screen_capturing and self.capture_module is module and not self._framed
            and logger.warning(no_first_frame(self.display_id, str(getattr(settings, 'codec', '?')))))
        self._schedule_active_codec_settle()
        self.stream_watch.follow(module, self.encoder, bool(self.use_cpu))
    except Exception as e:
        logger.error(f"Failed to start screen capture: {e}", exc_info=True)
        self.capture_module = None
        self._is_screen_capturing = False
        raise MediaPipelineError(f"screen capture failed to start: {e}") from e

def _schedule_active_codec_settle(self, attempt: int = 0) -> None:
    """Read back, once frames flow, the codec the capture streams, and report a
    demotion through `on_encoder_demoted` so the senders and clients follow."""
    self.async_event_loop.call_later(
        1.0, lambda: asyncio.ensure_future(self._settle_active_codec(attempt)))

async def _settle_active_codec(self, attempt: int) -> None:
    module = self.capture_module
    if module is None or not self._is_screen_capturing or not hasattr(module, "active_codec"):
        return
    try:
        active = await asyncio.to_thread(module.active_codec)
    except Exception as e:
        logger.debug(f"Active codec unknown: {e}")
        return
    if active is None:
        if attempt < 5:
            self._schedule_active_codec_settle(attempt + 1)
        return
    if codec_for_encoder(self.encoder) == active:
        return
    demoted = encoder_for_codec(active)
    logger.warning(f"The capture streams {active} as '{demoted}': no encoder served '{self.encoder}'.")
    self.encoder = demoted
    self.on_encoder_demoted(demoted)

async def update_capture_region(self, x: int, y: int, w: int, h: int) -> None:
    """Re-target the capture to a new region of the extended framebuffer.

    Live on X11, with no restart. On Wayland the output is the capture
    region and a start on the live module reconfigures it in place, so it
    restarts; so does a failed live re-target or one with no live capture.
    """
    self.capture_region = (int(x), int(y))
    self.width, self.height = int(w), int(h)
    if (self._is_screen_capturing and self.capture_module is not None
            and not bool(app_settings.wayland[0])):
        try:
            await asyncio.to_thread(
                self.capture_module.update_capture_region, int(x), int(y), int(w), int(h)
            )
            return
        except Exception as e:
            logger.warning(f"Live capture re-target failed ({e}); restarting capture.")
    await self.restart_screen_capture()

async def stop_screen_capture(self) -> None:
    """Stop the pixelflux capture, dropping the module and its cursor
    callback even on error."""
    if not self._is_screen_capturing or self.capture_module is None:
        return
    self.stream_watch.stop()
    release_pixelflux_cursor_callback(self.capture_module)
    try:
        await asyncio.to_thread(self.capture_module.stop_capture)
        self.capture_module = None
        self._is_screen_capturing = False
        logger.info("Stopped screen capture module")
    except Exception as e:
        logger.error(f"Error stopping screen capture: {e}", exc_info=True)
        self.capture_module = None
        self._is_screen_capturing = False

async def pause_screen_capture(self) -> bool:
    """Consumer-aware pause: stop only the screen capture (audio and the
    pipeline's running state survive) once every consumer of this display
    is hidden; resume_screen_capture restarts it.

    Returns:
        Whether a live capture was actually stopped.
    """
    async with self.async_lock:
        if not self._running or not self._is_screen_capturing:
            return False
        await self.stop_screen_capture()
        return True

async def resume_screen_capture(self) -> bool:
    """Restart a capture stopped by pause_screen_capture, at the CURRENT
    settings (a resize/scale received while paused applies here).

    Returns:
        Whether a stopped capture was actually started.
    """
    async with self.async_lock:
        if not self._running or self._is_screen_capturing:
            return False
        await self.start_screen_capture()
        return True

async def pause_audio_capture(self) -> bool:
    """Stop only the audio capture once no consumer receives it; the
    pipeline keeps running and resume_audio_capture restarts it.

    Returns:
        Whether a live capture was actually stopped.
    """
    async with self.async_lock:
        if not self._running or not self._is_pcmflux_capturing:
            return False
        await self._stop_audio_pipeline()
        return True

async def resume_audio_capture(self) -> bool:
    """Start the audio capture a consumer now wants, unless audio is
    disabled or it already runs.

    Returns:
        Whether a stopped capture was actually started.
    """
    async with self.async_lock:
        if not self._running or not self.audio_enabled or self._is_pcmflux_capturing:
            return False
        await self._start_audio_pipeline()
        return self._is_pcmflux_capturing

def capture_ended(self) -> bool:
    """Whether the X11 capture this pipeline runs ended on its own while
    the pipeline still counts it running: pixelflux gives up on a region it
    cannot grab or an X server that does not come back. Wayland reports its
    own through the host reap."""
    module = self.capture_module
    if not self._is_screen_capturing or module is None or bool(app_settings.wayland[0]):
        return False
    try:
        return not bool(module.is_capturing)
    except Exception:
        return False

async def revive_screen_capture(self) -> bool:
    """Start again, at the current settings, a capture that ended on its
    own (`capture_ended`), as websockets rebuilds a stale capture on the
    next resize or start. Returns whether one was started."""
    async with self.async_lock:
        if not self.capture_ended():
            return False
        logger.warning(f"The capture of '{self.display_id}' ended on its own; starting it again.")
        await self.stop_screen_capture()
        try:
            await self.start_screen_capture()
        except MediaPipelineError as e:
            logger.error(f"The capture of '{self.display_id}' did not start again: {e}")
            return False
        return True

async def restart_screen_capture(self) -> None:
    """Reapply the current settings snapshot to the live capture module.

    Starting on the live module lets pixelflux apply the settings in
    place (a compatible Wayland encoder session survives; X11 cycles the
    capture internally). Only structural changes land here; rate/quality
    knobs go through the live `update_*` paths. The running check is
    made under the lock so a capture stopped by a concurrent
    `stop_media_pipeline` is not resurrected.
    """
    async with self.async_lock:
        if not self._is_screen_capturing:
            return
        try:
            settings = self.generate_capture_settings()
            await asyncio.to_thread(
                self.capture_module.start_capture,
                self._screen_capture_callback,
                settings,
            )
            logger.info("Screen capture reconfigured")
        except Exception as e:
            logger.error(f"Error restarting screen capture: {e}")

async def _start_audio_pipeline(self) -> None:
    """Start pcmflux Opus capture; best-effort (video survives its absence).

    The capture sink is provisioned before the device check (websockets
    order): a PipeWire that comes up with no sink has no `output.monitor`
    to validate against yet, and the check would rewrite the device name
    to `auto_null.monitor` for good. Opus runs VBR around `audio_bitrate`
    (better quality per bit than CBR; RTP carries variable payloads and
    browsers decode it natively, with no `cbr=1` in the offer fmtp to
    contradict), at the configured frame duration (RTP has no fixed
    ptime, so shorter frames flow unchanged) and without pcmflux's 2-byte
    header, since WebRTC repacketizes into RTP. `start_capture` returns
    Ok while the worker is still bringing the device up, so a device that
    dies during bring-up surfaces later through `recover_audio_if_failed`.
    """
    if self._is_pcmflux_capturing:
        return

    if AudioCapture is None or AudioCaptureSettings is None:
        logger.error(
            "pcmflux is unavailable (missing or ABI/version-skewed wheel); "
            "skipping audio capture"
        )
        return

    control = self._get_audio_control()
    await control.ensure_capture_sink(self.audio_device_name)
    await self._ensure_audio_device()
    try:
        capture_settings = AudioCaptureSettings()
        device_name_bytes = (
            self.audio_device_name.encode("utf-8")
            if self.audio_device_name
            else None
        )
        capture_settings.device_name = device_name_bytes
        capture_settings.sample_rate = 48000
        capture_settings.channels = self.audio_channels
        capture_settings.opus_bitrate = int(self.audio_bitrate)
        frame_ms = float(getattr(app_settings, 'audio_frame_duration_ms', '20') or 20)
        capture_settings.frame_duration_ms = frame_ms
        self._audio_frame_samples = max(1, int(48000 * frame_ms / 1000))
        capture_settings.use_vbr = True
        # See opus_capture_settings: silence costs nothing on the wire, and the
        # receiver's concealment counter rises across the gap without anything
        # audible changing.
        capture_settings.use_silence_gate = True
        capture_settings.latency_ms = int(min(10, frame_ms))
        capture_settings.debug_logging = bool(app_settings.debug[0])
        capture_settings.omit_audio_header = True
        pcmflux_settings = capture_settings

        logger.info(
            f"Starting pcmflux audio pipeline: device '{self.audio_device_name}', "
            f"{capture_settings.opus_bitrate} bps, {capture_settings.channels} ch."
        )

        def audio_capture_callback(frame: Any) -> None:
            """Deliver one Opus frame; runs on the pcmflux capture thread.

            The frame goes downstream as a zero-copy memoryview that
            `produce_data` keeps a reference to, stamped on the
            pipeline's continuous audio clock (`_audio_rtp_pts`).
            """
            try:
                if len(frame) > 0:
                    data_bytes = memoryview(frame)
                    pts = self._audio_rtp_pts(int(frame.pts), time.monotonic())
                    # A surround capture's stereo companion shares its frame's pts.
                    kind = ("audio_stereo" if self.audio_channels > 2
                            and frame.channels < self.audio_channels else "audio")

                    self.async_event_loop.call_soon_threadsafe(
                        self.produce_data, data_bytes, pts, kind
                    )
            except Exception as e:
                logger.info(f"Error audio capture callback: {e}")

        self.pcmflux_module = AudioCapture()
        if self.audio_channels > 2:
            self.pcmflux_module.set_stereo_companion(self.stereo_companion)
        # Before start_capture, so the first frame re-anchors on the new epoch.
        self._audio_capture_epoch += 1
        await asyncio.to_thread(
            self.pcmflux_module.start_capture,
            pcmflux_settings,
            audio_capture_callback,
        )
        self._is_pcmflux_capturing = True
        self._audio_recover_last_attempt = 0.0
        self._audio_routing_task = asyncio.create_task(self._enforce_audio_routing())
        self._audio_routing_task.add_done_callback(
            lambda t: (not t.cancelled() and t.exception() is not None)
            and logger.error(f"Audio routing task failed: {t.exception()}")
        )
        state = getattr(self.pcmflux_module, "state", "running")
        logger.debug(f"pcmflux audio capture started (state: {state}).")
    except Exception as e:
        logger.error(f"Failed to start pcmflux audio pipeline: {e}", exc_info=True)
        await self._stop_audio_pipeline()
        return

def _audio_rtp_pts(self, raw_pts: int, now: float) -> int:
    """The pts, on the pipeline's continuous 48 kHz audio clock, of a
    capture frame whose own pts is `raw_pts` and which was delivered at
    monotonic time `now`.

    pcmflux re-zeros its sample clock on every capture start, so the
    first frame of a new `_audio_capture_epoch` re-anchors the offset: it
    lands as far past the last frame as the wall clock moved between the
    two deliveries, and at least one frame past it. A backward RTP jump on
    a live sender plays as a glitch, and one that ignores the time the
    capture was stopped (an audio pause and resume) makes the receiver
    take the first packet after it for a packet that late: libwebrtc's
    jitter buffer then holds that much audio, a second for a one-second
    pause, for about twenty seconds.
    """
    if self._audio_cb_epoch != self._audio_capture_epoch:
        self._audio_cb_epoch = self._audio_capture_epoch
        if self._audio_last_pts >= 0:
            elapsed = round((now - self._audio_last_wall) * 48000)
            self._audio_pts_offset = (
                self._audio_last_pts + max(self._audio_frame_samples, elapsed) - raw_pts)
        else:
            self._audio_pts_offset = 0
    pts = self._audio_pts_offset + raw_pts
    self._audio_last_pts = pts
    self._audio_last_wall = now
    return pts

def _get_audio_control(self) -> AudioControl:
    """The sound-server control client, created on the pipeline's loop."""
    if self._audio_control is None:
        self._audio_control = AudioControl("selkies-webrtc-audio")
    return self._audio_control

async def _enforce_audio_routing(self) -> None:
    """Move the pcmflux stream onto the configured source if PipeWire strayed.

    PipeWire often ignores the requested audio device and connects
    recording apps to the default source, particularly when switching
    between streaming modes.
    """
    # Give pcmflux a fraction of a second to initialize its PA stream.
    await asyncio.sleep(0.5)
    try:
        await self._get_audio_control().route_pcmflux([self.audio_device_name])
    except Exception as e:
        logger.error(f"Error enforcing WebRTC audio routing: {e}")

async def _ensure_audio_device(self) -> None:
    """Verify audio_device_name is a valid source, else fall back to the
    default sink's monitor (or PipeWire's `auto_null.monitor`)."""
    try:
        resolved = await self._get_audio_control().resolve_capture_source(
            self.audio_device_name
        )
        if resolved:
            self.audio_device_name = resolved
    except Exception as e:
        logger.error(f"Error validating the audio device: {e}")

async def _stop_audio_pipeline(self) -> None:
    """Cancel routing enforcement, release the sound-server control
    connection, and stop the pcmflux capture.

    Routing enforcement belongs to the capture session that armed it: a
    stop inside its start-up delay must not move a stream that is gone.
    It is gathered with `return_exceptions` so its own cancellation is
    absorbed while a cancel of this coroutine still propagates.
    """
    task = self._audio_routing_task
    self._audio_routing_task = None
    if task is not None:
        task.cancel()
        await asyncio.gather(task, return_exceptions=True)
    if self._audio_control is not None:
        await self._audio_control.aclose()
    if not self._is_pcmflux_capturing or not self.pcmflux_module:
        return

    logger.info("Stopping pcmflux audio pipeline...")
    self._is_pcmflux_capturing = False
    if self.pcmflux_module:
        try:
            await asyncio.to_thread(self.pcmflux_module.stop_capture)
        except Exception as e:
            logger.error(f"Error during pcmflux stop_capture: {e}")
        finally:
            self.pcmflux_module = None

        logger.debug("pcmflux audio pipeline stopped.")
    return

async def recover_audio_if_failed(self) -> bool:
    """Restart the audio capture if its worker has died, else no-op.

    pcmflux reports the start handshake as Ok while the worker is still
    "starting", so a device that fails right after start surfaces only
    later, as ``last_error`` on the capture object. This is polled where
    pipeline health is already evaluated and cycles the audio capture
    through the normal stop/start path; the video pipeline is untouched.
    A short floor between attempts keeps a permanently broken device from
    restarting every tick, and the module is re-checked under the lock
    since a concurrent restart may have replaced it or the pipeline may
    have stopped.

    Returns:
        True when a restart was attempted.
    """
    if not self.audio_enabled or not self._is_pcmflux_capturing:
        return False
    module = self.pcmflux_module
    if module is None:
        return False
    err = getattr(module, "last_error", None)
    if not err:
        return False
    now = time.monotonic()
    if now - self._audio_recover_last_attempt < 5.0:
        return False
    self._audio_recover_last_attempt = now
    async with self.async_lock:
        if self.pcmflux_module is not module or not self._running:
            return False
        logger.warning(f"pcmflux audio worker failed ({err}); restarting audio capture.")
        await self._stop_audio_pipeline()
        await self._start_audio_pipeline()
    return True

async def start_media_pipeline(self, video: bool = True, audio: bool = True) -> None:
    """Mark the pipeline running and start the captures asked for: the
    screen when `video`, and pcmflux when `audio` and audio is enabled.
    A capture left off here starts through its resume call. A start
    failure tears both back down and stays stopped. A raising
    `on_pipeline_started` does not abort a started pipeline."""
    async with self.async_lock:
        if self._running:
            return

        logger.debug("Starting media pipeline...")
        try:
            if video:
                await self.start_screen_capture()
            else:
                logger.info("Screen capture starts paused: no consumer receives video yet.")

            if not self.audio_enabled:
                logger.debug(
                    "Audio pipeline is disabled, skipping audio capture startup."
                )
            elif audio:
                await self._start_audio_pipeline()
            else:
                logger.info("Audio capture starts paused: no consumer receives audio yet.")
            self._running = True
            try:
                self.on_pipeline_started()
            except Exception:
                logger.warning(
                    "on_pipeline_started callback raised; pipeline remains running",
                    exc_info=True,
                )
        except Exception as e:
            logger.error(f"Error starting media pipelines: {e}", exc_info=True)
            # Not stop_media_pipeline(): it would re-enter the held lock.
            try:
                await self.stop_screen_capture()
                if self.audio_enabled:
                    await self._stop_audio_pipeline()
            except Exception:
                logger.error("Error during start-failure cleanup", exc_info=True)
            self._running = False

async def stop_media_pipeline(self) -> None:
    """Stop both captures and mark the pipeline stopped."""
    async with self.async_lock:
        if not self._running:
            return

        logger.debug("Stopping media pipeline...")
        try:
            await self.stop_screen_capture()

            if self.audio_enabled:
                await self._stop_audio_pipeline()
            self._running = False
        except Exception as e:
            logger.error(f"Error stopping media pipelines: {e}", exc_info=True)

def is_media_pipeline_running(self) -> bool:
    return self._running

def is_screen_capturing(self) -> bool:
    """True once the screen capture starts, which precedes `_running` (the
    audio half of the pipeline may still be starting); settings that must
    reach a live capture key on this, not on the whole pipeline."""
    return self._is_screen_capturing