Debug and Logging

def enableTrace(
    traceable: bool,
    handler: logging.StreamHandler = logging.StreamHandler(),
    level: str = "DEBUG",
) -> None:
    """
    Enable or disable WebSocket protocol tracing.

    Parameters:
    - traceable: True to enable tracing, False to disable
    - handler: Logging handler for trace output (default: StreamHandler to stdout)
    - level: Logging level for trace messages (default: "DEBUG")

    Note: When enabled, shows detailed frame-level communication,
    connection events, and protocol analysis information.
    """

def isEnabledForError() -> bool:
    """
    Check if ERROR level logging is enabled.

    Returns:
    bool: True if error logging is active
    """

def isEnabledForDebug() -> bool:
    """
    Check if DEBUG level logging is enabled.

    Returns:
    bool: True if debug logging is active
    """

def isEnabledForTrace() -> bool:
    """
    Check if trace logging is enabled.

    Returns:
    bool: True if trace logging is active (via enableTrace)
    """

def dump(title: str, message: str) -> None:
    """
    Output formatted debug dump with title and content.

    Parameters:
    - title: Section title for the dump
    - message: Content to dump

    Note: Only outputs when tracing is enabled
    """

def error(msg: str) -> None:
    """
    Log error message.

    Parameters:
    - msg: Error message text
    """

def warning(msg: str) -> None:
    """
    Log warning message.

    Parameters:
    - msg: Warning message text
    """

def debug(msg: str) -> None:
    """
    Log debug message.

    Parameters:
    - msg: Debug message text
    """

def info(msg: str) -> None:
    """
    Log informational message.

    Parameters:
    - msg: Information message text
    """

def trace(msg: str) -> None:
    """
    Log trace message (only when tracing enabled).

    Parameters:
    - msg: Trace message text

    Note: Only outputs when enableTrace(True) has been called
    """

import websocket

# Enable basic tracing to see WebSocket communication
websocket.enableTrace(True)

# Now all WebSocket operations will show detailed trace output
ws = websocket.create_connection("ws://echo.websocket.events/")
ws.send("Hello, World!")
response = ws.recv()
print(f"Echo: {response}")
ws.close()

# Disable tracing
websocket.enableTrace(False)

import websocket
import logging
import sys

# Create custom handler for trace output
trace_handler = logging.StreamHandler(sys.stderr)
trace_handler.setFormatter(logging.Formatter(
    '[%(asctime)s] %(levelname)s: %(message)s'
))

# Enable tracing with custom handler and level
websocket.enableTrace(True, handler=trace_handler, level="INFO")

# Create connection - will show formatted trace output
ws = websocket.WebSocketApp("ws://echo.websocket.events/",
                          on_open=lambda ws: ws.send("Test message"))
ws.run_forever()

import websocket
import logging

# Set up file logging for WebSocket trace
file_handler = logging.FileHandler('websocket_trace.log')
file_handler.setFormatter(logging.Formatter(
    '%(asctime)s - %(name)s - %(levelname)s - %(message)s'
))

# Enable tracing to file
websocket.enableTrace(True, handler=file_handler, level="DEBUG")

try:
    ws = websocket.create_connection("ws://echo.websocket.events/")
    ws.send("Logged message")
    response = ws.recv()
    ws.close()
    print("Check websocket_trace.log for detailed trace")
except Exception as e:
    print(f"Error: {e}")

import websocket

def smart_logging_example():
    # Check logging status before expensive operations
    if websocket.isEnabledForDebug():
        websocket.debug("Starting WebSocket connection analysis")
        
    if websocket.isEnabledForTrace():
        websocket.trace("Detailed protocol analysis enabled")
        
    try:
        ws = websocket.create_connection("ws://echo.websocket.events/")
        
        if websocket.isEnabledForDebug():
            websocket.debug(f"Connected to {ws.url}")
            websocket.debug(f"Response headers: {ws.headers}")
        
        ws.send("Test")
        response = ws.recv()
        
        if websocket.isEnabledForTrace():
            websocket.trace(f"Received response: {response}")
            
        ws.close()
        
    except Exception as e:
        if websocket.isEnabledForError():
            websocket.error(f"Connection failed: {e}")

# Run with different logging levels
websocket.enableTrace(True, level="DEBUG")
smart_logging_example()

import websocket
import json

def debug_connection_details(ws):
    """Output structured connection information."""
    if websocket.isEnabledForDebug():
        connection_info = {
            "url": getattr(ws, 'url', 'unknown'),
            "status": ws.getstatus() if hasattr(ws, 'getstatus') else None,
            "subprotocol": ws.getsubprotocol() if hasattr(ws, 'getsubprotocol') else None,
            "headers": dict(ws.getheaders()) if hasattr(ws, 'getheaders') and ws.getheaders() else {},
            "ssl": ws.is_ssl() if hasattr(ws, 'is_ssl') else False
        }
        
        websocket.dump("Connection Details", json.dumps(connection_info, indent=2))

