""" Hard-coded Command Classes for Trixy Network Protocol This module implements the hard-coded commands specified in CLAUDE.md for performance optimization. These commands are sent as simple text strings rather than serialized objects to minimize overhead. Hard-coded Commands (all start with "TRXI"): - TRXINOOP - No-op/heartbeat command - TRXIPING - Ping command for connectivity testing - TRXIPONG - Pong response to ping - TRXIPRNT - Print string command (for testing) - TRXYHELO - Hello command (for debugging) These commands bypass normal serialization for maximum performance and are handled specially by the protocol layer. """ import time from typing import Optional from dataclasses import dataclass, field from .base import TrixyCommand, CommandResponse, ValidationError def pprint(message: str) -> None: """Hard-coded command logging function.""" print(f"[CMD.HARDCODED] {message}") @dataclass class NoOpCommand(TrixyCommand): """ No-operation command for heartbeat/keepalive functionality. This command does nothing but confirm the connection is alive. Used for periodic heartbeat between client and server. """ def validate(self) -> bool: """Validate no-op command (always valid).""" return True def get_command_name(self) -> str: """Get command name.""" return "TRXINOOP" def get_timeout(self) -> float: """Short timeout for heartbeat.""" return 5.0 def get_priority(self) -> int: """Low priority for heartbeat.""" return 90 @dataclass class PingCommand(TrixyCommand): """ Ping command for connectivity testing and latency measurement. The recipient should respond with a PongCommand containing the same ping_id for round-trip time calculation. """ ping_id: str = field(default_factory=lambda: str(int(time.time() * 1000))) def validate(self) -> bool: """Validate ping command.""" if not self.ping_id: raise ValidationError("PingCommand must have a ping_id") return True def get_command_name(self) -> str: """Get command name.""" return "TRXIPING" def get_response_class(self): """Expect a pong response.""" return PongCommand def get_timeout(self) -> float: """Short timeout for ping.""" return 10.0 def get_priority(self) -> int: """High priority for connectivity testing.""" return 20 @dataclass class PongCommand(TrixyCommand): """ Pong response to ping command. This command is sent in response to a PingCommand and should contain the same ping_id for latency calculation. """ ping_id: str = field(default="") original_timestamp: Optional[float] = field(default=None) def validate(self) -> bool: """Validate pong command.""" if not self.ping_id: raise ValidationError("PongCommand must have a ping_id") return True def get_command_name(self) -> str: """Get command name.""" return "TRXIPONG" def get_timeout(self) -> float: """Short timeout for pong.""" return 5.0 def get_priority(self) -> int: """High priority for connectivity testing.""" return 20 def calculate_rtt(self) -> Optional[float]: """ Calculate round-trip time if original timestamp is available. Returns: Optional[float]: RTT in seconds or None if no original timestamp """ if self.original_timestamp: return time.time() - self.original_timestamp return None @dataclass class PrintCommand(TrixyCommand): """ Print command for debugging and testing. This command instructs the recipient to print/log the provided text. Useful for debugging network connectivity and command flow. """ text: str = field(default="") def validate(self) -> bool: """Validate print command.""" if not isinstance(self.text, str): raise ValidationError("PrintCommand text must be a string") return True def get_command_name(self) -> str: """Get command name.""" return f"TRXIPRNT {self.text}" def get_timeout(self) -> float: """Short timeout for print.""" return 5.0 def get_priority(self) -> int: """Low priority for debugging.""" return 80 @dataclass class HelloCommand(TrixyCommand): """ Hello command for debugging and initial handshake. This command can be used for debugging connectivity or as part of an initial handshake sequence. """ message: Optional[str] = field(default=None) def validate(self) -> bool: """Validate hello command.""" return True def get_command_name(self) -> str: """Get command name.""" if self.message: return f"TRXYHELO {self.message}" return "TRXYHELO" def get_timeout(self) -> float: """Short timeout for hello.""" return 10.0 def get_priority(self) -> int: """Normal priority for hello.""" return 50 # Response classes for hard-coded commands @dataclass class PingPongResponse(CommandResponse): """Response for ping/pong commands with latency information.""" ping_id: str = field(default="") rtt_ms: Optional[float] = field(default=None) def validate(self) -> bool: """Validate ping-pong response.""" if not self.ping_id: raise ValidationError("PingPongResponse must have a ping_id") return True def get_response_name(self) -> str: """Get response name.""" return "PingPongResponse" # Utility functions for hard-coded commands def create_heartbeat() -> NoOpCommand: """ Create a heartbeat (no-op) command. Returns: NoOpCommand: Heartbeat command """ return NoOpCommand() def create_ping(ping_id: Optional[str] = None) -> PingCommand: """ Create a ping command with optional custom ping ID. Args: ping_id: Optional ping ID (uses timestamp if None) Returns: PingCommand: Ping command """ cmd = PingCommand() if ping_id: cmd.ping_id = ping_id return cmd def create_pong(ping_id: str, original_timestamp: Optional[float] = None) -> PongCommand: """ Create a pong response to a ping command. Args: ping_id: Ping ID from the original ping command original_timestamp: Timestamp from original ping for RTT calculation Returns: PongCommand: Pong command """ return PongCommand( ping_id=ping_id, original_timestamp=original_timestamp ) def create_debug_print(text: str) -> PrintCommand: """ Create a debug print command. Args: text: Text to print Returns: PrintCommand: Print command """ return PrintCommand(text=text) def create_hello(message: Optional[str] = None) -> HelloCommand: """ Create a hello command. Args: message: Optional hello message Returns: HelloCommand: Hello command """ return HelloCommand(message=message) def is_hardcoded_command(command_text: str) -> bool: """ Check if a command text is a hard-coded command. Args: command_text: Command text to check Returns: bool: True if it's a hard-coded command """ hardcoded_prefixes = ["TRXINOOP", "TRXIPING", "TRXIPONG", "TRXIPRNT", "TRXYHELO"] return any(command_text.startswith(prefix) for prefix in hardcoded_prefixes) def parse_hardcoded_command(command_text: str) -> TrixyCommand: """ Parse a hard-coded command text into a command object. Args: command_text: Command text to parse Returns: TrixyCommand: Parsed command object Raises: ValidationError: If command text is invalid """ command_text = command_text.strip() if command_text == "TRXINOOP": return NoOpCommand() elif command_text.startswith("TRXIPING"): parts = command_text.split(' ', 1) ping_id = parts[1] if len(parts) > 1 else str(int(time.time() * 1000)) return PingCommand(ping_id=ping_id) elif command_text.startswith("TRXIPONG"): parts = command_text.split(' ', 1) ping_id = parts[1] if len(parts) > 1 else "" return PongCommand(ping_id=ping_id) elif command_text.startswith("TRXIPRNT"): parts = command_text.split(' ', 1) text = parts[1] if len(parts) > 1 else "" return PrintCommand(text=text) elif command_text.startswith("TRXYHELO"): parts = command_text.split(' ', 1) message = parts[1] if len(parts) > 1 else None return HelloCommand(message=message) else: raise ValidationError(f"Unknown hard-coded command: {command_text}") def format_hardcoded_command(command: TrixyCommand) -> str: """ Format a command object as hard-coded command text. Args: command: Command object to format Returns: str: Hard-coded command text Raises: ValidationError: If command is not a hard-coded command """ if isinstance(command, NoOpCommand): return "TRXINOOP" elif isinstance(command, PingCommand): return f"TRXIPING {command.ping_id}" elif isinstance(command, PongCommand): return f"TRXIPONG {command.ping_id}" elif isinstance(command, PrintCommand): return f"TRXIPRNT {command.text}" elif isinstance(command, HelloCommand): if command.message: return f"TRXYHELO {command.message}" return "TRXYHELO" else: raise ValidationError(f"Command {type(command)} is not a hard-coded command") # Module exports __all__ = [ "NoOpCommand", "PingCommand", "PongCommand", "PrintCommand", "HelloCommand", "PingPongResponse", "create_heartbeat", "create_ping", "create_pong", "create_debug_print", "create_hello", "is_hardcoded_command", "parse_hardcoded_command", "format_hardcoded_command", ]