network.py

Network module for client-server communication with proper architecture.
Uses JSON for safe serialization and implements a clean protocol.
FIXED: Server now properly responds to ping messages
ADDED: Host class that runs server in thread and connects as client

class MessageType

Description

Types of messages in the protocol

Attributes
HANDSHAKE: Any = 'handshake'
AUTHENTICATION: Any = 'auth'
DATA: Any = 'data'
COMMAND: Any = 'command'
EVENT: Any = 'event'
ERROR: Any = 'error'
PING: Any = 'ping'
PONG: Any = 'pong'
DISCONNECT: Any = 'disconnect'
Methods

No methods defined.

class UserType

Description

Types of network users

Attributes
SERVER: Any = 'server'
CLIENT: Any = 'client'
HOST: Any = 'host'
Methods

No methods defined.

class NetworkMessage

Description

Standard message format for network communication

Attributes
message_id: str = None
message_type: MessageType = None
sender_id: str = None
sender_type: UserType = None
timestamp: float = None
payload: Optional[Any] = None
target: Optional[str] = None
Methods
def to_dict(self: Any) -> Dict[str, Any]
Convert to dictionary for JSON serialization
def from_dict(cls: Any, data: Dict[str, Any]) -> 'NetworkMessage'
Create from dictionary

class NetworkClient

Description

Client for connecting to network server

Methods
def __init__(self: Any, client_id: Optional[str] = None) -> Any
No documentation
def connect(self: Any, host: str, port: int, timeout: int = 5) -> bool
Connect to server with timeout
def disconnect(self: Any) -> None
Disconnect from server
def send(self: Any, message_type: MessageType, payload: Any = None, target: Optional[str] = None) -> bool
Send message to server
def _send_raw(self: Any, data: Dict[str, Any]) -> bool
Send raw data to server
def _receive_loop(self: Any) -> None
Receive messages in a loop
def _handle_message(self: Any, data: Dict[str, Any]) -> None
Handle incoming message
def _ping_loop(self: Any) -> None
Send periodic ping to keep connection alive
def _send_pong(self: Any, ping_message_id: str) -> None
Send pong response to a ping
def register_callback(self: Any, message_type: MessageType, callback: Callable) -> None
Register callback for specific message type
def get_message(self: Any, timeout: Optional[float] = None) -> Optional[NetworkMessage]
Get message from queue with timeout
def __enter__(self: Any) -> Any
No documentation
def __exit__(self: Any, exc_type: Any, exc_val: Any, exc_tb: Any) -> Any
No documentation

class NetworkServer

Description

Server for handling multiple client connections

Methods
def __init__(self: Any, host: str = '0.0.0.0', port: int = 4723, max_clients: int = 10) -> Any
No documentation
def start(self: Any) -> bool
Start the server
def stop(self: Any) -> None
Stop the server
def _accept_connections(self: Any) -> None
Accept incoming connections (non-blocking)
def _handle_client(self: Any, client_id: str, client_socket: socket.socket) -> None
Handle communication with a single client
def _process_client_message(self: Any, client_id: str, data: Dict[str, Any]) -> None
Process message from client
def _handle_auth(self: Any, client_id: str, payload: Any) -> None
Handle client authentication
def _send_pong(self: Any, client_id: str, ping_message: NetworkMessage) -> None
Send pong response to client ping
def _send_to_client(self: Any, client_id: str, data: Dict[str, Any]) -> bool
Send data to specific client
def broadcast(self: Any, data: Dict[str, Any], exclude: Optional[List[str]] = None) -> int
Broadcast data to all connected clients
def _send_error(self: Any, client_id: str, message: str) -> None
Send error message to client
def _disconnect_client(self: Any, client_id: str, send_notification: bool = True) -> None
Disconnect and clean up client
def get_client_count(self: Any) -> int
Get number of connected clients
def get_client_ids(self: Any) -> List[str]
Get list of connected client IDs
def register_callback(self: Any, message_type: MessageType, callback: Callable) -> None
Register callback for specific message type
def get_message(self: Any, timeout: Optional[float] = None) -> Optional[tuple]
Get message from queue with timeout
def enable_auth(self: Any, tokens: List[str]) -> None
Enable authentication with provided tokens
def generate_auth_token(self: Any) -> str
Generate a new authentication token
def __enter__(self: Any) -> Any
No documentation
def __exit__(self: Any, exc_type: Any, exc_val: Any, exc_tb: Any) -> Any
No documentation

class NetworkHost

Description

Host class that runs a server in thread and connects as a client

Methods
def __init__(self: Any, host: str = '127.0.0.1', port: int = 4723, max_clients: int = 10) -> Any
No documentation
def start(self: Any) -> bool
Start the host (server + connect as client)
def _run_server(self: Any) -> Any
Run the server (called in thread)
def stop(self: Any) -> Any
Stop the host
def send_as_host(self: Any, message_type: MessageType, payload: Any = None, target: Optional[str] = None) -> bool
Send message as host client
def broadcast_as_server(self: Any, data: Dict[str, Any], exclude: Optional[List[str]] = None) -> int
Broadcast message as server to all clients
def send_to_client(self: Any, client_id: str, message_type: MessageType, payload: Any = None) -> bool
Send message to specific client as server
def get_message(self: Any, timeout: Optional[float] = None) -> Optional[tuple]
Get message from queue with timeout
def get_client_count(self: Any) -> int
Get number of connected clients (excluding host client)
def get_client_ids(self: Any) -> List[str]
Get list of connected client IDs (excluding host client)
def __enter__(self: Any) -> Any
No documentation
def __exit__(self: Any, exc_type: Any, exc_val: Any, exc_tb: Any) -> Any
No documentation

Global Functions

def generate_id() -> str

Generate unique ID for clients/messages

def validate_port(port: int) -> bool

Validate port number

def safe_json_dumps(data: Any) -> bytes

Safely serialize data to JSON bytes

def safe_json_loads(data: bytes) -> Any

Safely deserialize JSON bytes

Back to Backend Module