pyqt_reactive.protocols.zmq_server_protocol

ZMQ server protocol abstractions for generic server browser widgets.

This module provides data structures and ABCs for building ZMQ server browser widgets that work with any zmqruntime-based server.

Design Principles: - Frozen dataclasses for immutability - ABCs for extensibility (no protocols) - Enum-based type safety - No hasattr/getattr/get with fallback

Functions

freeze_mapping([data])

Create immutable mapping for frozen dataclass fields.

Classes

`ProgressUpdate`(task_id, percent, phase, ...)	Generic progress update.
`ServerKind`(*values)	Server kind for type-safe dispatch.
`ServerNode`(port, kind, server_name, ready, ...)	Generic server node for UI display.
`ZMQServerRenderer`()	ABC for rendering server nodes.

pyqt_reactive.protocols.zmq_server_protocol.freeze_mapping(data: Mapping[str, Any] | None = None) → Mapping[str, Any][source]: Create immutable mapping for frozen dataclass fields.

class pyqt_reactive.protocols.zmq_server_protocol.ServerKind(*values)[source]

Server kind for type-safe dispatch.

EXECUTION = 1

VIEWER = 2

GENERIC = 3

class pyqt_reactive.protocols.zmq_server_protocol.ServerNode(port: int, kind: ServerKind, server_name: str, ready: bool, status_icon: str, log_file_path: str | None, children: ServerNode, ...]=(), metadata: Mapping[str, ~typing.Any]=<factory>)[source]

Generic server node for UI display.

All domain-specific data goes into metadata as a dict.

port: int

kind: ServerKind

server_name: str

ready: bool

status_icon: str

log_file_path: str | None

children: Tuple[ServerNode, ...] = ()

metadata: Mapping[str, Any]

__init__(port: int, kind: ServerKind, server_name: str, ready: bool, status_icon: str, log_file_path: str | None, children: ServerNode, ...]=(), metadata: Mapping[str, ~typing.Any]=<factory>) → None

class pyqt_reactive.protocols.zmq_server_protocol.ProgressUpdate(task_id: str, percent: float, phase: str, status: str, step_name: str, metadata: Mapping[str, ~typing.Any]=<factory>)[source]

Generic progress update.

task_id: str

percent: float

phase: str

status: str

step_name: str

metadata: Mapping[str, Any]

__init__(task_id: str, percent: float, phase: str, status: str, step_name: str, metadata: Mapping[str, ~typing.Any]=<factory>) → None

class pyqt_reactive.protocols.zmq_server_protocol.ZMQServerRenderer[source]

ABC for rendering server nodes.

Subclasses implement render_server_node and render_child_node to provide domain-specific formatting.

abstractmethod render_server_node(node: ServerNode) → Tuple[str, str, str][source]

Return (display_text, status_text, info_text) for server node.

The returned tuple is used to populate the QTreeWidgetItem columns: - display_text: Shown in “Server” column - status_text: Shown in “Status” column - info_text: Shown in “Info” column

abstractmethod render_child_node(node: ServerNode) → Tuple[str, str, str][source]: Return (display_text, status_text, info_text) for child node.