hermes.server.websocket.HermesServer Class Reference

Public Member Functions
None	__init__ (self, SharedMemoryManager shm, Scheduler|None scheduler=None, ServerConfig|None config=None, HermesConfig|None hermes_config=None, ProcessManager|None process_mgr=None)
int	client_count (self)
bool	is_running (self)
None	start (self)
None	start_background (self)
None	stop (self)
None	broadcast_telemetry (self)
None	telemetry_loop (self, float|None rate_hz=None)
asyncio.Task[None]	start_telemetry_loop (self, float|None rate_hz=None)

Protected Member Functions
None	_register_default_handlers (self)
None	_handle_client (self, ServerConnection ws)
None	_send_schema (self, ClientState client)
dict[str, Any]	_build_rich_schema (self)
dict[str, Any]	_build_basic_schema (self)
None	_handle_message (self, ClientState client, str message)
ServerMessage|None	_handle_subscribe (self, ClientState client, Command cmd)
ServerMessage|None	_handle_introspect (self, ClientState _client, Command cmd)
ServerMessage|None	_handle_pause (self, ClientState _client, Command _cmd)
ServerMessage|None	_handle_resume (self, ClientState _client, Command _cmd)
ServerMessage|None	_handle_reset (self, ClientState _client, Command _cmd)
ServerMessage|None	_handle_step (self, ClientState _client, Command cmd)
ServerMessage|None	_handle_set (self, ClientState _client, Command cmd)
list[str]	_expand_signal_patterns (self, list[str] patterns)
None	_broadcast_event (self, EventType event)
bool	_module_exists (self, str module_name)
tuple[str, str]	_split_signal_for_basic_schema (self, str signal_name)
str|None	_get_module_type (self, str module_name)
dict[str, Any]|None	_get_module_introspection (self, str module_name)
None	_broadcast_json (self, str json_str)

Protected Attributes
	_shm = shm
	_scheduler = scheduler
	_config = config or ServerConfig()
	_hermes_config = hermes_config
	_pm = process_mgr
dict	_clients = {}
Server|None	_server = None
bool	_running = False
asyncio.Task[None]|None	_telemetry_task = None
dict	_handlers = {}
	_handle_client

Detailed Description

WebSocket server for Hermes telemetry and control.

Handles client connections, schema distribution, command processing,
and telemetry streaming.

Example:
    server = HermesServer(shm, scheduler)
    await server.start()  # Runs forever

Constructor & Destructor Documentation

__init__()

None hermes.server.websocket.HermesServer.__init__	(		self,
		SharedMemoryManager	shm,
		Scheduler | None	scheduler = None,
		ServerConfig | None	config = None,
		HermesConfig | None	hermes_config = None,
		ProcessManager | None	process_mgr = None )

Initialize the Hermes server.

Args:
    shm: Shared memory manager to read telemetry from
    scheduler: Optional scheduler for control commands
    config: Server configuration
    hermes_config: Full Hermes config for schema metadata (wiring, signal info)
    process_mgr: Optional process manager for module introspection

Member Function Documentation

_broadcast_event()

None hermes.server.websocket.HermesServer._broadcast_event ( self, EventType event )

protected

Broadcast state change event to all clients.

_broadcast_json()

None hermes.server.websocket.HermesServer._broadcast_json ( self, str json_str )

protected

Broadcast JSON message to all clients.

_build_basic_schema()

dict[str, Any] hermes.server.websocket.HermesServer._build_basic_schema ( self )

protected

Build basic schema from shared memory signal names.

_build_rich_schema()

dict[str, Any] hermes.server.websocket.HermesServer._build_rich_schema ( self )

protected

Build schema with full metadata from HermesConfig.

For icarus modules (no config-declared signals), signal metadata
is derived from the shared memory signal names at runtime.

_expand_signal_patterns()

list[str] hermes.server.websocket.HermesServer._expand_signal_patterns ( self, list[str] patterns )

protected

Expand signal patterns to full signal names.

Patterns:
    - "*" matches all signals
    - "module.*" matches all signals from module
    - "module.signal" matches exact signal

_get_module_introspection()

dict[str, Any] | None hermes.server.websocket.HermesServer._get_module_introspection ( self, str module_name )

protected

Get module introspection payload for supported in-process modules.

_get_module_type()

str | None hermes.server.websocket.HermesServer._get_module_type ( self, str module_name )

protected

Get module type string for a module if available.

_handle_client()

None hermes.server.websocket.HermesServer._handle_client ( self, ServerConnection ws )

protected

Handle a new client connection.

