diff --git a/python-ecosys/debugpy/README.md b/python-ecosys/debugpy/README.md new file mode 100644 index 000000000..70859b974 --- /dev/null +++ b/python-ecosys/debugpy/README.md @@ -0,0 +1,172 @@ +# MicroPython debugpy + +A minimal implementation of debugpy for MicroPython, enabling remote debugging +such as VS Code debugging support. + +## Features + +- Debug Adapter Protocol (DAP) support for VS Code integration +- Basic debugging operations: + - Breakpoints + - Step over/into/out + - Stack trace inspection + - Variable inspection (globals, locals generally not supported) + - Expression evaluation + - Pause/continue execution + +## Requirements + +- MicroPython with `sys.settrace` support (enabled with `MICROPY_PY_SYS_SETTRACE`) +- Socket support for network communication +- JSON support for DAP message parsing + +## Usage + +### Basic Usage + +```python +import debugpy + +# Start listening for debugger connections +host, port = debugpy.listen() # Default: 127.0.0.1:5678 +print(f"Debugger listening on {host}:{port}") + +# Enable debugging for current thread +debugpy.debug_this_thread() + +# Your code here... +def my_function(): + x = 10 + y = 20 + result = x + y # Set breakpoint here in VS Code + return result + +result = my_function() +print(f"Result: {result}") + +# Manual breakpoint +debugpy.breakpoint() +``` + +### VS Code Configuration + +Create a `.vscode/launch.json` file in your project: + +```json +{ + "version": "0.2.0", + "configurations": [ + { + "name": "Attach to MicroPython", + "type": "python", + "request": "attach", + "connect": { + "host": "127.0.0.1", + "port": 5678 + }, + "pathMappings": [ + { + "localRoot": "${workspaceFolder}", + "remoteRoot": "." + } + ], + "justMyCode": false + } + ] +} +``` + +### Testing + +1. Build the MicroPython Unix coverage port: + ```bash + cd ports/unix + make CFLAGS_EXTRA="-DMICROPY_PY_SYS_SETTRACE=1" + ``` + +2. Run the test script: + ```bash + cd lib/micropython-lib/python-ecosys/debugpy + ../../../../ports/unix/build-coverage/micropython test_debugpy.py + ``` + +3. In VS Code, open the debugpy folder and press F5 to attach the debugger + +4. Set breakpoints in the test script and observe debugging functionality + +## API Reference + +### `debugpy.listen(port=5678, host="127.0.0.1")` + +Start listening for debugger connections. + +**Parameters:** +- `port`: Port number to listen on (default: 5678) +- `host`: Host address to bind to (default: "127.0.0.1") + +**Returns:** Tuple of (host, port) actually used + +### `debugpy.debug_this_thread()` + +Enable debugging for the current thread by installing the trace function. + +### `debugpy.breakpoint()` + +Trigger a manual breakpoint that will pause execution if a debugger is attached. + +### `debugpy.wait_for_client()` + +Wait for the debugger client to connect and initialize. + +### `debugpy.is_client_connected()` + +Check if a debugger client is currently connected. + +**Returns:** Boolean indicating connection status + +### `debugpy.disconnect()` + +Disconnect from the debugger client and clean up resources. + +## Architecture + +The implementation consists of several key components: + +1. **Public API** (`public_api.py`): Main entry points for users +2. **Debug Session** (`server/debug_session.py`): Handles DAP protocol communication +3. **PDB Adapter** (`server/pdb_adapter.py`): Bridges DAP and MicroPython's trace system +4. **Messaging** (`common/messaging.py`): JSON message handling for DAP +5. **Constants** (`common/constants.py`): DAP protocol constants + +## Limitations + +This is a minimal implementation with the following limitations: + +- Single-threaded debugging only +- No conditional breakpoints +- No function breakpoints +- Limited variable inspection (no nested object expansion) +- No step back functionality +- No hot code reloading +- Simplified stepping implementation + +## Compatibility + +Tested with: +- MicroPython Unix port +- VS Code with Python/debugpy extension +- CPython 3.x (for comparison) + +## Contributing + +This implementation provides a foundation for MicroPython debugging. Contributions are welcome to add: + +- Conditional breakpoint support +- Better variable inspection +- Multi-threading support +- Performance optimizations +- Additional DAP features + +## License + +MIT License - see the MicroPython project license for details. diff --git a/python-ecosys/debugpy/dap_monitor.py b/python-ecosys/debugpy/dap_monitor.py new file mode 100644 index 000000000..85455c9a6 --- /dev/null +++ b/python-ecosys/debugpy/dap_monitor.py @@ -0,0 +1,199 @@ +#!/usr/bin/env python3 +"""DAP protocol monitor - sits between VS Code and MicroPython debugpy.""" + +import socket +import threading +import json +import time +import sys +import argparse + + +class DAPMonitor: + def __init__(self, listen_port=5679, target_host="127.0.0.1", target_port=5678): + self.disconnect = False + self.listen_port = listen_port + self.target_host = target_host + self.target_port = target_port + self.client_sock = None + self.server_sock = None + + def start(self): + """Start the DAP monitor proxy.""" + print(f"DAP Monitor starting on port {self.listen_port}") + print(f"Will forward to {self.target_host}:{self.target_port}") + print("Start MicroPython debugpy server first, then connect VS Code to port 5679") + + # Create listening socket + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + listener.bind(("127.0.0.1", self.listen_port)) + listener.listen(1) + + print(f"Listening for VS Code connection on port {self.listen_port}...") + + try: + # Wait for VS Code to connect + self.client_sock, client_addr = listener.accept() + print(f"VS Code connected from {client_addr}") + + # Connect to MicroPython debugpy server + self.server_sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.server_sock.connect((self.target_host, self.target_port)) + print(f"Connected to MicroPython debugpy at {self.target_host}:{self.target_port}") + + # Start forwarding threads + threading.Thread(target=self.forward_client_to_server, daemon=True).start() + threading.Thread(target=self.forward_server_to_client, daemon=True).start() + + print("DAP Monitor active - press Ctrl+C to stop") + while not self.disconnect: + time.sleep(1) + + except KeyboardInterrupt: + print("\nStopping DAP Monitor...") + except Exception as e: + print(f"Error: {e}") + finally: + self.cleanup() + + def forward_client_to_server(self): + """Forward messages from VS Code client to MicroPython server.""" + try: + while True: + data = self.receive_dap_message(self.client_sock, "VS Code") + if data is None: + break + self.send_raw_data(self.server_sock, data) + except Exception as e: + print(f"Client->Server forwarding error: {e}") + + def forward_server_to_client(self): + """Forward messages from MicroPython server to VS Code client.""" + try: + while True: + data = self.receive_dap_message(self.server_sock, "MicroPython") + if data is None: + break + self.send_raw_data(self.client_sock, data) + except Exception as e: + print(f"Server->Client forwarding error: {e}") + + def receive_dap_message(self, sock, source): + """Receive and log a DAP message.""" + try: + # Read headers + header = b"" + while b"\r\n\r\n" not in header: + byte = sock.recv(1) + if not byte: + return None + header += byte + + # Parse content length + header_str = header.decode("utf-8") + content_length = 0 + for line in header_str.split("\r\n"): + if line.startswith("Content-Length:"): + content_length = int(line.split(":", 1)[1].strip()) + break + + if content_length == 0: + return None + + # Read content + content = b"" + while len(content) < content_length: + chunk = sock.recv(content_length - len(content)) + if not chunk: + return None + content += chunk + + # Parse and Log the message + message = self.parse_dap(source, content) + self.log_dap_message(source, message) + # Check for disconnect command + if message: + if "disconnect" == message.get("command", message.get("event", "unknown")): + print(f"\n[{source}] Disconnect command received, stopping monitor.") + self.disconnect = True + return header + content + except Exception as e: + print(f"Error receiving from {source}: {e}") + return None + + def parse_dap(self, source, content): + """Parse DAP message and log it.""" + try: + message = json.loads(content.decode("utf-8")) + return message + except json.JSONDecodeError: + print(f"\n[{source}] Invalid JSON: {content}") + return None + + def log_dap_message(self, source, message): + """Log DAP message details.""" + msg_type = message.get("type", "unknown") + command = message.get("command", message.get("event", "unknown")) + seq = message.get("seq", 0) + + print(f"\n[{source}] {msg_type.upper()}: {command} (seq={seq})") + + if msg_type == "request": + args = message.get("arguments", {}) + if args: + print(f" Arguments: {json.dumps(args, indent=2)}") + elif msg_type == "response": + success = message.get("success", False) + req_seq = message.get("request_seq", 0) + print(f" Success: {success}, Request Seq: {req_seq}") + body = message.get("body") + if body: + print(f" Body: {json.dumps(body, indent=2)}") + msg = message.get("message") + if msg: + print(f" Message: {msg}") + elif msg_type == "event": + body = message.get("body", {}) + if body: + print(f" Body: {json.dumps(body, indent=2)}") + + def send_raw_data(self, sock, data): + """Send raw data to socket.""" + try: + sock.send(data) + except Exception as e: + print(f"Error sending data: {e}") + + def cleanup(self): + """Clean up sockets.""" + if self.client_sock: + self.client_sock.close() + if self.server_sock: + self.server_sock.close() + + +if __name__ == "__main__": + parser = argparse.ArgumentParser(description="DAP protocol monitor proxy") + parser.add_argument( + "--target-host", + "--th", + default="127.0.0.1", + help="Target debugpy host (default: 127.0.0.1)", + ) + parser.add_argument( + "--target-port", "--tp", type=int, default=5678, help="Target debugpy port (default: 5678)" + ) + parser.add_argument( + "--listen-port", + "--lp", + type=int, + default=5679, + help="Port to listen for VS Code (default: 5679)", + ) + args = parser.parse_args() + + monitor = DAPMonitor( + listen_port=args.listen_port, target_host=args.target_host, target_port=args.target_port + ) + monitor.start() diff --git a/python-ecosys/debugpy/debugpy/__init__.py b/python-ecosys/debugpy/debugpy/__init__.py new file mode 100644 index 000000000..cce6cb870 --- /dev/null +++ b/python-ecosys/debugpy/debugpy/__init__.py @@ -0,0 +1,31 @@ +"""MicroPython debugpy implementation. + +A minimal port of debugpy for MicroPython to enable VS Code debugging support. +This implementation focuses on the core DAP (Debug Adapter Protocol) functionality +needed for basic debugging operations like breakpoints, stepping, and variable inspection. +""" + +__version__ = "0.1.0" + +from .public_api import ( + breakpoint, + debug_this_thread, + disconnect, + get_capabilities, + is_client_connected, + listen, + wait_for_client, +) +from .common.constants import DEFAULT_HOST, DEFAULT_PORT + +__all__ = [ + "DEFAULT_HOST", + "DEFAULT_PORT", + "breakpoint", + "debug_this_thread", + "disconnect", + "get_capabilities", + "is_client_connected", + "listen", + "wait_for_client", +] diff --git a/python-ecosys/debugpy/debugpy/common/__init__.py b/python-ecosys/debugpy/debugpy/common/__init__.py new file mode 100644 index 000000000..c53632010 --- /dev/null +++ b/python-ecosys/debugpy/debugpy/common/__init__.py @@ -0,0 +1 @@ +# Common utilities and constants for debugpy diff --git a/python-ecosys/debugpy/debugpy/common/constants.py b/python-ecosys/debugpy/debugpy/common/constants.py new file mode 100644 index 000000000..5d9a52204 --- /dev/null +++ b/python-ecosys/debugpy/debugpy/common/constants.py @@ -0,0 +1,74 @@ +"""Constants used throughout debugpy.""" + +from micropython import const + +# Default networking settings +DEFAULT_HOST = "127.0.0.1" +DEFAULT_PORT = 5678 + +# DAP message types +MSG_TYPE_REQUEST = const("request") +MSG_TYPE_RESPONSE = const("response") +MSG_TYPE_EVENT = const("event") + +# DAP events +EVENT_INITIALIZED = const("initialized") +EVENT_STOPPED = const("stopped") +EVENT_CONTINUED = const("continued") +EVENT_THREAD = const("thread") +EVENT_BREAKPOINT = const("breakpoint") +EVENT_OUTPUT = const("output") +EVENT_TERMINATED = const("terminated") +EVENT_EXITED = const("exited") + +# DAP commands +CMD_INITIALIZE = const("initialize") +CMD_LAUNCH = const("launch") +CMD_ATTACH = const("attach") +CMD_SET_BREAKPOINTS = const("setBreakpoints") +CMD_CONTINUE = const("continue") +CMD_NEXT = const("next") +CMD_STEP_IN = const("stepIn") +CMD_STEP_OUT = const("stepOut") +CMD_PAUSE = const("pause") +CMD_STACK_TRACE = const("stackTrace") +CMD_SCOPES = const("scopes") +CMD_VARIABLES = const("variables") +CMD_SET_VARIABLE = const("setVariable") +CMD_EVALUATE = const("evaluate") +CMD_DISCONNECT = const("disconnect") +CMD_CONFIGURATION_DONE = const("configurationDone") +CMD_THREADS = const("threads") +CMD_SOURCE = const("source") + +# Stop reasons +STOP_REASON_STEP = const("step") +STOP_REASON_BREAKPOINT = const("breakpoint") +STOP_REASON_EXCEPTION = const("exception") +STOP_REASON_PAUSE = const("pause") +STOP_REASON_ENTRY = const("entry") + +# Thread reasons +THREAD_REASON_STARTED = const("started") +THREAD_REASON_EXITED = const("exited") + +# Trace events +TRACE_CALL = const("call") +TRACE_LINE = const("line") +TRACE_RETURN = const("return") +TRACE_EXCEPTION = const("exception") + +# Step modes +STEP_INTO = const("into") +STEP_OVER = const("over") +STEP_OUT = const("out") + + +# Scope types +SCOPE_LOCALS = const("locals") +SCOPE_GLOBALS = const("globals") + +# Bounded wait for the DAP client to send configurationDone (seconds). There is +# no server thread, so a hang here would spin forever with no diagnostic; a +# timeout with a clear message replaces a silent guessed delay. +WAIT_FOR_CLIENT_TIMEOUT_S = const(30) diff --git a/python-ecosys/debugpy/debugpy/common/messaging.py b/python-ecosys/debugpy/debugpy/common/messaging.py new file mode 100644 index 000000000..7d704f6a8 --- /dev/null +++ b/python-ecosys/debugpy/debugpy/common/messaging.py @@ -0,0 +1,156 @@ +"""JSON message handling for DAP protocol.""" + +import json +from .constants import MSG_TYPE_REQUEST, MSG_TYPE_RESPONSE, MSG_TYPE_EVENT + + +class JsonMessageChannel: + """Handles JSON message communication over a socket using DAP format.""" + + def __init__(self, sock, debug_callback=None): + self.sock = sock + self.seq = 0 + self.closed = False + self._recv_buffer = b"" + self._debug_print = debug_callback or (lambda x: None) # Default to no-op + + def send_message(self, msg_type, command=None, **kwargs): + """Send a DAP message.""" + if self.closed: + return + + self.seq += 1 + message = { + "seq": self.seq, + "type": msg_type, + } + + if command: + if msg_type == MSG_TYPE_REQUEST: + message["command"] = command + if kwargs: + message["arguments"] = kwargs + elif msg_type == MSG_TYPE_RESPONSE: + message["command"] = command + message["request_seq"] = kwargs.get("request_seq", 0) + message["success"] = kwargs.get("success", True) + if "body" in kwargs: + message["body"] = kwargs["body"] + if "message" in kwargs: + message["message"] = kwargs["message"] + elif msg_type == MSG_TYPE_EVENT: + message["event"] = command + if kwargs: + message["body"] = kwargs + + json_str = json.dumps(message) + content = json_str.encode("utf-8") + header = f"Content-Length: {len(content)}\r\n\r\n".encode("utf-8") + + try: + self.sock.send(header + content) + except OSError: + self.closed = True + + def send_request(self, command, **kwargs): + """Send a request message.""" + self.send_message(MSG_TYPE_REQUEST, command, **kwargs) + + def send_response(self, command, request_seq, success=True, body=None, message=None): + """Send a response message.""" + kwargs = {"request_seq": request_seq, "success": success} + if body is not None: + kwargs["body"] = body + if message is not None: + kwargs["message"] = message + + self._debug_print( + f"[DAP] SEND: response {command} (req_seq={request_seq}, success={success})" + ) + if body: + self._debug_print(f"[DAP] body: {body}") + if message: + self._debug_print(f"[DAP] message: {message}") + + self.send_message(MSG_TYPE_RESPONSE, command, **kwargs) + + def send_event(self, event, **kwargs): + """Send an event message.""" + self._debug_print(f"[DAP] SEND: event {event}") + if kwargs: + self._debug_print(f"[DAP] body: {kwargs}") + self.send_message(MSG_TYPE_EVENT, event, **kwargs) + + def recv_message(self): + """Receive a DAP message, or None if a full one isn't available yet. + + Called repeatedly against a socket with a short recv timeout (see + `DebugSession.process_pending_messages`), so a single message's + header and body routinely arrive across several calls. Everything + read so far - including an already-located header - is kept in + `self._recv_buffer` verbatim until the *entire* message (header + + `Content-Length` body bytes) is available, and only then is it + parsed and sliced off. Parsing the header again on each call is + cheap and avoids having to separately persist "header already + parsed, N body bytes still outstanding" state between calls: a + prior version stripped the header out of the buffer as soon as it + was found, which discarded that state and desynchronised framing + for the rest of the connection whenever the body arrived in a + later read than the header. + """ + if self.closed: + return None + + # Non-blocking top-up: pull in whatever is available right now + # without blocking if there's nothing new yet. + try: + data = self.sock.recv(4096) + if not data: + # A truly empty read (as opposed to EAGAIN/EWOULDBLOCK, + # handled below) means the peer closed the connection. + self.closed = True + return None + self._recv_buffer += data + except OSError as e: + if not (hasattr(e, "errno") and e.errno in (11, 35)): # EAGAIN, EWOULDBLOCK + self.closed = True + return None + # No new data available right now - fall through and try to + # parse a complete message out of whatever is already buffered. + + recv_buffer = self._recv_buffer + header_end = recv_buffer.find(b"\r\n\r\n") + if header_end < 0: + return None # Header not fully received yet. + + header_str = recv_buffer[:header_end].decode("utf-8") + content_length = 0 + for line in header_str.split("\r\n"): + if line.startswith("Content-Length:"): + content_length = int(line.split(":", 1)[1].strip()) + break + + body_start = header_end + 4 + if len(recv_buffer) < body_start + content_length: + return None # Body not fully received yet. + + body = recv_buffer[body_start : body_start + content_length] + self._recv_buffer = recv_buffer[body_start + content_length :] + + try: + message = json.loads(body.decode("utf-8")) + self._debug_print( + f"[DAP] Successfully received message: {message.get('type')} {message.get('command', message.get('event', 'unknown'))}" + ) + return message + except (ValueError, UnicodeDecodeError) as e: + print(f"[DAP] JSON parse error: {e}") + return None + + def close(self): + """Close the channel.""" + self.closed = True + try: + self.sock.close() + except OSError: + pass diff --git a/python-ecosys/debugpy/debugpy/public_api.py b/python-ecosys/debugpy/debugpy/public_api.py new file mode 100644 index 000000000..6379d4523 --- /dev/null +++ b/python-ecosys/debugpy/debugpy/public_api.py @@ -0,0 +1,218 @@ +"""Public API for debugpy.""" + +import socket +import struct +import sys +from .common.constants import DEFAULT_HOST, DEFAULT_PORT +from .server.debug_session import DebugSession + +_debug_session = None +# Bound-but-not-yet-accepted socket, held between listen() and the accept that +# wait_for_client() performs. +_listener = None + + +def listen(port=DEFAULT_PORT, host=DEFAULT_HOST): + """Bind a listening socket and return the address it is bound to. + + Returns as soon as the socket is bound, WITHOUT waiting for a client, so + the caller can publish the endpoint that a client then connects to. The + accept and the `initialize` handshake happen in `wait_for_client()`. This + matches CPython debugpy, where `listen()` reports the endpoint and + `wait_for_client()` blocks. + + Args: + port: Port number to listen on, or 0 to let the system choose + (default: 5678) + host: Host address to bind to (default: "127.0.0.1") + + Returns: + (host, port) tuple of the actual bound address + """ + global _listener + + if _listener is not None or _debug_session is not None: + raise RuntimeError("Already listening for debugger") + + # Create listening socket + listener = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + try: + listener.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + except: + pass # Not supported in MicroPython + + # Use getaddrinfo for MicroPython compatibility + addr_info = socket.getaddrinfo(host, port) + addr = addr_info[0][-1] # Get the sockaddr + listener.bind(addr) + listener.listen(1) + + # Resolve the actual bound port (needed when the caller asked for port 0 / + # auto). Not every MicroPython port implements getsockname(). + requested_port = port + try: + bound_addr = listener.getsockname() + if isinstance(bound_addr, (tuple, list)) and len(bound_addr) >= 2: + port = bound_addr[1] + except Exception: + pass + if requested_port == 0 and port == 0: + # Callers act on the endpoint this returns, so reporting a port + # nothing can connect to would be worse than refusing: substituting + # DEFAULT_PORT here would advertise an address the socket is not + # bound to. Ask for an explicit port on a target whose getsockname() + # cannot report the OS-assigned one. + listener.close() + raise OSError( + "port=0 needs getsockname() to report the assigned port, which " + "this target does not implement; pass an explicit port" + ) + + _listener = listener + print(f"Debugpy listening on {host}:{port}") + return (host, port) + + +def _accept_and_initialize(): + """Accept the pending connection and handle the client's `initialize`. + + Split out of `listen()` so the endpoint can be published before a client + exists. Returns True once a session is ready. + """ + global _debug_session, _listener + + if _listener is None: + print("[DAP] no listening socket; call listen() first") + return False + + listener, _listener = _listener, None + client_sock = None + try: + client_sock, client_addr = listener.accept() + print(f"Debugger connected from {format_client_addr(client_addr)}") + + _debug_session = DebugSession(client_sock) + + print("[DAP] Waiting for initialize request...") + init_message = _debug_session.channel.recv_message() + if init_message and init_message.get("command") == "initialize": + _debug_session._handle_message(init_message) + print("[DAP] Initialize request handled - returning control immediately") + else: + print(f"[DAP] Warning: Expected initialize, got {init_message}") + + # Set socket to non-blocking for subsequent message processing + _debug_session.channel.sock.settimeout(0.001) + + print("[DAP] Debug session ready - all other messages will be handled in trace function") + return True + + except Exception as e: + print(f"[DAP] Connection error: {e}") + if client_sock: + client_sock.close() + _debug_session = None + return False + finally: + # The accepted client socket is independent of the listener; closing + # the listener does not affect it. This is a single-connection server, + # so stop listening once the client is accepted. + listener.close() + + +def format_client_addr(client_addr): + """Format client address using socket module methods""" + if isinstance(client_addr, (tuple, list)): + # Already in (ip, port) format + return f"{client_addr[0]}:{client_addr[1]}" + elif isinstance(client_addr, bytes) and len(client_addr) >= 8: + # Extract port (bytes 2-4, network byte order) + port = struct.unpack("!H", client_addr[2:4])[0] + # Extract IP address (bytes 4-8) using inet_ntoa + ip_packed = client_addr[4:8] + try: + # inet_ntoa expects 4-byte string in network byte order + ip_addr = socket.inet_ntoa(ip_packed) + return f"{ip_addr}:{port}" + except: + # Fallback if inet_ntoa not available (MicroPython) + ip_addr = ".".join(str(b) for b in ip_packed) + return f"{ip_addr}:{port}" + else: + return str(client_addr) + + +def wait_for_client(timeout_s=None): + """Block until a client has attached and finished configuring. + + Accepts the connection and handles `initialize` (both deferred by + `listen()` so the endpoint can be published first), then waits for + `configurationDone`. Breakpoints the client sets before then are honoured + because this drains the socket the whole time it waits. Returns True once + configurationDone arrives, False after a bounded timeout (logged, not + silent) or if nothing is listening. + """ + global _debug_session + if _debug_session is None and not _accept_and_initialize(): + print("[DAP] wait_for_client: nothing is listening, nothing to wait for") + return False + if timeout_s is None: + return _debug_session.wait_for_client() + return _debug_session.wait_for_client(timeout_s) + + +def get_capabilities(): + """Return the firmware capability dict (settrace/save_names/set_local/f_back). + + Uses the active session's probe result if a session exists, otherwise + probes directly. Values always come from probing the running + interpreter, never from a build/variant name. + """ + global _debug_session + if _debug_session is not None: + return _debug_session.capabilities + return DebugSession.probe_capabilities() + + +def breakpoint(): + """Trigger a breakpoint in the debugger.""" + global _debug_session + if _debug_session: + _debug_session.trigger_breakpoint() + else: + # Fallback to built-in breakpoint if available + if hasattr(__builtins__, "breakpoint"): + __builtins__.breakpoint() + + +def debug_this_thread(): + """Enable debugging for the current thread.""" + global _debug_session + if _debug_session: + _debug_session.debug_this_thread() + else: + # Install trace function even if no session yet + if hasattr(sys, "settrace"): + sys.settrace(_default_trace_func) + else: + raise RuntimeError("MICROPY_PY_SYS_SETTRACE required") + + +def _default_trace_func(frame, event, arg): + """Default trace function when no debug session is active.""" + # Just return None to continue execution + return None + + +def is_client_connected(): + """Check if a debugger client is connected.""" + global _debug_session + return _debug_session is not None and _debug_session.is_connected() + + +def disconnect(): + """Disconnect from the debugger client.""" + global _debug_session + if _debug_session: + _debug_session.disconnect() + _debug_session = None diff --git a/python-ecosys/debugpy/debugpy/server/__init__.py b/python-ecosys/debugpy/debugpy/server/__init__.py new file mode 100644 index 000000000..1ab7a0ff5 --- /dev/null +++ b/python-ecosys/debugpy/debugpy/server/__init__.py @@ -0,0 +1 @@ +# Debug server components diff --git a/python-ecosys/debugpy/debugpy/server/debug_session.py b/python-ecosys/debugpy/debugpy/server/debug_session.py new file mode 100644 index 000000000..8a594a246 --- /dev/null +++ b/python-ecosys/debugpy/debugpy/server/debug_session.py @@ -0,0 +1,619 @@ +"""Main debug session handling DAP protocol communication.""" + +import sys +import time + +from ..common.constants import ( + CMD_ATTACH, + CMD_CONFIGURATION_DONE, + CMD_CONTINUE, + CMD_DISCONNECT, + CMD_EVALUATE, + CMD_INITIALIZE, + CMD_LAUNCH, + CMD_NEXT, + CMD_PAUSE, + CMD_SCOPES, + CMD_SET_BREAKPOINTS, + CMD_SET_VARIABLE, + CMD_SOURCE, + CMD_STACK_TRACE, + CMD_STEP_IN, + CMD_STEP_OUT, + CMD_THREADS, + CMD_VARIABLES, + EVENT_CONTINUED, + EVENT_INITIALIZED, + EVENT_STOPPED, + EVENT_TERMINATED, + STOP_REASON_BREAKPOINT, + STOP_REASON_PAUSE, + STOP_REASON_STEP, + TRACE_CALL, + TRACE_EXCEPTION, + TRACE_LINE, + TRACE_RETURN, + WAIT_FOR_CLIENT_TIMEOUT_S, +) +from ..common.messaging import JsonMessageChannel +from .pdb_adapter import PdbAdapter + + +def _is_placeholder_local_name(name): + """True if `name` is a positional `local_N` placeholder, not a real name. + + Without MICROPY_PY_SYS_SETTRACE_SAVE_NAMES, frame.f_locals synthesizes + names as `local_1`, `local_2`, ... (see py/profile.c). This is the only + reliable signal that separates the two cases at runtime. + """ + if not name.startswith("local_"): + return False + return name[len("local_") :].isdigit() + + +class DebugSession: + """Manages a debugging session with a DAP client.""" + + def __init__(self, client_socket): + self.debug_logging = False # Initialize first + self.channel = JsonMessageChannel(client_socket, self._debug_print) + self.pdb = PdbAdapter() + self.pdb._debug_session = self # Allow PDB to process messages during wait # type: ignore[assignment] + self.initialized = False + self.connected = True + self.thread_id = 1 # Simple single-thread model + self.stepping = False + self.paused = False + self.configuration_done = False + self._pumping = False + # Probed once at session start; never inferred from a build/variant name. + self.capabilities = self.probe_capabilities() + self.pdb.capabilities = self.capabilities + + def _debug_print(self, message): + """Print debug message only if debug logging is enabled.""" + if self.debug_logging: + print(message) + + @property + def _baremetal(self) -> bool: + return sys.platform not in ("linux") # to be expanded + + @staticmethod + def probe_capabilities(): + """Probe what the running firmware actually supports. + + Returns a dict with at least `settrace`, `save_names`, `set_local` and + `f_back`, each derived by exercising the real interpreter - never by + reading a build/variant name, which does not reliably reflect what a + given firmware image supports (see BACKGROUND.md). Safe to call on + both the unix port and bare-metal builds; never raises. + """ + caps = { + "settrace": hasattr(sys, "settrace"), + "f_back": False, + "save_names": False, + "set_local": False, + } + if not caps["settrace"]: + return caps + + try: + frame = sys._getframe() + except Exception: + return caps + + try: + caps["f_back"] = hasattr(frame, "f_back") + except Exception: + pass + + try: + caps["set_local"] = hasattr(frame, "_set_local") + except Exception: + pass + + try: + local_names = list(frame.f_locals.keys()) + # An empty locals dict (e.g. probing from module scope) proves + # nothing either way; only trust the signal when there is at + # least one local name to inspect for the placeholder pattern. + caps["save_names"] = bool(local_names) and not any( + _is_placeholder_local_name(n) for n in local_names + ) + except Exception: + pass + + return caps + + def start(self): + """Start the debug session message loop.""" + try: + while self.connected and not self.channel.closed: + message = self.channel.recv_message() + if message is None: + break + + self._handle_message(message) + + except Exception as e: + print(f"Debug session error: {e}") + finally: + self.disconnect() + + def initialize_connection(self): + """Initialize the connection - handle just the essential initial messages then return.""" + # Note: debug_logging not available yet during init, so we always show these messages + print("[DAP] Processing initial DAP messages...") + + try: + # Process initial messages quickly and return control to main thread + # We'll handle ongoing messages in the trace function + attached = False + message_count = 0 + max_init_messages = 6 # Just handle the first few essential messages + + while message_count < max_init_messages and not attached: + try: + # Short timeout - don't block the main thread for long + self.channel.sock.settimeout(1.0) + message = self.channel.recv_message() + if message is None: + print("[DAP] No more messages in initial batch") + break + + print(f"[DAP] Initial message #{message_count + 1}: {message.get('command')}") + self._handle_message(message) + message_count += 1 + + # Just wait for attach, then we can return control + if message.get("command") == "attach": + attached = True + print("[DAP] ✅ Attach received - returning control to main thread") + break + + except Exception as e: + print(f"[DAP] Exception in initial processing: {e}") + break + finally: + self.channel.sock.settimeout(None) + + # After attach, continue processing a few more messages quickly + if attached: + self._debug_print("[DAP] Processing remaining setup messages...") + additional_count = 0 + while additional_count < 4: # Just a few more + try: + self.channel.sock.settimeout(0.5) # Short timeout + message = self.channel.recv_message() + if message is None: + break + self._debug_print(f"[DAP] Setup message: {message.get('command')}") + self._handle_message(message) + additional_count += 1 + except: + break + finally: + self.channel.sock.settimeout(None) + + print("[DAP] Initial setup complete - main thread can continue") + + except Exception as e: + print(f"[DAP] Initialization error: {e}") + + def process_pending_messages(self): + """Process any pending DAP messages without blocking. + + Not re-entered: the trace function calls this on entry to every new + frame, so handling a message here can call it again. A nested call + must not touch the socket timeout, because its `finally` would put the + socket back into blocking mode underneath the outer loop, whose next + recv() then waits for a message the client will not send until it has + seen an event this loop is what produces. MicroPython sockets have no + gettimeout(), so the nesting is tracked rather than the timeout saved. + """ + if self._pumping: + return + self._pumping = True + try: + # Set socket to non-blocking mode for message processing + self.channel.sock.settimeout(0.001) # Very short timeout + + while True: + message = self.channel.recv_message() + if message is None: + break + self._handle_message(message) + + except Exception: + # No messages available or socket error + pass + finally: + # Reset to blocking mode + self.channel.sock.settimeout(None) + self._pumping = False + + def _handle_message(self, message): + """Handle incoming DAP messages.""" + msg_type = message.get("type") + command = message.get("command", message.get("event", "unknown")) + seq = message.get("seq", 0) + + self._debug_print(f"[DAP] RECV: {msg_type} {command} (seq={seq})") + if message.get("arguments"): + self._debug_print(f"[DAP] args: {message['arguments']}") + + if msg_type == "request": + self._handle_request(message) + elif msg_type == "response": + # We don't expect responses from client + self._debug_print(f"[DAP] Unexpected response from client: {message}") + elif msg_type == "event": + # We don't expect events from client + self._debug_print(f"[DAP] Unexpected event from client: {message}") + + def _handle_request(self, message): + """Handle DAP request messages.""" + command = message.get("command") + seq = message.get("seq", 0) + args = message.get("arguments", {}) + + try: + if command == CMD_INITIALIZE: + self._handle_initialize(seq, args) + elif command == CMD_LAUNCH: + self._handle_launch(seq, args) + elif command == CMD_ATTACH: + self._handle_attach(seq, args) + elif command == CMD_SET_BREAKPOINTS: + self._handle_set_breakpoints(seq, args) + elif command == CMD_CONTINUE: + self._handle_continue(seq, args) + elif command == CMD_NEXT: + self._handle_next(seq, args) + elif command == CMD_STEP_IN: + self._handle_step_in(seq, args) + elif command == CMD_STEP_OUT: + self._handle_step_out(seq, args) + elif command == CMD_PAUSE: + self._handle_pause(seq, args) + elif command == CMD_STACK_TRACE: + self._handle_stack_trace(seq, args) + elif command == CMD_SCOPES: + self._handle_scopes(seq, args) + elif command == CMD_VARIABLES: + self._handle_variables(seq, args) + elif command == CMD_SET_VARIABLE: + self._handle_set_variable(seq, args) + elif command == CMD_EVALUATE: + self._handle_evaluate(seq, args) + elif command == CMD_DISCONNECT: + self._handle_disconnect(seq, args) + elif command == CMD_CONFIGURATION_DONE: + self._handle_configuration_done(seq, args) + elif command == CMD_THREADS: + self._handle_threads(seq, args) + elif command == CMD_SOURCE: + self._handle_source(seq, args) + else: + self.channel.send_response( + command, seq, success=False, message=f"Unknown command: {command}" + ) + + except Exception as e: + self.channel.send_response(command, seq, success=False, message=str(e)) + + def _handle_initialize(self, seq, args): + """Handle initialize request.""" + capabilities = { + "supportsConfigurationDoneRequest": True, + "supportsEvaluateForHovers": True, + "supportTerminateDebuggee": True, + "supportSuspendDebuggee": True, + "supportsTerminateRequest": True, + "supportsSetVariable": True, + # "supportsFunctionBreakpoints": False, + # "supportsConditionalBreakpoints": False, + # "supportsHitConditionalBreakpoints": False, + # "supportsStepBack": False, + # "supportsRestartFrame": False, + # "supportsGotoTargetsRequest": False, + # "supportsStepInTargetsRequest": False, + # "supportsCompletionsRequest": False, + # "supportsModulesRequest": False, + # "additionalModuleColumns": [], + # "supportedChecksumAlgorithms": [], + # "supportsRestartRequest": False, + # "supportsExceptionOptions": False, + # "supportsValueFormattingOptions": False, + # "supportsExceptionInfoRequest": False, + # "supportsDelayedStackTraceLoading": False, + # "supportsLoadedSourcesRequest": False, + # "supportsLogPoints": False, + # "supportsTerminateThreadsRequest": False, + # "supportsSetExpression": False, + # "supportsDataBreakpoints": False, + # "supportsReadMemoryRequest": False, + # "supportsWriteMemoryRequest": False, + # "supportsDisassembleRequest": False, + # "supportsCancelRequest": False, + # "supportsBreakpointLocationsRequest": False, + # "supportsClipboardContext": False, + } + + self.channel.send_response(CMD_INITIALIZE, seq, body=capabilities) + self.channel.send_event(EVENT_INITIALIZED) + self.initialized = True + + def _handle_launch(self, seq, args): + """Handle launch request.""" + # For attach-mode debugging, we don't need to launch anything + self.channel.send_response(CMD_LAUNCH, seq) + + def _handle_attach(self, seq, args): + """Handle attach request.""" + # Check if debug logging should be enabled + self.debug_logging = args.get("logToFile", False) + + self._debug_print(f"[DAP] Processing attach request with args: {args}") + print( + f"[DAP] Debug logging {'enabled' if self.debug_logging else 'disabled'} (logToFile={self.debug_logging})" + ) + + # get debugger root and debugee root from pathMappings + for pm in args.get("pathMappings", []): + # debugee - debugger + self.pdb.path_mappings.append((pm.get("remoteRoot", "./"), pm.get("localRoot", "./"))) + # # TODO: justMyCode, debugOptions , + + # Enable trace function + self.pdb.set_trace_function(self._trace_function) + self.channel.send_response(CMD_ATTACH, seq) + + # After successful attach, we might need to send additional events + # Some debuggers expect a 'process' event or thread events + self._debug_print("[DAP] Attach completed, debugging is now active") + + def _handle_set_breakpoints(self, seq, args): + """Handle setBreakpoints request.""" + source = args.get("source", {}) + filename = source.get("path", "") + breakpoints = args.get("breakpoints", []) + + # Debug log the source information + self._debug_print(f"[DAP] setBreakpoints source info: {source}") + + # Set breakpoints in pdb adapter + actual_breakpoints = self.pdb.set_breakpoints(filename, breakpoints) + + self.channel.send_response( + CMD_SET_BREAKPOINTS, seq, body={"breakpoints": actual_breakpoints} + ) + + def _handle_continue(self, seq, args): + """Handle continue request.""" + self.stepping = False + self.paused = False + self.pdb.continue_execution() + self.channel.send_response(CMD_CONTINUE, seq) + + def _handle_next(self, seq, args): + """Handle next (step over) request.""" + self.stepping = True + self.paused = False + self.pdb.step_over() + self.channel.send_response(CMD_NEXT, seq) + + def _handle_step_in(self, seq, args): + """Handle stepIn request.""" + self.stepping = True + self.paused = False + self.pdb.step_into() + self.channel.send_response(CMD_STEP_IN, seq) + + def _handle_step_out(self, seq, args): + """Handle stepOut request.""" + self.stepping = True + self.paused = False + self.pdb.step_out() + self.channel.send_response(CMD_STEP_OUT, seq) + + def _handle_pause(self, seq, args): + """Handle pause request.""" + self.paused = True + self.pdb.pause() + self.channel.send_response(CMD_PAUSE, seq) + + def _handle_stack_trace(self, seq, args): + """Handle stackTrace request.""" + stack_frames = self.pdb.get_stack_trace() + self.channel.send_response( + CMD_STACK_TRACE, + seq, + body={"stackFrames": stack_frames, "totalFrames": len(stack_frames)}, + ) + + def _handle_scopes(self, seq, args): + """Handle scopes request.""" + frame_id = args.get("frameId", 0) + self._debug_print(f"[DAP] Processing scopes request for frameId={frame_id}") + scopes = self.pdb.get_scopes(frame_id) + self._debug_print(f"[DAP] Generated scopes: {scopes}") + self.channel.send_response(CMD_SCOPES, seq, body={"scopes": scopes}) + + def _handle_variables(self, seq, args): + """Handle variables request.""" + variables_ref = args.get("variablesReference", 0) + variables = self.pdb.get_variables(variables_ref) + self.channel.send_response(CMD_VARIABLES, seq, body={"variables": variables}) + + def _handle_set_variable(self, seq, args): + """Handle setVariable request.""" + variables_ref = args.get("variablesReference", 0) + name = args.get("name", "") + value = args.get("value", "") + + if not name: + self.channel.send_response( + CMD_SET_VARIABLE, seq, success=False, message="No variable name provided" + ) + return + + self._debug_print( + f"[DAP] Processing setVariable request: name={name}, value={value}, ref={variables_ref}" + ) + + try: + updated_variable = self.pdb.set_variable(variables_ref, name, value) + self.channel.send_response(CMD_SET_VARIABLE, seq, body=updated_variable) + except Exception as e: + self.channel.send_response(CMD_SET_VARIABLE, seq, success=False, message=str(e)) + + def _handle_evaluate(self, seq, args): + """Handle evaluate request. + + `context` selects the contract PdbAdapter.evaluate_expression applies: + `repl`/`clipboard` (Debug Console, "Copy as Expression") may execute a + statement when `expression` isn't a valid expression; `watch`/`hover` + and any other or absent context stay read-only eval. + """ + expression = args.get("expression", "") + frame_id = args.get("frameId") + context = args.get("context", "watch") + if not expression: + self.channel.send_response( + CMD_EVALUATE, seq, success=False, message="No expression provided" + ) + return + try: + result = self.pdb.evaluate_expression(expression, frame_id, context) + self.channel.send_response( + CMD_EVALUATE, seq, body={"result": str(result), "variablesReference": 0} + ) + except Exception as e: + self.channel.send_response(CMD_EVALUATE, seq, success=False, message=str(e)) + + def _handle_disconnect(self, seq, args): + """Handle disconnect request.""" + self.channel.send_response(CMD_DISCONNECT, seq) + self.disconnect() + + def _handle_configuration_done(self, seq, args): + """Handle configurationDone request.""" + # This indicates that the client has finished configuring breakpoints + # and is ready to start debugging + self.configuration_done = True + self.channel.send_response(CMD_CONFIGURATION_DONE, seq) + + def _handle_threads(self, seq, args): + """Handle threads request.""" + # MicroPython is single-threaded, so return one thread + threads = [{"id": self.thread_id, "name": "main"}] + self.channel.send_response(CMD_THREADS, seq, body={"threads": threads}) + + def _handle_source(self, seq, args): + """Handle source request.""" + source = args.get("source", {}) + source_path = source.get("path", "") + if self._baremetal or not source_path: + # BUG: unable to read the source on ESP32 + # Possible an effect of the import / inialization sequence ? + # Nothe that other source files ( other.py) do not seem to get requested in the same way + self.channel.send_response(CMD_SOURCE, seq, success=False) + return + self._debug_print(f"[DAP] Processing source request for path: {source}") + try: + # Try to read the source file + with open(source_path) as f: + content = f.read() + self.channel.send_response(CMD_SOURCE, seq, body={"content": content}) + except Exception: + self.channel.send_response( + CMD_SOURCE, + seq, + success=False, + message="cancelled", + # message=f"Could not read source: {e}" + ) + + def _trace_function(self, frame, event: str, arg): + """Trace function called by sys.settrace.""" + # https://docs.python.org/3/library/sys.html#sys.settrace + global _twiddel + # Process any pending DAP messages frequently + + self.process_pending_messages() + # Handle breakpoints and stepping + if self.pdb.should_stop(frame, event, arg): + self._send_stopped_event( + STOP_REASON_BREAKPOINT + if self.pdb.hit_breakpoint + else STOP_REASON_STEP + if self.stepping + else STOP_REASON_PAUSE + ) + # Wait for continue command + self.pdb.wait_for_continue() + + # The trace function is invoked (with event set to 'call') whenever a new local scope is entered; + # it should return a reference to a local trace function to be used for the new scope, + # or None if the scope shouldn't be traced. + + return self._trace_function + + def _send_stopped_event(self, reason): + """Send stopped event to client.""" + self.channel.send_event( + EVENT_STOPPED, reason=reason, threadId=self.thread_id, allThreadsStopped=True + ) + + def wait_for_client(self, timeout_s=WAIT_FOR_CLIENT_TIMEOUT_S): + """Block until the client has sent configurationDone, or time out. + + Same busy-poll shape as PdbAdapter.wait_for_continue(): there is no + server thread, so nothing services the socket unless this loop drains + it. Replaces a fixed sleep with a deterministic handshake - breakpoints + set before configurationDone are already applied by the time this + returns because process_pending_messages() has drained them too. + Returns True once configurationDone arrives, False if the bounded + timeout elapses first (a hard failure is worse than continuing with a + clear log message: a client that never configures is a client bug or + a dropped connection, not something to hang on forever). + """ + start = time.ticks_ms() + while not self.configuration_done: + self.process_pending_messages() + if not self.connected or self.channel.closed: + print("[DAP] wait_for_client: connection closed before configurationDone") + return False + if time.ticks_diff(time.ticks_ms(), start) > timeout_s * 1000: + print( + "[DAP] wait_for_client: timed out after {}s waiting for configurationDone".format( + timeout_s + ) + ) + return False + time.sleep(0.01) + return True + + def trigger_breakpoint(self): + """Trigger a manual breakpoint.""" + if self.initialized: + self._send_stopped_event(STOP_REASON_BREAKPOINT) + + def debug_this_thread(self): + """Enable debugging for current thread.""" + if hasattr(sys, "settrace"): + sys.settrace(self._trace_function) + + def is_connected(self): + """Check if client is connected.""" + return self.connected and not self.channel.closed + + def disconnect(self): + """Disconnect from client.""" + self.connected = False + if hasattr(sys, "settrace"): + sys.settrace(None) + self.pdb.cleanup() + self.channel.close() diff --git a/python-ecosys/debugpy/debugpy/server/pdb_adapter.py b/python-ecosys/debugpy/debugpy/server/pdb_adapter.py new file mode 100644 index 000000000..da9144d06 --- /dev/null +++ b/python-ecosys/debugpy/debugpy/server/pdb_adapter.py @@ -0,0 +1,908 @@ +"""PDB adapter for integrating with MicroPython's trace system.""" + +import os +import sys +import time + +from micropython import const # type: ignore[import-untyped] + +from ..common.constants import ( + SCOPE_GLOBALS, + SCOPE_LOCALS, + STEP_INTO, + STEP_OUT, + STEP_OVER, + TRACE_CALL, + TRACE_EXCEPTION, + TRACE_LINE, + TRACE_RETURN, +) + +Any = object + +VARREF_LOCALS = const(1) +VARREF_GLOBALS = const(2) +VARREF_LOCALS_SPECIAL = const(3) +VARREF_GLOBALS_SPECIAL = const(4) + +# New constants for complex variable references +VARREF_COMPLEX_BASE = const(10000) # Base for complex variable references +MAX_CACHE_SIZE = const(50) # Limit cache size for memory constraints + + +class VariableReferenceCache: + """Lightweight cache for complex variable references optimized for MicroPython.""" + + def __init__(self, max_size: int = MAX_CACHE_SIZE): + self.cache: dict[int, Any] = {} + self.insertion_order: list[int] = [] # Track insertion order for proper FIFO + self.next_ref: int = VARREF_COMPLEX_BASE + self.max_size: int = max_size + + def add_variable(self, value: Any) -> int: + """Add a complex variable and return its reference ID.""" + # Clean cache if approaching limit + if len(self.cache) >= self.max_size: + self._cleanup_oldest() + + ref_id = self.next_ref + self.cache[ref_id] = value + self.insertion_order.append(ref_id) + self.next_ref += 1 + return ref_id + + def get_variable(self, ref_id: int): # -> Optional[Any] + """Get variable by reference ID.""" + return self.cache.get(ref_id) + + def _cleanup_oldest(self) -> None: + """Remove oldest entries to free memory - optimized for MicroPython.""" + if not self.cache or not self.insertion_order: + return + to_remove = max(1, len(self.cache) // 3) + # Direct list slicing is more memory efficient than iteration + keys_to_remove = self.insertion_order[:to_remove] + # Batch delete for efficiency + for key in keys_to_remove: + self.cache.pop(key, None) # Use pop with default to avoid KeyError + # Update insertion order in one operation + self.insertion_order = self.insertion_order[to_remove:] + + def clear(self) -> None: + """Clear all cached variables.""" + self.cache.clear() + self.insertion_order.clear() + + +# Also try checking by basename for path mismatches +def basename(path: str): + return path.split("/")[-1] if "/" in path else path + + +# Check if this might be a relative path match +def ends_with_path(full_path: str, relative_path: str): + """Check if full_path ends with relative_path components.""" + full_parts = full_path.replace("\\", "/").split("/") + rel_parts = relative_path.replace("\\", "/").split("/") + if len(rel_parts) > len(full_parts): + return False + return full_parts[-len(rel_parts) :] == rel_parts + + +# Augmented-assignment operators checked longest-first so e.g. "**=" is not +# mistaken for "*=" followed by stray text. +_AUG_ASSIGN_OPS = ("**=", "//=", ">>=", "<<=", "+=", "-=", "*=", "/=", "%=", "&=", "|=", "^=", "=") + + +def _is_ident_char(ch: str) -> bool: + """True for `[A-Za-z0-9_]` - MicroPython's `str` has no `.isalnum()`.""" + return ch.isalpha() or ch.isdigit() or ch == "_" + + +def _assigned_name(statement: str): + """Return the target name of a simple top-level assignment, or None. + + Recognises only `...` where the identifier is the very + first token and `` is `=` or an augmented-assignment operator. This + is a deliberately narrow, best-effort check - it does NOT catch: + multi-target assignment (`a = b = 1`, only `a` is seen), tuple/list + unpacking (`a, b = 1, 2`), attribute/subscript targets (`obj.x = 1`, + `d[k] = 1`), `def`/`class` statements (which also bind a name), a + `for`/`with ... as` binding, or an assignment that is not the first + statement on the line (e.g. after `;`). Those forms pass through + undetected; callers must treat a `None` result as "not proven safe", + never as "proven no shadowing". + """ + stripped = statement.strip() + if not stripped or stripped[0].isdigit() or not _is_ident_char(stripped[0]): + return None + i = 1 + n = len(stripped) + while i < n and _is_ident_char(stripped[i]): + i += 1 + name = stripped[:i] + rest = stripped[i:].lstrip() + for op in _AUG_ASSIGN_OPS: + if rest.startswith(op): + if op == "=" and rest[1:2] == "=": + return None # `==`, a comparison, not an assignment + return name + return None + + +def _shadowed_local_warning(statement: str, locals_dict): + """Build the honesty-rule warning for `statement`, or None if it doesn't apply. + + Fires when `_assigned_name` recognises a top-level assignment whose + target name is also a key in `locals_dict` (the paused frame's + `f_locals` snapshot): that name is about to be rebound in `f_globals` + only, so the LOCAL of the same name stays exactly as it was. On + firmware without local-name capture (`save_names` capability False), + `locals_dict` keys are synthetic `local_N` placeholders rather than + real identifiers, so a real name can never match and this warning + silently cannot fire there - a known limitation, not a bug. + """ + name = _assigned_name(statement) + if name and name in locals_dict: + return ( + f"Warning: '{name}' also exists as a LOCAL in this frame; " + "the local is unchanged (statement ran against globals only)." + ) + return None + + +class PdbAdapter: + """Adapter between DAP protocol and MicroPython's sys.settrace functionality.""" + + def __init__(self): + self.breakpoints: dict[str, dict[int, dict]] = {} + # filename -> {line_no: breakpoint_info} # todo - simplify + self.current_frame = None + self.step_mode = None # None, 'over', 'into', 'out' + self.step_frame = None + self.step_depth = 0 + self.paused = False + self.hit_breakpoint = False + self.continue_event = False + self.variables_cache = {} # frameId -> variables + self.var_cache = VariableReferenceCache() # Enhanced variable reference cache + self.frame_id_counter = 1 + self.path_mappings: list[tuple[str, str]] = [] + # list of [runtime_path -> vscode_path mapping] + self.file_mappings: dict[str, str] = {} + # runtime_path -> vscode_path mapping # todo : merge with .breakpoints + self.capabilities: dict = {} + # set by DebugSession at session start (see DebugSession.probe_capabilities); + # empty dict here means "not yet probed", treated as no set_local support + + def _debug_print(self, message): + """Print debug message only if debug logging is enabled.""" + if hasattr(self, "_debug_session") and self._debug_session.debug_logging: # type: ignore[attr-defined] + print(message) + + def _normalize_path(self, path: str): + """Normalize a file path for consistent comparisons.""" + # Convert to absolute path if possible + try: + if hasattr(os.path, "abspath"): + path = os.path.abspath(path) + elif hasattr(os.path, "realpath"): + path = os.path.realpath(path) + except: + pass + # Ensure consistent separators + path = path.replace("\\", "/") + return path + + def set_trace_function(self, trace_func): + """Install the trace function.""" + if hasattr(sys, "settrace"): + sys.settrace(trace_func) + else: + raise RuntimeError("sys.settrace not available") + + def _filename_as_debugee(self, path: str): + # check if we have a 1:1 file mapping for this path + if self.file_mappings.get(path): + return self.file_mappings[path] + # Check if we have a folder mapping for this path + for runtime_path, vscode_path in self.path_mappings: + if path.startswith(vscode_path): + path = path.replace(vscode_path, runtime_path, 1) + if path.startswith("//"): + path = path[1:] + # If no mapping found, return the original path + return path + + def _filename_as_debugger(self, path: str): + """Convert a file path to the debugger's expected format.""" + path = path or "" + if not path: + return path + if path.startswith("<"): + # Special case for or similar + return path + # Check if we have a 1:1 file mapping for this path + for runtime_path, vscode_path in self.path_mappings: + if path.startswith(runtime_path): + path = path.replace(runtime_path, vscode_path, 1) + return path + + # Check if we have a folder mapping for this path + for runtime_path, vscode_path in self.path_mappings: + if path.startswith(runtime_path): + path = path.replace(runtime_path, vscode_path, 1) + if path.startswith("//"): + path = path[1:] + # If no mapping found, return the original path + return path + + def set_breakpoints(self, filename: str, breakpoints: list[dict]): + """Set breakpoints for a file.""" + self.breakpoints[filename] = {} + local_name = self._filename_as_debugee(filename) + self.file_mappings[local_name] = filename + actual_breakpoints = [] + self._debug_print(f"[PDB] Setting breakpoints for file: {filename}") + + for bp in breakpoints: + line = bp.get("line") + if line: + if local_name != filename: + self.breakpoints[local_name] = {} + self._debug_print(f"[>>>] Setting breakpoints for local: {local_name}:{line}") + self.breakpoints[local_name][line] = {} + self.breakpoints[filename][line] = {} + actual_breakpoints.append( + {"line": line, "verified": True, "source": {"path": filename}} + ) + + self._debug_print(f"[PDB] Breakpoints set : {self.breakpoints}") + + return actual_breakpoints + + def should_stop(self, frame, event: str, arg): + """Determine if execution should stop at this point.""" + # HOT path - no debug printing here + self.current_frame = frame + self.hit_breakpoint = False + + # Get frame information + filename = frame.f_code.co_filename + lineno = frame.f_lineno + # Check for exact filename match first + if self.paused or (filename in self.breakpoints and lineno in self.breakpoints[filename]): + self._debug_print(f"[PDB] HIT BREAKPOINT (exact match) at {filename}:{lineno}") + # Record the path mapping (in this case, they're already the same) + # self.file_mappings[filename] = self._filename_as_debugger(filename) + # Cache frame attributes to reduce lookup overhead + _frame_code = frame.f_code + _filename = _frame_code.co_filename + _lineno = frame.f_lineno + + # Optimize dictionary lookups - use .get() to avoid double lookup + file_breakpoints = self.breakpoints.get(_filename) + if file_breakpoints and _lineno in file_breakpoints: + self.hit_breakpoint = True + return True + else: + # file not (yet) matched - this is slow so we do not want to do this often. + # TODO: use sys.path[] method to find the file, does not work for frozen .... + # if we have a path match , but no breakpoints - add it to the file_mappings dict simplify this check + if file_breakpoints is None: + self.breakpoints[_filename] = {} # Ensure the filename is in the breakpoints dict + if _filename not in self.file_mappings: + self.file_mappings[_filename] = self._filename_as_debugger(_filename) + + # Check stepping + _step_mode = self.step_mode + if _step_mode == STEP_INTO: + if event in (TRACE_CALL, TRACE_LINE): + self.step_mode = None + return True + + elif _step_mode == STEP_OVER: + if event == TRACE_LINE and frame == self.step_frame: + self.step_mode = None + return True + elif event == TRACE_RETURN and frame == self.step_frame: + # Continue stepping in caller + if hasattr(frame, "f_back") and frame.f_back: + self.step_frame = frame.f_back + else: + self.step_mode = None + + elif _step_mode == STEP_OUT: + if event == TRACE_RETURN and frame == self.step_frame: + self.step_mode = None + return True + + return False + + def continue_execution(self): + """Continue execution.""" + self.step_mode = None + self.continue_event = True + + def step_over(self): + """Step over (next line).""" + self.step_mode = "over" + self.step_frame = self.current_frame + self.continue_event = True + + def step_into(self): + """Step into function calls.""" + self.step_mode = "into" + self.continue_event = True + + def step_out(self): + """Step out of current function.""" + self.step_mode = "out" + self.step_frame = self.current_frame + self.continue_event = True + + def pause(self): + """Pause execution at next opportunity.""" + # This is handled by the debug session + self.paused = True + + def wait_for_continue(self): + """Wait for continue command (simplified implementation).""" + # In a real implementation, this would block until continue + # For MicroPython, we'll use a simple polling approach + self.continue_event = False + + # Process DAP messages while waiting for continue + self._debug_print("[PDB] Waiting for continue command...") + while not self.continue_event: + # Process any pending DAP messages (scopes, variables, etc.) + if hasattr(self, "_debug_session"): + self._debug_session.process_pending_messages() # type: ignore[arg-type] + time.sleep(0.01) + + def get_stack_trace(self): + """Get the current stack trace.""" + if not self.current_frame: + return [] + + frames = [] + frame = self.current_frame + frame_id = 0 + + while frame: + filename = frame.f_code.co_filename + name = frame.f_code.co_name + line = frame.f_lineno + if "" in filename or filename.endswith("debugpy.py"): + hint = "subtle" + else: + hint = "normal" + + # Use the VS Code path if we have a mapping, otherwise use the original path + debugger_path = self._filename_as_debugger(filename) + # Create StackFrame info + frames.append( + { + "id": frame_id, + "name": name, + "source": {"path": debugger_path}, + "line": line, + "column": 1, + "endLine": line, + "endColumn": 1, + "presentationHint": hint, + } + ) + + # Cache frame for variable access + self.variables_cache[frame_id] = frame + + # MicroPython doesn't have f_back attribute + if hasattr(frame, "f_back"): + frame = frame.f_back + else: + # Only return the current frame for MicroPython + break + frame_id += 1 + + return frames + + def get_scopes(self, frame_id): + """Get variable scopes for a frame.""" + scopes = [ + { + "name": SCOPE_LOCALS, + "variablesReference": frame_id * 1000 + VARREF_LOCALS, + "expensive": False, + }, + { + "name": SCOPE_GLOBALS, + "variablesReference": frame_id * 1000 + VARREF_GLOBALS, + "expensive": False, + }, + ] + return scopes + + def _process_special_variables(self, var_dict, read_only=False): + """Process special variables (those starting and ending with __).""" + variables = [] + for name, value in var_dict.items(): + if name.startswith("__") and name.endswith("__"): + try: + # Use lightweight serialization instead of json.dumps + value_str = self._lightweight_serialize(value) + type_str = type(value).__name__ + info = { + "name": name, + "value": value_str, + "type": type_str, + "variablesReference": 0, + } + if read_only: + info["presentationHint"] = {"attributes": ["readOnly"]} + variables.append(info) + except Exception: + variables.append(self._var_error(name)) + return variables + + def _process_regular_variables(self, var_dict, read_only=False): + """Process regular variables (excluding special ones) - optimized.""" + variables = [] + for name, value in var_dict.items(): + # Skip private/internal variables + if name.startswith("__") and name.endswith("__"): + continue + # Use fast path for variable info generation + info = self._get_variable_info_fast(name, value) + if read_only: + info["presentationHint"] = {"attributes": ["readOnly"]} + variables.append(info) + return variables + + def _is_expandable(self, value: Any) -> bool: + """Check if a variable can be expanded (has child elements).""" + return isinstance(value, (dict, list, tuple, set)) + + def _get_preview(self, value: Any, fallback_text: str = "") -> str: + """Get a 30-char preview of a variable value with '...' if truncated - optimized for MicroPython.""" + try: + # Get repr and truncate to exactly 30 chars with "..." if needed + repr_val = repr(value) + if len(repr_val) <= 30: + return repr_val + else: + return repr_val[:30] + "..." + except (TypeError, ValueError, MemoryError): + # Memory-safe fallback + return fallback_text or f"<{type(value).__name__} object>"[:30] + + def _get_variable_info(self, name: str, value: Any) -> dict[str, str | int]: + """Get DAP-compliant variable information with proper type handling.""" + try: + # Handle expandable types + if self._is_expandable(value): + var_ref = self.var_cache.add_variable(value) + preview = self._get_preview(value) # Always use consistent preview + + if isinstance(value, dict): + return { + "name": name, + "value": preview, + "type": "dict", + "variablesReference": var_ref, + "namedVariables": len(value), + "indexedVariables": 0, + } + elif isinstance(value, list): + return { + "name": name, + "value": preview, + "type": "list", + "variablesReference": var_ref, + "indexedVariables": len(value), + "namedVariables": 0, + } + elif isinstance(value, tuple): + return { + "name": name, + "value": preview, + "type": "tuple", + "variablesReference": var_ref, + "indexedVariables": len(value), + "namedVariables": 0, + } + elif isinstance(value, set): + return { + "name": name, + "value": preview, + "type": "set", + "variablesReference": var_ref, + "indexedVariables": len(value), + "namedVariables": 0, + } + + # Simple types - use the preview helper + preview = self._get_preview(value) + + return { + "name": name, + "value": preview, + "type": type(value).__name__, + "variablesReference": 0, + } + except Exception: + return self._var_error(name) + + def _get_variable_info_fast(self, name: str, value: Any) -> dict[str, str | int]: + """Fast path for variable info generation with reduced allocations.""" + try: + # Handle expandable types + if self._is_expandable(value): + var_ref = self.var_cache.add_variable(value) + preview = self._get_preview(value) # Always use consistent preview + + # Use pre-calculated length for better performance + length = 0 + try: + length = len(value) # type: ignore[arg-type] + except: + pass + + # Return optimized structure based on type + if isinstance(value, dict): + return { + "name": name, + "value": preview, + "type": "dict", + "variablesReference": var_ref, + "namedVariables": length if length < 1000 else 1000, # Cap for performance + "indexedVariables": 0, + } + elif isinstance(value, list): + return { + "name": name, + "value": preview, + "type": "list", + "variablesReference": var_ref, + "indexedVariables": min(length, 1000), # Cap for performance + "namedVariables": 0, + } + else: # tuple, set, other + return { + "name": name, + "value": preview, + "type": type(value).__name__, + "variablesReference": var_ref, + "indexedVariables": min(length, 1000), + "namedVariables": 0, + } + + # Simple types - optimized path + preview = self._get_preview(value) + return { + "name": name, + "value": preview, + "type": type(value).__name__, + "variablesReference": 0, + } + except Exception: + return {"name": name, "value": "", "type": "unknown", "variablesReference": 0} + + def _expand_complex_variable(self, ref_id: int) -> list[dict[str, str | int]]: + """Expand a complex variable into its child elements - optimized for memory.""" + value = self.var_cache.get_variable(ref_id) + if value is None: + return [] + + variables = [] + try: + if isinstance(value, dict): + # Limit dictionary expansion to prevent memory exhaustion + items = list(value.items()) + max_items = min(len(items), 50) # Limit to 50 items max + for i in range(max_items): + key, val = items[i] + key_str = str(key)[:50] # Limit key string length + variables.append(self._get_variable_info(key_str, val)) + if len(items) > max_items: + variables.append( + { + "name": f"<{len(items) - max_items} more items>", + "value": "...", + "type": "info", + "variablesReference": 0, + } + ) + elif isinstance(value, (list, tuple)): + # Limit list/tuple expansion + max_items = min(len(value), 100) # Limit to 100 items max + for i in range(max_items): + variables.append(self._get_variable_info(f"[{i}]", value[i])) + if len(value) > max_items: + variables.append( + { + "name": f"<{len(value) - max_items} more items>", + "value": "...", + "type": "info", + "variablesReference": 0, + } + ) + elif isinstance(value, set): + # Handle set elements with size limit + items = list(value) # Convert once + max_items = min(len(items), 50) + for i in range(max_items): + variables.append(self._get_variable_info(f"<{i}>", items[i])) + if len(items) > max_items: + variables.append( + { + "name": f"<{len(items) - max_items} more items>", + "value": "...", + "type": "info", + "variablesReference": 0, + } + ) + except Exception as e: + # Return error info for debugging + variables.append( + { + "name": "error", + "value": f"Failed to expand: {str(e)[:50]}", # Limit error message length + "type": "error", + "variablesReference": 0, + } + ) + + return variables + + @staticmethod + def _var_error(name: str): + return {"name": name, "value": "", "type": "unknown", "variablesReference": 0} + + @staticmethod + def _special_vars(varref: int): + return {"name": "Special", "value": "", "variablesReference": varref} + + def get_variables(self, variables_ref): + """Get variables for a scope with enhanced complex variable support.""" + # Handle complex variable expansion + if variables_ref >= VARREF_COMPLEX_BASE: + return self._expand_complex_variable(variables_ref) + + frame_id = variables_ref // 1000 + scope_type = variables_ref % 1000 + + if frame_id not in self.variables_cache: + return [] + + frame = self.variables_cache[frame_id] + + # Locals are read-only in DAP when this firmware has no _set_local + # (STORY-1.3): the edit affordance is greyed out client-side instead + # of setVariable failing with an error after the fact. Globals always + # stay editable - global write-back works on every firmware. + locals_read_only = not self.capabilities.get("set_local", False) + + # Handle special scope types first + if scope_type == VARREF_LOCALS_SPECIAL: + var_dict = frame.f_locals if hasattr(frame, "f_locals") else {} + return self._process_special_variables(var_dict, read_only=locals_read_only) + elif scope_type == VARREF_GLOBALS_SPECIAL: + var_dict = frame.f_globals if hasattr(frame, "f_globals") else {} + return self._process_special_variables(var_dict) + + # Handle regular scope types with special folder + variables = [] + if scope_type == VARREF_LOCALS: + var_dict = frame.f_locals if hasattr(frame, "f_locals") else {} + variables.append(self._special_vars(frame_id * 1000 + VARREF_LOCALS_SPECIAL)) + elif scope_type == VARREF_GLOBALS: + var_dict = frame.f_globals if hasattr(frame, "f_globals") else {} + variables.append(self._special_vars(frame_id * 1000 + VARREF_GLOBALS_SPECIAL)) + else: + # Invalid reference, return empty + return [] + + # Add regular variables with enhanced processing + read_only = locals_read_only if scope_type == VARREF_LOCALS else False + variables.extend(self._process_regular_variables(var_dict, read_only=read_only)) + return variables + + def evaluate_expression(self, expression, frame_id=None, context="watch"): + """Evaluate a DAP `evaluate` request in the context of a frame. + + `watch`/`hover` (and any other/absent `context`) keep the original, + read-only contract: `eval()` only - a statement is a `SyntaxError`, + surfaced as an evaluation error, exactly as before this method + gained statement support. + + `repl`/`clipboard` add statement execution: `eval()` is tried first + (so a plain expression like `1 + 1` still returns a value); a + `SyntaxError` falls back to `exec(expression, globals_dict)` against + the frame's live `f_globals` only. The locals snapshot is + deliberately never passed to `exec` as a namespace - `exec(code, g, + l)` binds a top-level assignment into `l`, and `l` here is a + disposable copy handed back to the caller and then discarded, so + the assignment would silently vanish instead of taking effect. Only + `globals_dict` is live, so a statement's top-level assignments land + in the running module namespace and are visible to the target + program after `continue`. See `_shadowed_local_warning` for the + honesty-rule warning this implies when the assigned name also + exists as a frame LOCAL. + """ + if frame_id is not None and frame_id in self.variables_cache: + frame = self.variables_cache[frame_id] + globals_dict = frame.f_globals if hasattr(frame, "f_globals") else {} + locals_dict = frame.f_locals if hasattr(frame, "f_locals") else {} + else: + # Use current frame + frame = self.current_frame + if frame: + globals_dict = frame.f_globals if hasattr(frame, "f_globals") else {} + locals_dict = frame.f_locals if hasattr(frame, "f_locals") else {} + else: + globals_dict = globals() + locals_dict = {} + + try: + result = eval(expression, globals_dict, locals_dict) + return result + except SyntaxError as e: + if context not in ("repl", "clipboard"): + raise Exception(f"Evaluation error: {e}") + except Exception as e: + raise Exception(f"Evaluation error: {e}") + + # Only repl/clipboard reach here, and only after eval() raised a + # SyntaxError - try `expression` as a statement instead. + try: + exec(expression, globals_dict) + except Exception as e: + raise Exception(f"Evaluation error: {e}") + + warning = _shadowed_local_warning(expression, locals_dict) + return warning if warning else "" + + def cleanup(self): + """Clean up resources with enhanced cache management.""" + self.variables_cache.clear() + self.var_cache.clear() # Clear variable reference cache + self.breakpoints.clear() + if hasattr(sys, "settrace"): + sys.settrace(None) + + def _lightweight_serialize(self, value): # noqa: PLR0911 + """Lightweight serialization optimized for MicroPython memory constraints.""" + if value is None: + return "None" + elif isinstance(value, bool): + return "true" if value else "false" + elif isinstance(value, (int, float)): + return str(value) + elif isinstance(value, str): + # Simple escaping for strings - avoid full JSON complexity + if len(value) > 30: + escaped = value[:27].replace('"', '\\"').replace("\n", "\\n") + return f'"{escaped}..."' + else: + escaped = value.replace('"', '\\"').replace("\n", "\\n") + return f'"{escaped}"' + elif isinstance(value, (list, tuple)): + if len(value) == 0: + return "[]" if isinstance(value, list) else "()" + elif len(value) <= 3: + # Show small collections in full + items = [self._lightweight_serialize(item) for item in value] + brackets = "[]" if isinstance(value, list) else "()" + return f"{brackets[0]}{', '.join(items)}{brackets[1]}" + else: + # Show preview for large collections + preview = f"{type(value).__name__}({len(value)} items)" + return preview + elif isinstance(value, dict): + if len(value) == 0: + return "{}" + elif len(value) <= 2: + # Show small dicts in preview form + items = [] + for k, v in value.items(): + key_str = self._lightweight_serialize(k) + val_str = self._lightweight_serialize(v) + items.append(f"{key_str}: {val_str}") + return "{" + ", ".join(items) + "}" + else: + return f"dict({len(value)} items)" + else: + # Fallback for other types + type_name = type(value).__name__ + try: + repr_val = repr(value) + if len(repr_val) > 30: + return f"<{type_name} object>" + else: + return repr_val + except: + return f"<{type_name} object>" + + def set_variable(self, variables_ref: int, name: str, value: str) -> dict[str, str | int]: + """Set a variable to a new value and return the updated variable info. + + This function can modify both global and local variables when using a MicroPython + build with settrace and local variable modification support (sys._set_local_var). + + For global variables: Works reliably on all MicroPython builds. + For local variables: Requires MicroPython build with C-level local variable support. + """ + # Handle complex variable references (not supported for setting) + if variables_ref >= VARREF_COMPLEX_BASE: + raise Exception("Cannot set variables in complex object expansions") + + frame_id = variables_ref // 1000 + scope_type = variables_ref % 1000 + + # Only allow setting variables in the topmost frame (frame_id = 0) + if frame_id != 0: + raise Exception("Variable modification is only allowed in the topmost frame") + + # Use the current frame for modification + frame = self.current_frame + if frame is None: + raise Exception("No current frame available") + + # Get the appropriate variable contexts + globals_dict = frame.f_globals if hasattr(frame, "f_globals") else {} + locals_dict = frame.f_locals if hasattr(frame, "f_locals") else {} + + try: + # Try to evaluate the new value as a Python expression + try: + new_value = eval(value, globals_dict, locals_dict) + except: + # If evaluation fails, treat as string literal + new_value = value + + if scope_type == VARREF_GLOBALS or scope_type == VARREF_GLOBALS_SPECIAL: + # Check if variable exists in globals + if name not in globals_dict: + raise Exception(f"Global variable '{name}' not found") + + # For global variables, direct assignment works reliably + globals_dict[name] = new_value + self._debug_print(f"[PDB] Successfully set global variable '{name}' = {new_value}") + + elif scope_type == VARREF_LOCALS or scope_type == VARREF_LOCALS_SPECIAL: + # Check if variable exists in locals + if name not in locals_dict: + raise Exception(f"Local variable '{name}' not found") + + # Try to use the frame._set_local method to set local variables + try: + if hasattr(frame, "_set_local"): + # Use the frame._set_local method (CPython-compatible API) + frame._set_local(name, new_value) + self._debug_print( + f"[PDB] Successfully set local variable '{name}' = {new_value}" + ) + else: + # Fallback error if the method is not available + raise Exception( + f"Cannot modify local variable '{name}'. " + f"This MicroPython build doesn't support local variable modification. " + f"Please use a MicroPython build with settrace and local variable support." + ) + except Exception as inner_e: + # If frame.set_local fails, provide detailed error + raise Exception( + f"Failed to modify local variable '{name}': {inner_e}. " + f"Local variables in MicroPython are stored in internal code_state->state[] slots. " + f"Consider using global variables for reliable modification during debugging." + ) + + else: + raise Exception("Invalid scope reference") + + # Return the updated variable info + return self._get_variable_info(name, new_value) + + except Exception as e: + raise Exception(f"Failed to set variable '{name}': {e}") diff --git a/python-ecosys/debugpy/demo.py b/python-ecosys/debugpy/demo.py new file mode 100644 index 000000000..fd88c0272 --- /dev/null +++ b/python-ecosys/debugpy/demo.py @@ -0,0 +1,74 @@ +#!/usr/bin/env python3 +"""Simple demo of MicroPython debugpy functionality.""" + +import sys + +sys.path.insert(0, ".") + +import debugpy + + +def simple_function(a, b): + """A simple function to demonstrate debugging.""" + result = a + b + print(f"Computing {a} + {b} = {result}") + return result + + +def main(): + print("MicroPython debugpy Demo") + print("========================") + print() + + # Demonstrate trace functionality + print("1. Testing trace functionality:") + + def trace_function(frame, event, arg): + if event == "call": + print(f" -> Entering function: {frame.f_code.co_name}") + elif event == "line": + print(f" -> Executing line {frame.f_lineno} in {frame.f_code.co_name}") + elif event == "return": + print(f" -> Returning from {frame.f_code.co_name} with value: {arg}") + return trace_function + + # Enable tracing + sys.settrace(trace_function) + + # Execute traced function + result = simple_function(5, 3) + + # Disable tracing + sys.settrace(None) + + print(f"Result: {result}") + print() + + # Demonstrate debugpy components + print("2. Testing debugpy components:") + + # Test PDB adapter + from debugpy.server.pdb_adapter import PdbAdapter + + pdb = PdbAdapter() + + # Set some mock breakpoints + breakpoints = pdb.set_breakpoints("demo.py", [{"line": 10}, {"line": 15}]) + print(f" Set breakpoints: {len(breakpoints)} breakpoints") + + # Test messaging + from debugpy.common.messaging import JsonMessageChannel + + print(" JsonMessageChannel available") + + print() + print("3. debugpy is ready for VS Code integration!") + print(" To use with VS Code:") + print(" - Import debugpy in your script") + print(" - Call debugpy.listen() to start the debug server") + print(" - Connect VS Code using the 'Attach to MicroPython' configuration") + print(" - Set breakpoints and debug normally") + + +if __name__ == "__main__": + main() diff --git a/python-ecosys/debugpy/development_guide.md b/python-ecosys/debugpy/development_guide.md new file mode 100644 index 000000000..94f06b420 --- /dev/null +++ b/python-ecosys/debugpy/development_guide.md @@ -0,0 +1,84 @@ +# Debugging MicroPython debugpy with VS Code + +## Method 1: Direct Connection with Enhanced Logging + +1. **Start MicroPython with enhanced logging:** + ```bash + ~/micropython2/ports/unix/build-standard/micropython test_vscode.py + ``` + + This will now show detailed DAP protocol messages like: + ``` + [DAP] RECV: request initialize (seq=1) + [DAP] args: {...} + [DAP] SEND: response initialize (req_seq=1, success=True) + ``` + +2. **Connect VS Code debugger:** + - Use the launch configuration in `.vscode/launch.json` + - Or manually attach to `127.0.0.1:5678` + +3. **Look for issues in the terminal output** - you'll see all DAP message exchanges + +## Method 2: Using DAP Monitor (Recommended for detailed analysis) + +1. **Start MicroPython debugpy server:** + ```bash + ~/micropython2/ports/unix/build-standard/micropython test_vscode.py + ``` + +2. **In another terminal, start the DAP monitor:** + ```bash + python3 dap_monitor.py + ``` + + The monitor listens on port 5679 and forwards to port 5678 + +3. **Connect VS Code to the monitor:** + - Modify your VS Code launch config to connect to port `5679` instead of `5678` + - Or create a new launch config: + ```json + { + "name": "Debug via Monitor", + "type": "python", + "request": "attach", + "connect": { + "host": "127.0.0.1", + "port": 5679 + } + } + ``` + +4. **Analyze the complete DAP conversation** in the monitor terminal + +## VS Code Debug Logging + +Enable VS Code's built-in DAP logging: + +1. **Open VS Code settings** (Ctrl+,) +2. **Search for:** `debug.console.verbosity` +3. **Set to:** `verbose` +4. **Also set:** `debug.allowBreakpointsEverywhere` to `true` + +## Common Issues to Look For + +1. **Missing required DAP capabilities** - check the `initialize` response +2. **Breakpoint verification failures** - look for `setBreakpoints` exchanges +3. **Thread/stack frame issues** - check `stackTrace` and `scopes` responses +4. **Evaluation problems** - monitor `evaluate` request/response pairs + +## Expected DAP Sequence + +A successful debug session should show this sequence: + +1. `initialize` request → response with capabilities +2. `initialized` event +3. `setBreakpoints` request → response with verified breakpoints +4. `configurationDone` request → response +5. `attach` request → response +6. When execution hits breakpoint: `stopped` event +7. `stackTrace` request → response with frames +8. `scopes` request → response with local/global scopes +9. `continue` request → response to resume + +If any step fails or is missing, that's where the issue lies. \ No newline at end of file diff --git a/python-ecosys/debugpy/manifest.py b/python-ecosys/debugpy/manifest.py new file mode 100644 index 000000000..6c4228298 --- /dev/null +++ b/python-ecosys/debugpy/manifest.py @@ -0,0 +1,6 @@ +metadata( + description="MicroPython implementation of debugpy for remote debugging", + version="0.1.0", +) + +package("debugpy") diff --git a/python-ecosys/debugpy/test_vscode.py b/python-ecosys/debugpy/test_vscode.py new file mode 100644 index 000000000..1d24fac81 --- /dev/null +++ b/python-ecosys/debugpy/test_vscode.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +"""Test script for VS Code debugging with MicroPython debugpy.""" + +import sys + +sys.path.insert(0, ".") + +import debugpy + +foo = 42 +bar = "Hello, MicroPython!" + + +def fibonacci(n): + """Calculate fibonacci number (iterative for efficiency).""" + if n <= 1: + return n + a, b = 0, 1 + for _ in range(2, n + 1): + a, b = b, a + b + return b + + +def debuggable_code(): + """The actual code we want to debug - wrapped in a function so sys.settrace will trace it.""" + global foo + print("Starting debuggable code...") + + # Test data - set breakpoint here (using smaller numbers to avoid slow fibonacci) + numbers = [3, 4, 5] + for i, num in enumerate(numbers): + print(f"Calculating fibonacci({num})...") + result = fibonacci(num) # <-- SET BREAKPOINT HERE (line 26) + foo += result # Modify foo to see if it gets traced + print(f"fibonacci({num}) = {result}") + print(sys.implementation) + import machine + + print(dir(machine)) + + # Test manual breakpoint + print("\nTriggering manual breakpoint...") + debugpy.breakpoint() + print("Manual breakpoint triggered!") + + print("Test completed successfully!") + + +def main(): + print("MicroPython VS Code Debugging Test") + print("==================================") + + # Start debug server + try: + debugpy.listen() + print("Debug server attached on 127.0.0.1:5678") + print("Connecting back to VS Code debugger now...") + # print("Set a breakpoint on line 26: 'result = fibonacci(num)'") + # print("Press Enter to continue after connecting debugger...") + # try: + # input() + # except: + # pass + + # Enable debugging for this thread + debugpy.debug_this_thread() + + # Give VS Code a moment to set breakpoints after attach + print("\nGiving VS Code time to set breakpoints...") + import time + + time.sleep(2) + + # Call the debuggable code function so it gets traced + debuggable_code() + + except KeyboardInterrupt: + print("\nTest interrupted by user") + except Exception as e: + print(f"Error: {e}") + + +if __name__ == "__main__": + main() diff --git a/python-ecosys/debugpy/vscode_launch_example.json b/python-ecosys/debugpy/vscode_launch_example.json new file mode 100644 index 000000000..388e696bd --- /dev/null +++ b/python-ecosys/debugpy/vscode_launch_example.json @@ -0,0 +1,22 @@ +{ + "version": "0.2.0", + "configurations": [ + { + "name": "Attach to MicroPython", + "type": "python", + "request": "attach", + "connect": { + "host": "localhost", + "port": 5678 + }, + "pathMappings": [ + { + "localRoot": "${workspaceFolder}", + "remoteRoot": "." + } + ], + "logToFile": true, + "justMyCode": false + } + ] +} \ No newline at end of file