MicroPythonOS/internal_filesystem/lib/websocket.py

# websocket.py
# MicroPython WebSocketApp implementation for python-nostr port
# Compatible with websocket-client's WebSocketApp API, using MicroPython aiohttp

import uasyncio as asyncio
import time
import ucollections
import aiohttp
from aiohttp import WSMsgType, ClientWebSocketResponse

# Simplified logging for MicroPython with timestamps
def _log_debug(msg):
    print(f"[DEBUG {time.ticks_ms()}] {msg}")

def _log_error(msg):
    print(f"[ERROR {time.ticks_ms()}] {msg}")

# Simplified ABNF for opcode compatibility
class ABNF:
    OPCODE_TEXT = 1
    OPCODE_BINARY = 2
    OPCODE_CLOSE = 8
    OPCODE_PING = 9
    OPCODE_PONG = 10

# Exceptions
class WebSocketException(Exception):
    pass

class WebSocketConnectionClosedException(WebSocketException):
    pass

class WebSocketTimeoutException(WebSocketException):
    pass

# Queue for callback dispatching
_callback_queue = ucollections.deque((), 100)  # Empty tuple, maxlen=100

def _run_callback(callback, *args):
    """Add callback to queue for execution."""
    try:
        _callback_queue.append((callback, args))
        #_log_debug(f"Queued callback {callback}, args={args}, queue size: {len(_callback_queue)}")
    except IndexError:
        _log_error("ERROR: websocket.py callback queue full, dropping callback")

async def _process_callbacks_async():
    """Process queued callbacks asynchronously."""
    import _thread
    while True:
        #print(f"thread {_thread.get_ident()}: _process_callbacks_async")
        while _callback_queue:
            _log_debug("Processing callbacks queue...")
            try:
                callback, args = _callback_queue.popleft()
                if callback is not None:
                    #_log_debug(f"Executing callback {callback} with {len(args)} args")
                    #for i, arg in enumerate(args):
                    #    _log_debug(f"Arg {i}: {arg}")
                    try:
                        callback(*args)
                    except Exception as e:
                        _log_error(f"Error in callback {callback}: {e}")
                else:
                    _log_debug("Skipping None callback")
            except IndexError:
                _log_debug("Callback queue empty")
                break
        await asyncio.sleep(0.5)  # Yield to other tasks