_handle_introspect()

ServerMessage | None hermes.server.websocket.HermesServer._handle_introspect ( self, ClientState _client, Command cmd )

protected

Handle introspect command.

_handle_message()

None hermes.server.websocket.HermesServer._handle_message ( self, ClientState client, str message )

protected

Handle a text message from a client.

_handle_pause()

ServerMessage | None hermes.server.websocket.HermesServer._handle_pause ( self, ClientState _client, Command _cmd )

protected

Handle pause command.

_handle_reset()

ServerMessage | None hermes.server.websocket.HermesServer._handle_reset ( self, ClientState _client, Command _cmd )

protected

Handle reset command.

_handle_resume()

ServerMessage | None hermes.server.websocket.HermesServer._handle_resume ( self, ClientState _client, Command _cmd )

protected

Handle resume command.

_handle_set()

ServerMessage | None hermes.server.websocket.HermesServer._handle_set ( self, ClientState _client, Command cmd )

protected

Handle set command.

_handle_step()

ServerMessage | None hermes.server.websocket.HermesServer._handle_step ( self, ClientState _client, Command cmd )

protected

Handle step command.

_handle_subscribe()

ServerMessage | None hermes.server.websocket.HermesServer._handle_subscribe ( self, ClientState client, Command cmd )

protected

Handle subscribe command.

_module_exists()

bool hermes.server.websocket.HermesServer._module_exists ( self, str module_name )

protected

Check if module name is known from config, process manager, or shm signals.

_register_default_handlers()

None hermes.server.websocket.HermesServer._register_default_handlers ( self )

protected

Register default command handlers.

_send_schema()

None hermes.server.websocket.HermesServer._send_schema ( self, ClientState client )

protected

Send signal schema to a client.

If hermes_config is available, includes full signal metadata
(unit, writable) and wiring configuration. Otherwise falls back
to basic schema from shared memory signal names.

_split_signal_for_basic_schema()

tuple[str, str] hermes.server.websocket.HermesServer._split_signal_for_basic_schema ( self, str signal_name )

protected

Split a signal name into (module, local_signal) using basic-schema grouping rules.

broadcast_telemetry()

None hermes.server.websocket.HermesServer.broadcast_telemetry

(

self

)

Broadcast binary telemetry to subscribed clients.

client_count()

int hermes.server.websocket.HermesServer.client_count

(

self

)

Number of connected clients.

is_running()

bool hermes.server.websocket.HermesServer.is_running

(

self

)

Whether the server is currently running.

start()

None hermes.server.websocket.HermesServer.start

(

self

)

Start the WebSocket server.

Runs until stop() is called or the server is shut down.

start_background()

None hermes.server.websocket.HermesServer.start_background

(

self

)

Start server and telemetry loop as background tasks.

Returns immediately; use stop() to shut down.

start_telemetry_loop()

asyncio.Task[None] hermes.server.websocket.HermesServer.start_telemetry_loop	(		self,
		float | None	rate_hz = None )

Start telemetry loop as a background task.

Args:
    rate_hz: Telemetry rate in Hz (default from config)

Returns:
    The created task

stop()

None hermes.server.websocket.HermesServer.stop

(

self

)

Stop the server gracefully.

telemetry_loop()

None hermes.server.websocket.HermesServer.telemetry_loop	(		self,
		float | None	rate_hz = None )

Background task that broadcasts telemetry at fixed rate.

Args:
    rate_hz: Telemetry rate in Hz (default from config)

Raises:
    ValueError: If the effective rate is <= 0

Member Data Documentation

_clients

hermes.server.websocket.HermesServer._clients = {}

protected

_config

hermes.server.websocket.HermesServer._config = config or ServerConfig()

protected

_handle_client

hermes.server.websocket.HermesServer._handle_client

protected

_handlers

dict hermes.server.websocket.HermesServer._handlers = {}

protected

_hermes_config

hermes.server.websocket.HermesServer._hermes_config = hermes_config

protected

_pm

hermes.server.websocket.HermesServer._pm = process_mgr

protected

_running

bool hermes.server.websocket.HermesServer._running = False

protected

_scheduler

hermes.server.websocket.HermesServer._scheduler = scheduler

protected

_server

Server | None hermes.server.websocket.HermesServer._server = None

protected

_shm

hermes.server.websocket.HermesServer._shm = shm

protected

_telemetry_task

asyncio.Task[None] | None hermes.server.websocket.HermesServer._telemetry_task = None

protected

---

The documentation for this class was generated from the following file:

• src/hermes/server/websocket.py