Skip to content

StreamerResource

StreamerResource

Streamer management — screenshots and OCR for PiKVM.

Source code in src/aiopikvm/resources/streamer.py 
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
class StreamerResource(BaseResource):
    """Streamer management — screenshots and OCR for PiKVM."""

    async def get_state(self) -> StreamerState:
        """Get the current streamer state.

        Returns:
            Current streamer subsystem state.
        """
        result = await self._get("/api/streamer")
        return StreamerState.model_validate(result)

    async def snapshot(self, *, allow_offline: bool = False) -> bytes:
        """Take a JPEG screenshot.

        Without ``allow_offline``, kvmd returns HTTP 503 whenever the
        video source is not online (host asleep, HDMI unplugged, etc.).
        Passing ``allow_offline=True`` makes kvmd return a "NO LIVE VIDEO"
        placeholder JPEG instead. The flag has no effect when the streamer
        process is fully stopped (no UI clients) — the call still fails
        with HTTP 503.

        Args:
            allow_offline: When ``True``, accept a placeholder frame if the
                video source is offline.

        Returns:
            Raw JPEG image bytes.
        """
        params: dict[str, Any] = {}
        if allow_offline:
            params["allow_offline"] = 1
        response = await self._get_raw(
            "/api/streamer/snapshot",
            params=params or None,
            accept="image/jpeg",
        )
        return response.content

    async def get_ocr_info(self) -> OCRInfo:
        """Get OCR capability metadata (enabled flag, available languages).

        Returns:
            Installed OCR languages and the default selection.
        """
        result = await self._get("/api/streamer/ocr")
        return OCRInfo.model_validate(result["ocr"])

    async def ocr(
        self,
        *,
        langs: list[str] | None = None,
        allow_offline: bool = False,
        timeout: float = 30.0,
    ) -> str:
        """Perform OCR on the current screen.

        Sends ``GET /api/streamer/snapshot?ocr=1`` — the kvmd snapshot
        endpoint with the ``ocr`` flag, which returns recognized text as
        ``text/plain`` instead of a JPEG.

        Args:
            langs: Tesseract language codes (e.g. ``["eng"]``,
                ``["eng", "rus"]``). When omitted the kvmd default is used.
                Available languages can be queried via :meth:`get_ocr_info`.
            allow_offline: When ``True``, run OCR on the "NO LIVE VIDEO"
                placeholder if the video source is offline; otherwise
                kvmd returns HTTP 503. No effect if the streamer process
                is fully stopped.
            timeout: Per-call timeout in seconds. OCR runs Tesseract on the
                Pi CPU and is intrinsically slow (10-20 s for full-screen),
                so the default is wider than the client-level default.

        Returns:
            Recognized text.
        """
        params: dict[str, Any] = {"ocr": 1}
        if langs:
            params["ocr_langs"] = ",".join(langs)
        if allow_offline:
            params["allow_offline"] = 1
        response = await self._get_raw(
            "/api/streamer/snapshot",
            params=params,
            accept="text/plain",
            timeout=timeout,
        )
        return response.text

    async def delete_snapshot(self) -> None:
        """Delete the cached snapshot."""
        await self._delete("/api/streamer/snapshot")

get_state() async

Get the current streamer state.

Returns:

Type Description
StreamerState

Current streamer subsystem state.
Source code in src/aiopikvm/resources/streamer.py 
12
13
14
15
16
17
18
19
async def get_state(self) -> StreamerState:
    """Get the current streamer state.

    Returns:
        Current streamer subsystem state.
    """
    result = await self._get("/api/streamer")
    return StreamerState.model_validate(result)

snapshot(*, allow_offline=False) async

Take a JPEG screenshot.

Without allow_offline, kvmd returns HTTP 503 whenever the video source is not online (host asleep, HDMI unplugged, etc.). Passing allow_offline=True makes kvmd return a "NO LIVE VIDEO" placeholder JPEG instead. The flag has no effect when the streamer process is fully stopped (no UI clients) — the call still fails with HTTP 503.

Parameters:

Name Type Description Default
allow_offline bool

When True, accept a placeholder frame if the video source is offline.

 False

Returns:

Type Description
bytes