class WebSocketApp:
    def __init__(
        self,
        url,
        header=None,
        on_open=None,
        on_reconnect=None,
        on_message=None,
        on_error=None,
        on_close=None,
        on_ping=None,
        on_pong=None,
        on_cont_message=None,
        keep_running=True,  # Ignored for compatibility
        get_mask_key=None,
        cookie=None,
        subprotocols=None,
        on_data=None,
        socket=None,
    ):
        self.url = url
        self.header = header if header is not None else {}
        self.cookie = cookie
        self.on_open = on_open
        self.on_reconnect = on_reconnect
        self.on_message = on_message
        self.on_data = on_data
        self.on_error = on_error
        self.on_close = on_close
        self.on_ping = on_ping
        self.on_pong = on_pong
        self.on_cont_message = on_cont_message
        self.get_mask_key = get_mask_key
        self.subprotocols = subprotocols
        self.prepared_socket = socket  # Ignored, not supported
        self.ws = None
        self.session = None
        self.running = False
        self.ping_interval = 0
        self.ping_timeout = None
        self.ping_payload = ""
        self.last_ping_tm = 0
        self.last_pong_tm = 0
        self.has_errored = False
        self._loop = asyncio.get_event_loop()

    def send(self, data, opcode=ABNF.OPCODE_TEXT):
        """Send a message."""
        if not self.ws or not self.running:
            _log_error("Send failed: Connection closed or not running")
            raise WebSocketConnectionClosedException("Connection is already closed.")
        _log_debug(f"Scheduling send: opcode={opcode}, data={str(data)[:20]}...")
        asyncio.create_task(self._send_async(data, opcode))

    def send_text(self, text_data):
        """Send UTF-8 text."""
        self.send(text_data, ABNF.OPCODE_TEXT)

    def send_bytes(self, data):
        """Send binary data."""
        self.send(data, ABNF.OPCODE_BINARY)

    def close(self, **kwargs):
        """Close the WebSocket connection."""
        _log_debug("Close requested")
        self.running = False
        asyncio.create_task(self._close_async())

    async def _close_async(self):
        """Async close implementation."""
        _log_debug("Closing WebSocket connection")
        try:
            if self.ws and not self.ws.ws.closed:
                _log_debug("Sending WebSocket close frame")
                await self.ws.close()
            else:
                _log_debug("WebSocket already closed or not initialized")
            if self.session:
                _log_debug("Closing ClientSession")
                await self.session.__aexit__(None, None, None)
            else:
                _log_debug("No ClientSession to close")
        except Exception as e:
            _log_error(f"Error closing WebSocket: {e}")

    def _start_ping_task(self):
        """Start ping task."""
        if self.ping_interval:
            _log_debug(f"NOT Starting ping task with interval {self.ping_interval}s")
            asyncio.create_task(self._send_ping_async())

    def _stop_ping_thread(self):
        """No-op, ping handled in async task."""
        pass

    async def _send_ping_async(self):
        """Send periodic pings."""
        while self.running and self.ping_interval:
            self.last_ping_tm = time.time()
            try:
                if self.ws and not self.ws.ws.closed:
                    self.ping_payload = "ping"
                    _log_debug(f"Sending ping with payload: {self.ping_payload}")
                    await self.ws.send_bytes(self.ping_payload.encode() if isinstance(self.ping_payload, str) else self.ping_payload)
                else:
                    _log_debug("Skipping ping: WebSocket not connected")
            except Exception as e:
                _log_error(f"Failed to send ping: {e}")
            await asyncio.sleep(self.ping_interval)

    def ready(self):
        """Check if connection is active."""
        status = self.ws is not None and self.running
        _log_debug(f"Connection status: ready={status}")
        return status

    def run_forever(
        self,
        sockopt=None,
        sslopt=None,
        ping_interval=0,
        ping_timeout=None,
        ping_payload="",
        http_proxy_host=None,
        http_proxy_port=None,
        http_no_proxy=None,
        http_proxy_auth=None,
        http_proxy_timeout=None,
        skip_utf8_validation=False,
        host=None,
        origin=None,
        dispatcher=None,
        suppress_origin=False,
        proxy_type=None,
        reconnect=None,
    ):
        """Run the WebSocket event loop in the main thread."""
        _log_debug("Starting run_forever")
        if sockopt or http_proxy_host or http_proxy_port or http_no_proxy or http_proxy_auth or proxy_type:
            raise WebSocketException("Proxy and sockopt not supported in MicroPython")
        if dispatcher:
            raise WebSocketException("Custom dispatcher not supported")
        if ping_timeout is not None and ping_timeout <= 0:
            raise WebSocketException("Ensure ping_timeout > 0")
        if ping_interval is not None and ping_interval < 0:
            raise WebSocketException("Ensure ping_interval >= 0")
        if ping_timeout and ping_interval and ping_interval <= ping_timeout:
            raise WebSocketException("Ensure ping_interval > ping_timeout")

        self.ping_interval = ping_interval
        self.ping_timeout = ping_timeout
        self.ping_payload = ping_payload
        self.running = True

        # Run the event loop in the main thread
        try:
            self._loop.run_until_complete(self._async_main())
        except KeyboardInterrupt:
            _log_debug("run_forever got KeyboardInterrupt")
            self.close()
            return False
        except Exception as e:
            _log_error(f"run_forever's _loop.run_until_complete() got general exception: {e}")
            self.has_errored = True
            self.running = False
            #return True
        _log_debug("run_forever completed")
        return self.has_errored

    async def _async_main(self):
        """Main async loop for WebSocket handling."""
        _log_debug("Starting _async_main")
        reconnect = 0  # Default, as RECONNECT may not be defined
        try:
            from websocket import RECONNECT
            reconnect = RECONNECT
        except ImportError:
            pass
        if reconnect is not None:
            reconnect = reconnect
        _log_debug(f"Reconnect interval set to {reconnect}s")

        # Start callback processing task
        try:
            callback_task = asyncio.create_task(_process_callbacks_async())
            _log_debug("Started callback processing task")
        except Exception as e:
            print(f"websocket.py: create_ask(_process_callbacks_async()) had exception {e}")

        while self.running:
            _log_debug("Main loop iteration: self.running=True")
            try:
                await self._connect_and_run()
            except Exception as e:
                _log_error(f"_async_main got exception: {e}")
                self.has_errored = True
                _run_callback(self.on_error, self, e)
                if not reconnect:
                    _log_debug("No reconnect configured, breaking loop")
                    break
                _log_debug(f"Reconnecting after error in {reconnect}s")
                await asyncio.sleep(reconnect)
                if self.on_reconnect:
                    _run_callback(self.on_reconnect, self)

        # Cleanup
        _log_debug("Initiating cleanup")
        self.running = False
        callback_task.cancel()  # Stop callback task
        try:
            await callback_task
        except asyncio.CancelledError:
            _log_debug("Callback task cancelled")
        await self._close_async()
        _run_callback(self.on_close, self, None, None)
        _log_debug("_async_main completed")

    async def _connect_and_run(self):
        """Connect and handle WebSocket messages."""
        _log_debug(f"Connecting to {self.url}")
        ssl_context = None
        if self.url.startswith("wss://"):
            import ssl
            ssl_context = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT)
            ssl_context.verify_mode = ssl.CERT_NONE
            _log_debug("Using SSL with no certificate verification")

        self.session = aiohttp.ClientSession(headers=self.header)
        async with self.session.ws_connect(self.url, ssl=ssl_context) as ws:
            self.ws = ws
            _log_debug("WebSocket connected, running on_open callback")
            _run_callback(self.on_open, self)
            #self._start_ping_task() this ping task isn't part of the protocol, pings are sent by the server

            async for msg in ws:
                import micropython
                print(f"websocket.py _connect_and_run thread stack used: {micropython.stack_use()}")
                _log_debug(f"websocket.py _connect_and_run received msg: type={msg.type}, length: {len(msg.data)} and data={str(msg.data)[:120]}...")
                if not self.running:
                    _log_debug("Not running, breaking message loop")
                    break

                # Handle ping/pong timeout
                if self.ping_timeout and self.last_ping_tm:
                    if time.time() - self.last_ping_tm > self.ping_timeout:
                        _log_error("Ping/pong timed out")
                        raise WebSocketTimeoutException("ping/pong timed out")

                # Process message
                if msg.type == WSMsgType.TEXT:
                    data = msg.data
                    _run_callback(self.on_data, self, data, ABNF.OPCODE_TEXT, True)
                    _run_callback(self.on_message, self, data)  # Standard websocket-client
                elif msg.type == WSMsgType.BINARY:
                    data = msg.data
                    _run_callback(self.on_data, self, data, ABNF.OPCODE_BINARY, True)
                    _run_callback(self.on_message, self, data)  # Standard websocket-client
                elif msg.type == WSMsgType.ERROR or ws.ws.closed:
                    _log_error("WebSocket error or closed")
                    raise WebSocketConnectionClosedException("WebSocket closed")
                elif msg.type == ABNF.OPCODE_PING:
                    data = msg.data
                    _run_callback(self.on_ping, self, data, ABNF.OPCODE_PING, True)

    async def _send_async(self, data, opcode):
        """Async send implementation."""
        _log_debug(f"Sending: opcode={opcode}, data={str(data)[:700]}")
        try:
            if opcode == ABNF.OPCODE_TEXT:
                await self.ws.send_str(data)
            elif opcode == ABNF.OPCODE_BINARY:
                await self.ws.send_bytes(data)
            else:
                raise WebSocketException(f"Unsupported opcode: {opcode}")
            _log_debug("Send successful")
        except Exception as e:
            _log_error(f"Send failed: {e}")
            _run_callback(self.on_error, self, e)

    def _callback(self, callback, *args):
        """Compatibility wrapper for callback execution."""
        _run_callback(callback, self, *args)

    def _get_close_args(self, close_frame):
        """Extract close code and reason (simplified)."""
        _log_debug("Getting close args (not supported)")
        return [None, None]  # aiohttp doesn't provide close frame details

    def create_dispatcher(self, ping_timeout, dispatcher, is_ssl, handleDisconnect):
        """Not supported."""
        _log_error("Custom dispatcher not supported")
        raise WebSocketException("Custom dispatcher not supported")