trio_ws Module

Overview

Trio-websocket specific implementation for Neuro API interactions.

This module provides concrete implementations of the Neuro API using the trio-websocket library for asynchronous websocket communication. It includes both a basic API client and a full event-driven component for integration with the libcomponent system.

Key Features

  • Async websocket communication using trio-websocket

  • Automatic connection management and error handling

  • Integration with trio’s structured concurrency

  • Event-driven component architecture

  • Graceful connection cleanup and shutdown

Classes

TrioNeuroAPI

class neuro_api.trio_ws.TrioNeuroAPI(game_title: str, connection: WebSocketConnection | None = None)

Bases: AbstractNeuroAPI

Trio-websocket specific Neuro API implementation.

A concrete implementation of AbstractNeuroAPI using trio-websocket for websocket communication. This class provides the core websocket functionality without the event system integration.

Connection Management:

  • Manages websocket connection state

  • Provides connection status checking

  • Handles graceful connection setup and teardown

__init__(game_title: str, connection: WebSocketConnection | None = None) → None

Initialize NeuroAPI.

Parameters:

  • game_title (str) – The title of the game being managed.

  • connection (trio_websocket.WebSocketConnection, optional) – Initial websocket connection. Defaults to None.

property not_connected: bool

Check if websocket connection is not established.

Returns:

True if connection is None, False otherwise.

Return type:

bool

property connection: WebSocketConnection

Get current websocket connection.

Returns:

The active websocket connection.

Return type:

trio_websocket.WebSocketConnection

Raises:

RuntimeError – If no websocket connection is established.

connect(websocket: WebSocketConnection | None) → None

Set the internal websocket connection.

Parameters:

websocket (trio_websocket.WebSocketConnection | None) – Websocket connection to set. Can be None to clear connection.

async write_to_websocket(data: str) → None

Write a message to the websocket.

Parameters:

data (str) – Message to be sent over the websocket.

Raises:

ConnectionClosed – If websocket connection is closed or closing.

async read_from_websocket() → bytes | bytearray | memoryview | str

Read a message from the websocket.

Returns:

The received message.

Return type:

bytes | bytearray | memoryview | str

Raises:

  • trio_websocket.ConnectionClosed – On websocket connection error.

  • trio.BrokenResourceError – If internal memory channel is broken (rarely occurs).

  • AssertionError – If received message types are unexpected.

TrioNeuroAPIComponent

class neuro_api.trio_ws.TrioNeuroAPIComponent(component_name: str, game_title: str, connection: WebSocketConnection | None = None)

Bases: AbstractNeuroAPIComponent, TrioNeuroAPI

Trio-websocket Neuro API Component.

Combines AbstractNeuroAPIComponent and TrioNeuroAPI functionality for Trio-based websocket interactions with Neuro.

A full-featured component combining event-driven architecture with trio-websocket communication. This is the recommended class for most game integrations.

Event Handling:

  • Automatic websocket connection management

  • Built-in message reading loop

  • Connection event handling

  • Graceful shutdown support

__init__(component_name: str, game_title: str, connection: WebSocketConnection | None = None) → None

Initialize Trio-websocket Neuro API Component.

Parameters:

  • component_name (str) – Name of the component.

  • game_title (str) – Title of the game being managed.

  • connection (trio_websocket.WebSocketConnection, optional) – Initial websocket connection. Defaults to None.

async read_message() → None

Read message from Neuro websocket.

Automatically processes various types of commands: - actions/reregister_all - Graceful and immediate shutdown requests - Action commands - Unknown commands

Raises:

  • ValueError – If extra or missing keys in action command data.

  • TypeError – On action command key type mismatch.

  • trio_websocket.ConnectionClosed – If websocket connection is lost.

websocket_connect_failed() → None

Handle websocket connection handshake failure.

Default behavior is to print an error message. This method can be overridden for custom error handling.

async websocket_connect_successful() → None

Handle successful websocket connection.

Default behavior is to print a success message and create a trio checkpoint.

This method can be overridden for custom connection handling.

async handle_connect(event: Event[str]) → None

Handle websocket connection event.

Manages the full websocket connection process, including: - Opening the websocket - Handling connection errors - Reading messages until disconnected

Parameters:

event (Event[str]) – Connection event containing the websocket URL.

Note

Does not stop unless stop() function is explicitly called.

async stop(code: int = 1000, reason: str | None = None) → None

Close websocket and trigger disconnection.

Parameters:

  • code (int, optional) – Websocket close code. Defaults to 1000 (normal closure).

  • reason (str, optional) – Reason for closing the connection. Defaults to None.

name

Usage Examples

Basic Usage with Component Manager

import trio
from libcomponent.component import Event, ExternalRaiseManager
from neuro_api.trio_ws import TrioNeuroAPIComponent

async def main():
    async with trio.open_nursery() as nursery:
        manager = ExternalRaiseManager("game_manager", nursery)

        component = TrioNeuroAPIComponent("neuro_api", "My Game")
        manager.add_component(component)

        # Connect to websocket
        await manager.raise_event(Event("connect", "ws://localhost:8000"))

        # Wait for connection and setup
        await component.wait_for_websocket()

        if not component.not_connected:
            # Game logic here
            await trio.sleep(10)
            await component.stop()

Custom Connection Handling

class CustomGameComponent(TrioNeuroAPIComponent):
    def __init__(self, name, title):
        super().__init__(name, title)
        self.connection_ready = trio.Event()

    def websocket_connect_failed(self):
        print("Connection failed - retrying...")
        self.connection_ready.set()

    async def websocket_connect_successful(self):
        print("Successfully connected to Neuro!")
        await self.send_startup_command()
        await self.send_context("Game is ready!")
        self.connection_ready.set()

    async def wait_for_connection(self):
        await self.connection_ready.wait()

Direct API Usage

import trio
import trio_websocket
from neuro_api.trio_ws import TrioNeuroAPI

async def direct_usage():
    async with trio_websocket.open_websocket_url("ws://localhost:8000") as ws:
        api = TrioNeuroAPI("My Game", ws)

        await api.send_startup_command()
        await api.send_context("Hello from my game!")

        # Read messages manually
        while True:
            try:
                await api.read_message()
            except trio_websocket.ConnectionClosed:
                break

Connection Management

Connection States

The component manages several connection states:

  • Disconnected: No websocket connection (not_connected returns True)

  • Connecting: Websocket handshake in progress

  • Connected: Active websocket connection ready for communication

  • Closing: Connection shutdown initiated

Error Handling

The implementation handles various error conditions:

  • HandshakeError: Connection setup failures

  • ConnectionClosed: Unexpected disconnections

  • BrokenResourceError: Internal trio channel issues

Connection events trigger appropriate callbacks:

def websocket_connect_failed(self):
    """Called when websocket handshake fails"""
    # Custom error handling

async def websocket_connect_successful(self):
    """Called when websocket connects successfully"""
    # Custom setup logic

Message Processing

The component automatically processes various message types:

  • action: Neuro action requests

  • actions/reregister_all: Action re-registration requests

  • shutdown/graceful: Graceful shutdown requests

  • shutdown/immediate: Immediate shutdown requests

Message processing runs in a continuous loop until the connection is closed or stop() is called.

Component Management

The TrioNeuroAPIComponent is designed to work with libcomponent’s ExternalRaiseManager. The component must be added to a manager and the connection initiated through a “connect” event.

Shutdown Handling

The component supports graceful shutdown:

# Graceful shutdown with custom close code
await component.stop(code=1001, reason="Game ended")

The stop() method cleanly closes the websocket connection and sets the component to a disconnected state.