Raw JPEG image bytes.
Source code in src/aiopikvm/resources/streamer.py 
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
async def snapshot(self, *, allow_offline: bool = False) -> bytes:
    """Take a JPEG screenshot.

    Without ``allow_offline``, kvmd returns HTTP 503 whenever the
    video source is not online (host asleep, HDMI unplugged, etc.).
    Passing ``allow_offline=True`` makes kvmd return a "NO LIVE VIDEO"
    placeholder JPEG instead. The flag has no effect when the streamer
    process is fully stopped (no UI clients) — the call still fails
    with HTTP 503.

    Args:
        allow_offline: When ``True``, accept a placeholder frame if the
            video source is offline.

    Returns:
        Raw JPEG image bytes.
    """
    params: dict[str, Any] = {}
    if allow_offline:
        params["allow_offline"] = 1
    response = await self._get_raw(
        "/api/streamer/snapshot",
        params=params or None,
        accept="image/jpeg",
    )
    return response.content

get_ocr_info() async

Get OCR capability metadata (enabled flag, available languages).

Returns:

Type Description
OCRInfo

Installed OCR languages and the default selection.
Source code in src/aiopikvm/resources/streamer.py 
48
49
50
51
52
53
54
55
async def get_ocr_info(self) -> OCRInfo:
    """Get OCR capability metadata (enabled flag, available languages).

    Returns:
        Installed OCR languages and the default selection.
    """
    result = await self._get("/api/streamer/ocr")
    return OCRInfo.model_validate(result["ocr"])

ocr(*, langs=None, allow_offline=False, timeout=30.0) async

Perform OCR on the current screen.

Sends GET /api/streamer/snapshot?ocr=1 — the kvmd snapshot endpoint with the ocr flag, which returns recognized text as text/plain instead of a JPEG.

Parameters:

Name Type Description Default
langs list[str] | None

Tesseract language codes (e.g. ["eng"], ["eng", "rus"]). When omitted the kvmd default is used. Available languages can be queried via :meth:get_ocr_info.

 None
allow_offline bool

When True, run OCR on the "NO LIVE VIDEO" placeholder if the video source is offline; otherwise kvmd returns HTTP 503. No effect if the streamer process is fully stopped.

 False
timeout float

Per-call timeout in seconds. OCR runs Tesseract on the Pi CPU and is intrinsically slow (10-20 s for full-screen), so the default is wider than the client-level default.

 30.0

Returns:

Type Description
str

Recognized text.
Source code in src/aiopikvm/resources/streamer.py 
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
async def ocr(
    self,
    *,
    langs: list[str] | None = None,
    allow_offline: bool = False,
    timeout: float = 30.0,
) -> str:
    """Perform OCR on the current screen.

    Sends ``GET /api/streamer/snapshot?ocr=1`` — the kvmd snapshot
    endpoint with the ``ocr`` flag, which returns recognized text as
    ``text/plain`` instead of a JPEG.

    Args:
        langs: Tesseract language codes (e.g. ``["eng"]``,
            ``["eng", "rus"]``). When omitted the kvmd default is used.
            Available languages can be queried via :meth:`get_ocr_info`.
        allow_offline: When ``True``, run OCR on the "NO LIVE VIDEO"
            placeholder if the video source is offline; otherwise
            kvmd returns HTTP 503. No effect if the streamer process
            is fully stopped.
        timeout: Per-call timeout in seconds. OCR runs Tesseract on the
            Pi CPU and is intrinsically slow (10-20 s for full-screen),
            so the default is wider than the client-level default.

    Returns:
        Recognized text.
    """
    params: dict[str, Any] = {"ocr": 1}
    if langs:
        params["ocr_langs"] = ",".join(langs)
    if allow_offline:
        params["allow_offline"] = 1
    response = await self._get_raw(
        "/api/streamer/snapshot",
        params=params,
        accept="text/plain",
        timeout=timeout,
    )
    return response.text

delete_snapshot() async

Delete the cached snapshot.

Source code in src/aiopikvm/resources/streamer.py 
 98
 99
100
async def delete_snapshot(self) -> None:
    """Delete the cached snapshot."""
    await self._delete("/api/streamer/snapshot")