Skip to content

PiKVMWebSocket

PiKVMWebSocket

WebSocket client for PiKVM realtime events and HID input.

Usage::

async with kvm.ws() as ws:
    async for event in ws.events():
        print(event)
Source code in src/aiopikvm/_ws.py 
 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
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
class PiKVMWebSocket:
    """WebSocket client for PiKVM realtime events and HID input.

    Usage::

        async with kvm.ws() as ws:
            async for event in ws.events():
                print(event)
    """

    def __init__(
        self,
        url: str,
        *,
        user: str,
        passwd: str,
        verify_ssl: bool = True,
        stream: int = 0,
        open_timeout: float = 10.0,
        close_timeout: float = 10.0,
    ) -> None:
        parsed = urlparse(url)
        scheme_map = {"https": "wss", "http": "ws"}
        ws_scheme = scheme_map.get(parsed.scheme, "")
        if not ws_scheme:
            raise ValueError(f"Unsupported URL scheme: {parsed.scheme!r}")
        ws_url = urlunparse(parsed._replace(scheme=ws_scheme))
        self._url = f"{ws_url}/api/ws?stream={stream}"
        self._user = user
        self._passwd = passwd
        self._verify_ssl = verify_ssl
        self._open_timeout = open_timeout
        self._close_timeout = close_timeout
        self._connection: websockets.asyncio.client.ClientConnection | None = None

    async def __aenter__(self) -> Self:
        ssl_context: ssl.SSLContext | bool | None = None
        if self._url.startswith("wss://"):
            if not self._verify_ssl:
                ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
                ssl_context.check_hostname = False
                ssl_context.verify_mode = ssl.CERT_NONE
            else:
                ssl_context = True

        headers = {
            "X-KVMD-User": self._user,
            "X-KVMD-Passwd": self._passwd,
        }

        try:
            self._connection = await websockets.asyncio.client.connect(
                self._url,
                additional_headers=headers,
                ssl=ssl_context,
                open_timeout=self._open_timeout,
                close_timeout=self._close_timeout,
            )
        except (OSError, websockets.exceptions.WebSocketException) as exc:
            raise WebSocketError(f"Failed to connect: {exc}") from exc

        return self

    async def __aexit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        if self._connection is not None:
            await self._connection.close()
            self._connection = None

    def _ensure_connected(self) -> websockets.asyncio.client.ClientConnection:
        """Return the active connection or raise."""
        if self._connection is None:
            raise WebSocketError("Not connected")
        return self._connection

    async def events(self) -> AsyncIterator[dict[str, Any]]:
        """Iterate over incoming events.

        Yields:
            Parsed JSON event dictionaries.
        """
        conn = self._ensure_connected()
        try:
            async for message in conn:
                try:
                    if isinstance(message, str):
                        yield json.loads(message)
                    else:
                        yield json.loads(message.decode())
                except (json.JSONDecodeError, UnicodeDecodeError) as exc:
                    logger.warning("Skipping malformed WebSocket message: %s", exc)
        except websockets.exceptions.ConnectionClosed:
            return

    async def send_key(self, key: str, *, state: bool) -> None:
        """Send a keyboard key event.

        Args:
            key: Key name.
            state: ``True`` for press, ``False`` for release.
        """
        conn = self._ensure_connected()
        event = {
            "event_type": "key",
            "event": {"key": key, "state": state},
        }
        await conn.send(json.dumps(event))

    async def send_mouse_move(self, to_x: int, to_y: int) -> None:
        """Move the mouse to absolute coordinates.

        Args:
            to_x: Target X coordinate.
            to_y: Target Y coordinate.
        """
        conn = self._ensure_connected()
        event = {
            "event_type": "mouse_move",
            "event": {"to": {"x": to_x, "y": to_y}},
        }
        await conn.send(json.dumps(event))

    async def send_mouse_button(self, button: str, state: bool) -> None:
        """Send a mouse button event.

        Args:
            button: Button name (e.g. ``"left"``).
            state: ``True`` for press, ``False`` for release.
        """
        conn = self._ensure_connected()
        event = {
            "event_type": "mouse_button",
            "event": {"button": button, "state": state},
        }
        await conn.send(json.dumps(event))

    async def send_mouse_wheel(self, delta_x: int, delta_y: int) -> None:
        """Send a mouse wheel event.

        Args:
            delta_x: Horizontal scroll delta.
            delta_y: Vertical scroll delta.
        """
        conn = self._ensure_connected()
        event = {
            "event_type": "mouse_wheel",
            "event": {"delta": {"x": delta_x, "y": delta_y}},
        }
        await conn.send(json.dumps(event))

    async def ping(self) -> None:
        """Send a ping event."""
        conn = self._ensure_connected()
        await conn.send(json.dumps({"event_type": "ping", "event": {}}))

