Trixy3/trixy_core/network/cmd/hardcoded.py

"""
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 <string> - 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",
]