navidrome/plugins/pdk/python/host/nd_host_websocket.py

# Code generated by ndpgen. DO NOT EDIT.
#
# This file contains client wrappers for the WebSocket host service.
# It is intended for use in Navidrome plugins built with extism-py.
#
# IMPORTANT: Due to a limitation in extism-py, you cannot import this file directly.
# The @extism.import_fn decorators are only detected when defined in the plugin's
# main __init__.py file. Copy the needed functions from this file into your plugin.

from dataclasses import dataclass
from typing import Any

import extism
import json
import base64


class HostFunctionError(Exception):
    """Raised when a host function returns an error."""
    pass


@extism.import_fn("extism:host/user", "websocket_connect")
def _websocket_connect(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


@extism.import_fn("extism:host/user", "websocket_sendtext")
def _websocket_sendtext(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


@extism.import_fn("extism:host/user", "websocket_sendbinary")
def _websocket_sendbinary(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


@extism.import_fn("extism:host/user", "websocket_closeconnection")
def _websocket_closeconnection(offset: int) -> int:
    """Raw host function - do not call directly."""
    ...


def websocket_connect(url: str, headers: Any, connection_id: str) -> str:
    """Connect establishes a WebSocket connection to the specified URL.

Plugins that use this function must also implement the WebSocketCallback capability
to receive incoming messages and connection events.

Parameters:
  - url: The WebSocket URL to connect to (ws:// or wss://)
  - headers: Optional HTTP headers to include in the handshake request
  - connectionID: Optional unique identifier for the connection. If empty, one will be generated

Returns the connection ID that can be used to send messages or close the connection,
or an error if the connection fails.

    Args:
        url: str parameter.
        headers: Any parameter.
        connection_id: str parameter.

    Returns:
        str: The result value.

    Raises:
        HostFunctionError: If the host function returns an error.
    """
    request = {
        "url": url,
        "headers": headers,
        "connectionId": connection_id,
    }
    request_bytes = json.dumps(request).encode("utf-8")
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _websocket_connect(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))

    if response.get("error"):
        raise HostFunctionError(response["error"])

    return response.get("newConnectionId", "")


def websocket_send_text(connection_id: str, message: str) -> None:
    """SendText sends a text message over an established WebSocket connection.

Parameters:
  - connectionID: The connection identifier returned by Connect
  - message: The text message to send

Returns an error if the connection is not found or if sending fails.

    Args:
        connection_id: str parameter.
        message: str parameter.

    Raises:
        HostFunctionError: If the host function returns an error.
    """
    request = {
        "connectionId": connection_id,
        "message": message,
    }
    request_bytes = json.dumps(request).encode("utf-8")
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _websocket_sendtext(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))

    if response.get("error"):
        raise HostFunctionError(response["error"])



def websocket_send_binary(connection_id: str, data: bytes) -> None:
    """SendBinary sends binary data over an established WebSocket connection.

Parameters:
  - connectionID: The connection identifier returned by Connect
  - data: The binary data to send

Returns an error if the connection is not found or if sending fails.

    Args:
        connection_id: str parameter.
        data: bytes parameter.

    Raises:
        HostFunctionError: If the host function returns an error.
    """
    request = {
        "connectionId": connection_id,
        "data": base64.b64encode(data).decode("ascii"),
    }
    request_bytes = json.dumps(request).encode("utf-8")
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _websocket_sendbinary(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))

    if response.get("error"):
        raise HostFunctionError(response["error"])



def websocket_close_connection(connection_id: str, code: int, reason: str) -> None:
    """CloseConnection gracefully closes a WebSocket connection.

Parameters:
  - connectionID: The connection identifier returned by Connect
  - code: WebSocket close status code (e.g., 1000 for normal closure)
  - reason: Optional human-readable reason for closing

Returns an error if the connection is not found or if closing fails.

    Args:
        connection_id: str parameter.
        code: int parameter.
        reason: str parameter.

    Raises:
        HostFunctionError: If the host function returns an error.
    """
    request = {
        "connectionId": connection_id,
        "code": code,
        "reason": reason,
    }
    request_bytes = json.dumps(request).encode("utf-8")
    request_mem = extism.memory.alloc(request_bytes)
    response_offset = _websocket_closeconnection(request_mem.offset)
    response_mem = extism.memory.find(response_offset)
    response = json.loads(extism.memory.string(response_mem))

    if response.get("error"):
        raise HostFunctionError(response["error"])