Skip to content

StreamerResource

StreamerResource

Streamer management — screenshots and OCR for PiKVM.

Source code in src/aiopikvm/resources/streamer.py 
 7
 8
 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
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) -> bytes:
        """Take a JPEG screenshot.

        Returns:
            Raw JPEG image bytes.
        """
        response = await self._get_raw("/api/streamer/snapshot", accept="image/jpeg")
        return response.content

    async def ocr(self) -> str:
        """Perform OCR on the current screen.

        Returns:
            Recognized text.
        """
        result = await self._get("/api/streamer/ocr")
        return OCRResult.model_validate(result).ocr

    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 
10
11
12
13
14
15
16
17
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() async

Take a JPEG screenshot.

Returns:

Type Description
bytes

Raw JPEG image bytes.
Source code in src/aiopikvm/resources/streamer.py 
19
20
21
22
23
24
25
26
async def snapshot(self) -> bytes:
    """Take a JPEG screenshot.

    Returns:
        Raw JPEG image bytes.
    """
    response = await self._get_raw("/api/streamer/snapshot", accept="image/jpeg")
    return response.content

ocr() async

Perform OCR on the current screen.

Returns:

Type Description
str

Recognized text.
Source code in src/aiopikvm/resources/streamer.py 
28
29
30
31
32
33
34
35
async def ocr(self) -> str:
    """Perform OCR on the current screen.

    Returns:
        Recognized text.
    """
    result = await self._get("/api/streamer/ocr")
    return OCRResult.model_validate(result).ocr

delete_snapshot() async

Delete the cached snapshot.

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