events() async

Iterate over incoming events.

Yields:

Type Description
AsyncIterator[dict[str, Any]]

Parsed JSON event dictionaries.
Source code in src/aiopikvm/_ws.py 
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
async def events(self) -> AsyncIterator[dict[str, Any]]:
    """Iterate over incoming events.

    Yields:
        Parsed JSON event dictionaries.
    """
    conn = self._ensure_connected()
    try:
        async for message in conn:
            try:
                if isinstance(message, str):
                    yield json.loads(message)
                else:
                    yield json.loads(message.decode())
            except (json.JSONDecodeError, UnicodeDecodeError) as exc:
                logger.warning("Skipping malformed WebSocket message: %s", exc)
    except websockets.exceptions.ConnectionClosed:
        return

send_key(key, *, state) async

Send a keyboard key event.

Parameters:

Name Type Description Default
key str

Key name.

 required
state bool

True for press, False for release.

 required
Source code in src/aiopikvm/_ws.py 
117
118
119
120
121
122
123
124
125
126
127
128
129
async def send_key(self, key: str, *, state: bool) -> None:
    """Send a keyboard key event.

    Args:
        key: Key name.
        state: ``True`` for press, ``False`` for release.
    """
    conn = self._ensure_connected()
    event = {
        "event_type": "key",
        "event": {"key": key, "state": state},
    }
    await conn.send(json.dumps(event))

send_mouse_move(to_x, to_y) async

Move the mouse to absolute coordinates.

Parameters:

Name Type Description Default
to_x int

Target X coordinate.

 required
to_y int

Target Y coordinate.

 required
Source code in src/aiopikvm/_ws.py 
131
132
133
134
135
136
137
138
139
140
141
142
143
async def send_mouse_move(self, to_x: int, to_y: int) -> None:
    """Move the mouse to absolute coordinates.

    Args:
        to_x: Target X coordinate.
        to_y: Target Y coordinate.
    """
    conn = self._ensure_connected()
    event = {
        "event_type": "mouse_move",
        "event": {"to": {"x": to_x, "y": to_y}},
    }
    await conn.send(json.dumps(event))

send_mouse_button(button, state) async

Send a mouse button event.

Parameters:

Name Type Description Default
button str

Button name (e.g. "left").

 required
state bool

True for press, False for release.

 required
Source code in src/aiopikvm/_ws.py 
145
146
147
148
149
150
151
152
153
154
155
156
157
async def send_mouse_button(self, button: str, state: bool) -> None:
    """Send a mouse button event.

    Args:
        button: Button name (e.g. ``"left"``).
        state: ``True`` for press, ``False`` for release.
    """
    conn = self._ensure_connected()
    event = {
        "event_type": "mouse_button",
        "event": {"button": button, "state": state},
    }
    await conn.send(json.dumps(event))

send_mouse_wheel(delta_x, delta_y) async

Send a mouse wheel event.

Parameters:

Name Type Description Default
delta_x int

Horizontal scroll delta.

 required
delta_y int

Vertical scroll delta.

 required
Source code in src/aiopikvm/_ws.py 
159
160
161
162
163
164
165
166
167
168
169
170
171
async def send_mouse_wheel(self, delta_x: int, delta_y: int) -> None:
    """Send a mouse wheel event.

    Args:
        delta_x: Horizontal scroll delta.
        delta_y: Vertical scroll delta.
    """
    conn = self._ensure_connected()
    event = {
        "event_type": "mouse_wheel",
        "event": {"delta": {"x": delta_x, "y": delta_y}},
    }
    await conn.send(json.dumps(event))

ping() async

Send a ping event.

Source code in src/aiopikvm/_ws.py 
173
174
175
176
async def ping(self) -> None:
    """Send a ping event."""
    conn = self._ensure_connected()
    await conn.send(json.dumps({"event_type": "ping", "event": {}}))