VirtualWebcam

def __init__(self) -> None:
    self._cam: Optional[Any] = None
    self._lock = asyncio.Lock()
    self._start_failed_logged = False
    self._push_orientation: Optional[bool] = None
    self._device_mjpeg = False
    self._reformat_blocked_logged = False
    self._reformat_next_check = 0.0

def needs_ensure(self, codec: Optional[int]) -> bool:
    """Whether ``ensure`` has work to do for a frame of this codec.

    The per-frame answer for a running camera whose format already suits the uplink is no,
    which is the whole of the hot path; the rest is the rare case where an ``auto`` device
    was shaped by an uplink of the other kind and may be worth re-creating.
    """
    if self._cam is None:
        return True
    # The kind comparison first: it answers every frame of a running camera, and the
    # setting lookup below builds a string the hot path has no use for.
    if codec is None or (codec == CODEC_MJPEG) == self._device_mjpeg:
        return False
    if not self._auto_format():
        return False
    return time.monotonic() >= self._reformat_next_check

@staticmethod
def _auto_format() -> bool:
    return str(app_settings.webcam_pixel_format or "auto").strip().lower() == "auto"

def _consumers(self) -> Optional[str]:
    """What is reading the camera right now, or None when nothing is.

    Each sink answers for its own: the interposer counts the clients on its socket,
    PipeWire reports whether a consumer is linked to its node, and a kernel device's
    openers are found the only way a process can, through /proc. A pixelflux
    that cannot report the node's consumers is taken to have one: a re-created
    device would take the picture away from whoever is watching.
    """
    cam = self._cam
    if cam is None:
        return None
    try:
        stats = cam.stats()
    except Exception:
        return "unknown"
    if int(stats.get("clients", 0) or 0) > 0:
        return "interposer client"
    if "pipewire_streaming" in stats:
        if stats.get("pipewire_streaming"):
            return "PipeWire consumer"
    elif stats.get("pipewire"):
        return "a PipeWire node whose consumers this pixelflux does not report"
    device = str(stats.get("device_path") or "")
    if device and _device_has_openers(device):
        return f"an application holding {device}"
    return None

def _settings(self, codec: Optional[int]) -> Any:
    s = VirtualCameraSettings()
    s.socket_path = webcam_socket_path()
    s.width = int(app_settings.webcam_width)
    s.height = int(app_settings.webcam_height)
    s.fps_num = 30
    s.fps_den = 1
    s.pixel_format = device_pixel_format(str(app_settings.webcam_pixel_format), codec)
    s.device_path = str(app_settings.webcam_device)
    s.pipewire = bool(app_settings.webcam_pipewire[0])
    return s