# Enable detailed tracing
websocket.enableTrace(True)

ws = websocket.create_connection("ws://echo.websocket.events/")
debug_connection_details(ws)
ws.close()

import websocket
import logging

# Set up application logging
app_logger = logging.getLogger("myapp.websocket")
app_logger.setLevel(logging.DEBUG)

# Create handler for application logs
app_handler = logging.StreamHandler()
app_handler.setFormatter(logging.Formatter(
    '[%(name)s] %(levelname)s: %(message)s'
))
app_logger.addHandler(app_handler)

class LoggingWebSocketApp:
    def __init__(self, url):
        self.url = url
        self.logger = app_logger
        
    def connect(self):
        self.logger.info(f"Connecting to {self.url}")
        
        # Enable websocket library tracing
        websocket.enableTrace(True, level="DEBUG")
        
        ws = websocket.WebSocketApp(
            self.url,
            on_open=self.on_open,
            on_message=self.on_message,
            on_error=self.on_error,
            on_close=self.on_close
        )
        
        ws.run_forever()
        
    def on_open(self, ws):
        self.logger.info("WebSocket connection established")
        
    def on_message(self, ws, message):
        self.logger.debug(f"Received message: {message[:100]}...")
        
    def on_error(self, ws, error):
        self.logger.error(f"WebSocket error: {error}")
        
    def on_close(self, ws, close_status_code, close_msg):
        self.logger.info(f"Connection closed: {close_status_code} - {close_msg}")

# Usage
app = LoggingWebSocketApp("ws://echo.websocket.events/")
app.connect()

import websocket
import time

def measure_logging_overhead():
    """Measure performance impact of different logging levels."""
    
    test_scenarios = [
        {"name": "No logging", "trace": False},
        {"name": "Error only", "trace": False, "level": "ERROR"},
        {"name": "Debug enabled", "trace": False, "level": "DEBUG"},
        {"name": "Full tracing", "trace": True, "level": "DEBUG"}
    ]
    
    for scenario in test_scenarios:
        print(f"\nTesting: {scenario['name']}")
        
        # Configure logging
        if scenario.get("trace"):
            websocket.enableTrace(True, level=scenario.get("level", "DEBUG"))
        else:
            websocket.enableTrace(False)
            
        # Measure connection and message exchange time
        start_time = time.time()
        
        try:
            ws = websocket.create_connection("ws://echo.websocket.events/")
            
            # Send multiple messages
            for i in range(10):
                ws.send(f"Test message {i}")
                response = ws.recv()
                
            ws.close()
            
            elapsed = time.time() - start_time
            print(f"  Completed in {elapsed:.3f} seconds")
            
        except Exception as e:
            print(f"  Error: {e}")
    
    # Reset to default
    websocket.enableTrace(False)

measure_logging_overhead()

import websocket

def analyze_websocket_frames():
    """Demonstrate custom debug dump usage."""
    websocket.enableTrace(True)
    
    ws = websocket.WebSocket()
    ws.connect("ws://echo.websocket.events/")
    
    # Send test data
    test_messages = [
        "Short message",
        "A" * 1000,  # Long message
        b"\x01\x02\x03\x04",  # Binary data
    ]
    
    for i, msg in enumerate(test_messages):
        websocket.dump(f"Test Message {i+1}", f"Sending: {repr(msg)}")
        
        if isinstance(msg, bytes):
            ws.send_bytes(msg)
        else:
            ws.send(msg)
            
        response = ws.recv()
        websocket.dump(f"Response {i+1}", f"Received: {repr(response)}")
    
    ws.close()

analyze_websocket_frames()

import websocket
import logging
import sys

class WebSocketTraceFilter(logging.Filter):
    """Custom filter for WebSocket trace messages."""
    
    def __init__(self, include_frames=True, include_handshake=True):
        super().__init__()
        self.include_frames = include_frames
        self.include_handshake = include_handshake
    
    def filter(self, record):
        message = record.getMessage()
        
        # Filter frame-level messages
        if not self.include_frames and ("++Sent raw:" in message or "++Rcv raw:" in message):
            return False
            
        # Filter handshake messages  
        if not self.include_handshake and "handshake" in message.lower():
            return False
            
        return True

# Create filtered handler
handler = logging.StreamHandler(sys.stdout)
handler.addFilter(WebSocketTraceFilter(include_frames=False, include_handshake=True))

# Enable tracing with filter
websocket.enableTrace(True, handler=handler, level="DEBUG")

# This will show handshake details but not frame-by-frame data
ws = websocket.create_connection("ws://echo.websocket.events/")
ws.send("Hello")
response = ws.recv()
ws.close()