async def ensure(self, codec: Optional[int] = None) -> Optional[Any]:
    """Returns the running camera, starting it on first use (off the event loop).

    An ``auto`` device already running in the other format is re-created for this codec
    when nothing is reading it, so a session that follows one of the other kind neither
    transcodes every frame (a video uplink fitted into an MJPEG device) nor decodes one it
    could have passed through. A device with a consumer is left exactly as it is:
    applications hold it open across mode switches, and pulling the format out from under
    one is worse than the transcode.

    Args:
        codec: Codec of the frame that brings the camera up, or that found the running
            device in the other format; an ``auto`` device format follows it (see
            ``device_pixel_format``).
    """
    if self._cam is not None:
        if not self.needs_ensure(codec):
            return self._cam
        async with self._lock:
            if self._cam is None or not self.needs_ensure(codec):
                return self._cam
            reader = await asyncio.to_thread(self._consumers)
            if reader is not None:
                self._reformat_next_check = time.monotonic() + REFORMAT_RECHECK_SECONDS
                if not self._reformat_blocked_logged:
                    self._reformat_blocked_logged = True
                    logger.info(
                        "Virtual webcam stays %s for %s: %s is reading it. Frames are "
                        "converted for the device; pin webcam_pixel_format to avoid it.",
                        "MJPEG" if self._device_mjpeg else "raw",
                        "an MJPEG uplink" if codec == CODEC_MJPEG else "a video uplink", reader)
                return self._cam
            logger.info("Virtual webcam re-created for the %s uplink now that nothing reads it.",
                        "MJPEG" if codec == CODEC_MJPEG else "video")
            await self._stop_locked()
    if not webcam_available():
        if not self._start_failed_logged:
            self._start_failed_logged = True
            logger.error("pixelflux VirtualCamera unavailable; webcam forwarding disabled.")
        return None
    async with self._lock:
        if self._cam is not None:
            return self._cam
        cam = VirtualCamera()
        try:
            settings = self._settings(codec)
            await asyncio.to_thread(cam.start, settings)
        except Exception as exc:
            if not self._start_failed_logged:
                self._start_failed_logged = True
                logger.error("Virtual webcam start failed: %s", exc)
            return None
        self._cam = cam
        self._device_mjpeg = str(settings.pixel_format).strip().upper() in ("MJPEG", "MJPG", "JPEG")
        self._reformat_blocked_logged = False
        self._reformat_next_check = 0.0
        stats = cam.stats()
        logger.info("Virtual webcam serving %s (%dx%d %s, kernel device: %s, PipeWire node: %s)",
                    cam.socket_path, settings.width, settings.height, settings.pixel_format,
                    cam.device_path or "none", "yes" if stats.get("pipewire") else "no")
        return cam

def push(self, data: Any, codec: int, keyframe: bool = False, offset: int = 0,
         rotation: int = 0, flip: bool = False) -> int:
    """Hands one encoded frame to the camera; returns its flags (``KEYFRAME_WANTED`` bit).

    Args:
        rotation: Clockwise degrees (0/90/180/270) that make the decoded frame upright.
        flip: Horizontal mirror, applied after the rotation.

    The orientation is forwarded only when the frame carries one and the installed
    pixelflux takes it, so a build whose ``push`` predates the arguments keeps
    working (announced once; such frames are published as sent). A camera that is
    not running yet (``ensure`` pending) drops the frame silently.
    """
    cam = self._cam
    if cam is None:
        return 0
    try:
        if (rotation or flip) and self._orientation_accepted(cam):
            return int(cam.push(data, codec, keyframe, offset, rotation, flip))
        return int(cam.push(data, codec, keyframe, offset))
    except Exception as exc:
        logger.error("Virtual webcam push failed: %s", exc)
        return 0

def _orientation_accepted(self, cam: Any) -> bool:
    """Whether this camera's ``push`` takes the orientation arguments.

    Decided once from the signature the extension publishes, rather than from a
    ``TypeError`` per frame: that exception is also what a bad argument raises,
    and mistaking one for an old build silently drops the orientation for the
    rest of the session. A build that publishes no signature is taken at its
    word, and a refusal then surfaces through ``push``'s own error path.
    """
    if self._push_orientation is not None:
        return self._push_orientation
    try:
        params = inspect.signature(cam.push).parameters
        accepted = "rotation" in params or any(
            p.kind in (p.VAR_POSITIONAL, p.VAR_KEYWORD) for p in params.values())
    except (TypeError, ValueError):
        accepted = True
    self._push_orientation = accepted
    if not accepted:
        logger.warning(
            "Installed pixelflux takes no webcam orientation; rotated uplinks are published as sent.")
    return accepted

def keyframe_wanted(self, flags: int) -> bool:
    return bool(flags & getattr(VirtualCamera, "KEYFRAME_WANTED", 1))

async def _stop_locked(self) -> None:
    """Stops the running camera; the caller holds the lock."""
    cam = self._cam
    self._cam = None
    if cam is not None:
        try:
            await asyncio.to_thread(cam.stop)
        except Exception:
            pass

async def stop(self) -> None:
    async with self._lock:
        await self._stop_locked()