diff --git a/readme.md b/readme.md index 6fafa4e..b625a2e 100644 --- a/readme.md +++ b/readme.md @@ -61,6 +61,48 @@ For the source please refer to the samples folder Recommended : install stubs for your MCU of choice - [ ] Install stubs for MicroPython syntax checking `pip install micropython-rp2-stubs` (or your port of choise) +### Docker Backend Support (New!) + +**For users without physical hardware**, micropython-magic now supports running MicroPython code in Docker containers: + +```bash +# Install Docker support +pip install -U "micropython-magic[widgets]" docker + +# Ensure Docker is running +docker --version +``` + +Use the Docker backend in your notebooks: +```python +# Load the extension +%load_ext micropython_magic + +# Run MicroPython in Docker instead of physical hardware +%%micropython --backend docker +print("Hello from MicroPython in Docker!") +import sys +print(f"Running: {sys.version}") +``` + +**Benefits of Docker Backend:** +- ✅ **No hardware required** - Perfect for learning and development +- ✅ **Consistent environment** - Same MicroPython version across machines +- ✅ **Easy setup** - Just requires Docker installation +- ✅ **CI/CD friendly** - Can be used in automated testing +- ✅ **Safe experimentation** - No risk of damaging hardware + +**Limitations:** +- ❌ No GPIO/hardware access (no `machine` module functionality) +- ❌ No hardware-specific features (WiFi, sensors, etc.) +- ℹ️ Variables don't persist between cells (each execution is independent) + +The Docker backend uses the official `micropython/unix:latest` image and is ideal for: +- Learning MicroPython syntax and concepts +- Algorithm development and testing +- Code validation before deployment to hardware +- Educational environments and workshops + ## Usage **1) Create a notebook** @@ -113,6 +155,7 @@ Please refer to the [samples folder](samples/) for more examples 1. [board_control](samples/board_control.ipynb) - basic board control 1. [board_selection.](samples/board_selection.ipynb) - list connected boards and loop through them 1. [device_info](samples/device_info.ipynb) - Get simple access to port, board and hardware and firmware information +1. [**Docker Backend Demo**](samples/docker_backend_demo.ipynb) - 🆕 Run MicroPython in Docker containers (no hardware required) 1. [WOKWI](samples/wokwi.ipynb) - Use MicroPython magic with WOKWI as a simulator (no device needed) 1. [Plot rp2 CPU temp](samples/plot_cpu_temp_rp2.ipynb) - create a plot of the CPU temperature of a rp2040 MCU(bqplot) 1. [Display Memory Map](samples/mem_info.ipynb) - Micropython memory map visualizer diff --git a/samples/docker_backend_demo.ipynb b/samples/docker_backend_demo.ipynb new file mode 100644 index 0000000..c976960 --- /dev/null +++ b/samples/docker_backend_demo.ipynb @@ -0,0 +1,310 @@ +{ + "cells": [ + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "# MicroPython Magic with Docker Backend\n", + "\n", + "This notebook demonstrates the new Docker backend support for micropython-magic, allowing you to run MicroPython code in Docker containers instead of requiring physical hardware.\n", + "\n", + "## Setup\n", + "\n", + "First, load the micropython magic extension:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%load_ext micropython_magic" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Basic Docker Backend Usage\n", + "\n", + "Use the `--backend docker` parameter to run MicroPython code in a Docker container using the `micropython/unix:latest` image:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%%micropython --backend docker\n", + "print(\"Hello from MicroPython in Docker!\")\n", + "print(f\"Running on platform: {__name__}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## System Information\n", + "\n", + "Get information about the MicroPython environment running in Docker:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%micropython --backend docker --info" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## MicroPython Features\n", + "\n", + "Test MicroPython-specific functionality:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%%micropython --backend docker\n", + "import sys\n", + "print(f\"MicroPython version: {sys.version}\")\n", + "print(f\"Platform: {sys.platform}\")\n", + "print(f\"Implementation: {sys.implementation.name}\")\n", + "\n", + "# Test some basic operations\n", + "numbers = [1, 2, 3, 4, 5]\n", + "squared = [x**2 for x in numbers]\n", + "print(f\"Squares: {squared}\")\n", + "\n", + "# Test dictionary operations\n", + "data = {'name': 'MicroPython', 'type': 'interpreter', 'container': 'docker'}\n", + "for key, value in data.items():\n", + " print(f\"{key}: {value}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Mathematical Operations\n", + "\n", + "Test mathematical operations and imports:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%%micropython --backend docker\n", + "import math\n", + "\n", + "# Test mathematical operations\n", + "angle = math.pi / 4\n", + "print(f\"sin(π/4) = {math.sin(angle):.6f}\")\n", + "print(f\"cos(π/4) = {math.cos(angle):.6f}\")\n", + "print(f\"sqrt(2) = {math.sqrt(2):.6f}\")\n", + "\n", + "# Test some calculations\n", + "factorial_5 = 1\n", + "for i in range(1, 6):\n", + " factorial_5 *= i\n", + "print(f\"5! = {factorial_5}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Line Magic Examples\n", + "\n", + "Use line magic for quick expressions:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%micropython --backend docker print(\"Quick hello from line magic!\")" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%micropython --backend docker --eval \"2**10\"" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%micropython --backend docker --eval \"[x*2 for x in range(5)]\"" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## JSON and Data Structures\n", + "\n", + "Test JSON handling and complex data structures:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%%micropython --backend docker\n", + "import json\n", + "\n", + "# Create a complex data structure\n", + "data = {\n", + " 'project': 'micropython-magic',\n", + " 'backend': 'docker',\n", + " 'features': ['container execution', 'no hardware required', 'easy setup'],\n", + " 'numbers': [1, 2, 3.14, 42],\n", + " 'metadata': {\n", + " 'version': '1.0',\n", + " 'author': 'micropython-magic'\n", + " }\n", + "}\n", + "\n", + "# Convert to JSON and back\n", + "json_str = json.dumps(data)\n", + "print(f\"JSON length: {len(json_str)}\")\n", + "\n", + "parsed = json.loads(json_str)\n", + "print(f\"Project: {parsed['project']}\")\n", + "print(f\"Features: {', '.join(parsed['features'])}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Error Handling\n", + "\n", + "Test how errors are handled in the Docker backend:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%%micropython --backend docker\n", + "# This should work\n", + "try:\n", + " result = 10 / 2\n", + " print(f\"10 / 2 = {result}\")\n", + "except Exception as e:\n", + " print(f\"Error: {e}\")\n", + "\n", + "# This will cause an error but should be handled\n", + "try:\n", + " result = 10 / 0\n", + " print(f\"This won't print\")\n", + "except ZeroDivisionError as e:\n", + " print(f\"Caught division by zero: {e}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Custom Docker Image\n", + "\n", + "You can specify a custom Docker image if needed:" + ] + }, + { + "cell_type": "code", + "execution_count": null, + "metadata": {}, + "outputs": [], + "source": [ + "%%micropython --backend docker --docker-image micropython/unix:latest\n", + "print(\"Using explicitly specified micropython/unix:latest image\")\n", + "import sys\n", + "print(f\"Version: {sys.version}\")" + ] + }, + { + "cell_type": "markdown", + "metadata": {}, + "source": [ + "## Comparison with Serial Backend\n", + "\n", + "Here's how you would use the traditional serial backend (requires physical hardware):\n", + "\n", + "```python\n", + "# Serial backend (default) - requires connected MCU\n", + "%%micropython --backend serial\n", + "print(\"This would run on physical hardware\")\n", + "\n", + "# Docker backend - runs in container\n", + "%%micropython --backend docker \n", + "print(\"This runs in Docker container\")\n", + "```\n", + "\n", + "## Advantages of Docker Backend\n", + "\n", + "- **No Hardware Required**: Run MicroPython code without physical MCUs\n", + "- **Consistent Environment**: Same MicroPython version across different machines\n", + "- **Easy Setup**: Just requires Docker installation\n", + "- **Development and Testing**: Perfect for learning and prototyping\n", + "- **CI/CD**: Can be used in automated testing pipelines\n", + "\n", + "## Limitations\n", + "\n", + "- **No GPIO/Hardware Access**: Cannot control physical pins or sensors\n", + "- **No Hardware-Specific Modules**: modules like `machine`, `time` may have limited functionality\n", + "- **Session Persistence**: Variables don't persist between cells (each execution is independent)\n", + "\n", + "The Docker backend is ideal for learning MicroPython syntax, testing algorithms, and developing code before deploying to physical hardware." + ] + } + ], + "metadata": { + "kernelspec": { + "display_name": "Python 3", + "language": "python", + "name": "python3" + }, + "language_info": { + "codemirror_mode": { + "name": "ipython", + "version": 3 + }, + "file_extension": ".py", + "mimetype": "text/x-python", + "name": "python", + "nbconvert_exporter": "python", + "pygments_lexer": "ipython3", + "version": "3.8.0" + } + }, + "nbformat": 4, + "nbformat_minor": 4 +} \ No newline at end of file diff --git a/src/micropython_magic/docker_backend.py b/src/micropython_magic/docker_backend.py new file mode 100644 index 0000000..ec465cf --- /dev/null +++ b/src/micropython_magic/docker_backend.py @@ -0,0 +1,364 @@ +"""Docker backend for MicroPython execution. + +This module provides Docker-based MicroPython execution as an alternative to +serial MCU connections, using the micropython/unix Docker image. +""" + +import json +import os +import tempfile +import time +from pathlib import Path +from typing import List, Optional, Union + +from IPython.core.interactiveshell import InteractiveShell +from loguru import logger as log + +from micropython_magic.logger import MCUException +from micropython_magic.interactive import TIMEOUT + +JSON_START = " 1: + expression = cmd[1] + return self._exec_micropython_code(f"print({expression})", timeout) + elif cmd[0] == "exec": + # Handle exec commands + if len(cmd) > 1: + code = cmd[1] + return self._exec_micropython_code(code, timeout) + elif cmd[0] == "soft-reset": + # Handle soft reset + return self._soft_reset() + elif cmd[0] == "reset": + # Handle hard reset + return self._hard_reset() + elif cmd[0] == "run": + # Handle run file command + if len(cmd) > 1: + filename = cmd[1] + return self._run_file(filename, timeout) + else: + log.warning(f"Unsupported command for Docker backend: {cmd}") + return [] + + except Exception as e: + if "failed to connect" in str(e).lower(): + raise ConnectionError(str(e)) + raise MCUException(str(e)) + + def _exec_micropython_code(self, code: str, timeout: float) -> List[str]: + """Execute MicroPython code in the container.""" + try: + # For simple execution, just run the code directly + # For more complex persistence, we would need to maintain a REPL session + + # Execute the code directly in micropython + result = self.container.exec_run( + ["micropython", "-c", code], + stdout=True, + stderr=True, + tty=False, + ) + + output = result.output.decode('utf-8', errors='ignore') + if result.exit_code != 0: + raise MCUException(f"MicroPython execution failed: {output}") + + return output.strip().split('\n') if output.strip() else [] + + except Exception as e: + log.error(f"Failed to execute MicroPython code: {e}") + raise MCUException(str(e)) + + def _run_file(self, filename: str, timeout: float) -> List[str]: + """Run a file in the container.""" + try: + # Copy file to container + container_path = "/tmp/run_file.py" + + # Create tar archive with the file + import tarfile + import io + + tar_data = io.BytesIO() + with tarfile.open(fileobj=tar_data, mode='w') as tar: + tar.add(filename, arcname="run_file.py") + tar_data.seek(0) + + # Put the archive in the container + self.container.put_archive("/tmp", tar_data.getvalue()) + + # Execute the file + result = self.container.exec_run( + ["micropython", container_path], + stdout=True, + stderr=True, + tty=False, + ) + + output = result.output.decode('utf-8', errors='ignore') + if result.exit_code != 0: + raise MCUException(f"MicroPython execution failed: {output}") + + return output.strip().split('\n') if output.strip() else [] + + except Exception as e: + log.error(f"Failed to run file: {e}") + raise MCUException(str(e)) + + def _soft_reset(self) -> List[str]: + """Perform a soft reset by restarting the container.""" + try: + log.info("Performing soft reset (restarting container)") + self.container.restart() + time.sleep(2) # Wait for container to restart + return ["Soft reset complete"] + except Exception as e: + raise MCUException(f"Soft reset failed: {e}") + + def _hard_reset(self) -> List[str]: + """Perform a hard reset by recreating the container.""" + try: + log.info("Performing hard reset (recreating container)") + self.stop() + self._container = None + self._ensure_container() + return ["Hard reset complete"] + except Exception as e: + raise MCUException(f"Hard reset failed: {e}") + + def run_cell( + self, + cell: str, + *, + timeout: Union[int, float] = TIMEOUT, + follow: bool = True, + mount: Optional[str] = None, + ) -> List[str]: + """Execute a cell of MicroPython code.""" + if mount: + log.warning("Mount option not supported in Docker backend") + + return self._exec_micropython_code(cell, timeout) + + def copy_cell_to_mcu(self, cell: str, *, filename: str): + """Copy cell content to a file in the container.""" + try: + # Create temporary file with cell content + with tempfile.NamedTemporaryFile(mode='w', suffix='.py', delete=False) as f: + f.write("# Jupyter cell\n") + f.write(cell) + temp_file = f.name + + try: + # Create tar archive + import tarfile + import io + + tar_data = io.BytesIO() + with tarfile.open(fileobj=tar_data, mode='w') as tar: + tar.add(temp_file, arcname=filename) + tar_data.seek(0) + + # Copy to container + self.container.put_archive("/tmp", tar_data.getvalue()) + log.info(f"Cell copied to container file: {filename}") + + finally: + os.unlink(temp_file) + + except Exception as e: + raise MCUException(f"Failed to copy cell to container: {e}") + + def cell_from_mcu_file(self, filename: str) -> str: + """Read a file from the container.""" + try: + # Get file from container + archive_data, _ = self.container.get_archive(f"/tmp/{filename}") + + # Extract file content + import tarfile + import io + + with tarfile.open(fileobj=io.BytesIO(archive_data.data)) as tar: + member = tar.getmembers()[0] + file_data = tar.extractfile(member) + if file_data: + content = file_data.read().decode('utf-8') + return content + else: + raise MCUException(f"Failed to read file {filename}") + + except Exception as e: + raise MCUException(f"Failed to read file from container: {e}") + + def select_device(self, port: Optional[str], verify: bool = False): + """Select device (no-op for Docker, just ensure container is running).""" + self._ensure_container() + return f"Docker container: {self.container_name}" + + def get_fw_info(self, timeout: float): + """Get firmware info from MicroPython.""" + try: + info_code = """ +import sys +import json +info = { + 'version': sys.version, + 'implementation': sys.implementation.name, + 'platform': sys.platform, + 'backend': 'docker' +} +print(json.dumps(info)) +""" + result = self._exec_micropython_code(info_code, timeout) + if result and result[0].startswith('{'): + return json.loads(result[0]) + return {"backend": "docker", "status": "running"} + except Exception as e: + log.error(f"Failed to get firmware info: {e}") + return {"backend": "docker", "status": "error", "error": str(e)} + + @staticmethod + def load_json_from_MCU(line: str): + """Try to load JSON output from MicroPython.""" + result = DONT_KNOW + if line.startswith(JSON_START) and line.endswith(JSON_END): + line = line[7:-7] + if line == "none": + return None + try: + result = json.loads(line) + except json.JSONDecodeError: + try: + result = eval(line) + except Exception: + pass + return result + + def stop(self): + """Stop and remove the container.""" + if self._container: + try: + log.info(f"Stopping container {self.container_name}") + self._container.stop() + self._container.remove() + self._container = None + except Exception as e: + log.warning(f"Failed to stop container: {e}") + + def __enter__(self): + """Context manager entry.""" + return self + + def __exit__(self, exc_type, exc_val, exc_tb): + """Context manager exit.""" + # Don't automatically stop container to keep it alive between cells + pass \ No newline at end of file diff --git a/src/micropython_magic/mpr.py b/src/micropython_magic/mpr.py index 18718a0..2ba2ff1 100644 --- a/src/micropython_magic/mpr.py +++ b/src/micropython_magic/mpr.py @@ -20,6 +20,15 @@ JSON_END = "~json>" DONT_KNOW = "<~?~>" +# Import Docker backend with graceful fallback +try: + from .docker_backend import DockerMicroPython + DOCKER_AVAILABLE = True +except ImportError as e: + log.debug(f"Docker backend not available: {e}") + DockerMicroPython = None + DOCKER_AVAILABLE = False + class MCUInfo(dict): """A dict with MCU firmware attributes""" @@ -37,11 +46,27 @@ def __init__( shell: InteractiveShell, port: str = "auto", resume: bool = True, + backend: str = "serial", + docker_image: str = "micropython/unix:latest", ): self.shell: InteractiveShell = shell self.port: str = port # by default connect to the first device self.resume = resume # by default resume the device to maintain state + self.backend = backend # "serial" or "docker" + self.docker_image = docker_image self.timeout = TIMEOUT + + # Initialize backend + self._docker_backend = None + if self.backend == "docker": + if not DOCKER_AVAILABLE: + raise ImportError("Docker backend requested but docker package not available. Install with: pip install docker") + self._docker_backend = DockerMicroPython(shell, image=docker_image) + + @property + def is_docker_backend(self) -> bool: + """Check if using Docker backend.""" + return self.backend == "docker" @property def cmd_prefix(self) -> List[str]: @@ -71,12 +96,21 @@ def run_cmd( ): """run a command on the device and return the output""" assert isinstance(cmd, list) + + # Route to appropriate backend + if self.is_docker_backend: + return self._docker_backend.run_cmd( + cmd, + auto_connect=auto_connect, + stream_out=stream_out, + shell=shell, + timeout=timeout, + follow=follow, + ) + + # Original serial implementation if auto_connect: cmd = self.cmd_prefix + cmd - # if isinstance(cmd, str): - # cmd = f"""{self.cmd_prefix} {cmd}""" - # else: - # log.warning(f"cmd is not a string: {cmd}") with log.contextualize(port=self.port): log.debug(cmd) return ipython_run( @@ -89,6 +123,12 @@ def run_cmd( def select_device(self, port: Optional[str], verify: bool = False): """try to select the device to connect to by specifying the serial port name.""" + + # Route to appropriate backend + if self.is_docker_backend: + return self._docker_backend.select_device(port, verify) + + # Original serial implementation _port = port.strip() if port else "auto" if not verify: self.port = _port @@ -111,6 +151,17 @@ def run_cell( mount: Optional[str] = None, ): """run a codeblock on the device and return the output""" + + # Route to appropriate backend + if self.is_docker_backend: + return self._docker_backend.run_cell( + cell, + timeout=timeout, + follow=follow, + mount=mount, + ) + + # Original serial implementation """copy cell to a file and run it on the MCU""" with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: self._cell_to_file(f, cell) @@ -161,6 +212,12 @@ def run_mcu_file( def copy_cell_to_mcu(self, cell, *, filename: str): """copy cell to a file to the MCU""" + + # Route to appropriate backend + if self.is_docker_backend: + return self._docker_backend.copy_cell_to_mcu(cell, filename=filename) + + # Original serial implementation with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: self._cell_to_file(f, cell) # copy the file to the device @@ -180,6 +237,12 @@ def _cell_to_file(self, f, cell): def cell_from_mcu_file(self, filename): """read a file from the device and return the contents""" + + # Route to appropriate backend + if self.is_docker_backend: + return self._docker_backend.cell_from_mcu_file(filename) + + # Original serial implementation with tempfile.NamedTemporaryFile(mode="w", suffix=".py", delete=False) as f: # copy_cmd = f"cp :{filename} {f.name}" copy_cmd = ["cp", f":{filename}", f.name] @@ -208,6 +271,11 @@ def load_json_from_MCU(line: str): return result def get_fw_info(self, timeout: float): + # Route to appropriate backend + if self.is_docker_backend: + return self._docker_backend.get_fw_info(timeout) + + # Original serial implementation fw_info = {} # load datafile from installed package cmd = ["run", str(path_for_script("fw_info.py"))] diff --git a/src/micropython_magic/octarine.py b/src/micropython_magic/octarine.py index c265a0c..68a8974 100644 --- a/src/micropython_magic/octarine.py +++ b/src/micropython_magic/octarine.py @@ -57,9 +57,24 @@ def __init__(self, shell: InteractiveShell): super(MicroPythonMagic, self).__init__(shell) self.shell: InteractiveShell self._MCU: list[MPRemote2] = [MPRemote2(shell)] + self._current_backend = "serial" # Track current backend # self.port: str = "auto" # by default connect to the first device # self.resume = True # by default resume the device to maintain state set_xmode(mode=str(self.xmode)) + + def _get_mcu_for_backend(self, backend: str, docker_image: str = "micropython/unix:latest") -> MPRemote2: + """Get or create MCU instance for the specified backend.""" + if backend != self._current_backend or len(self._MCU) == 0: + # Create new MPRemote2 instance for the backend + log.info(f"Switching to {backend} backend") + mcu = MPRemote2( + self.shell, + backend=backend, + docker_image=docker_image + ) + self._MCU = [mcu] + self._current_backend = backend + return self._MCU[0] @observe("loglevel") def _loglevel_changed(self, change): @@ -129,6 +144,17 @@ def MCU(self) -> MPRemote2: @argument( "--select", "-s", "--connect", nargs="+", help="serial port to connect to", metavar="PORT" ) + @argument( + "--backend", + choices=["serial", "docker"], + default="serial", + help="Backend to use: 'serial' for USB/serial devices or 'docker' for micropython/unix container" + ) + @argument( + "--docker-image", + default="micropython/unix:latest", + help="Docker image to use for docker backend" + ) @argument( "--reset", "--soft-reset", action="store_true", help="Reset device (before running cell)." ) @@ -142,6 +168,9 @@ def micropython(self, line: str, cell: str = ""): if args.timeout == -1: args.timeout = self.timeout assert isinstance(args.timeout, float) + + # Get MCU instance for the specified backend + mcu = self._get_mcu_for_backend(args.backend, args.docker_image) if args.select: if len(args.select) > 1: @@ -149,27 +178,27 @@ def micropython(self, line: str, cell: str = ""): log.warning(f"{args.select=} not yet implemented") return else: - self.select(args.select[0]) + mcu.select_device(args.select[0]) if line: log.debug(f"{args=}") # pre processing - these can be combined with the main processing if args.hard_reset: - self.hard_reset() + mcu.run_cmd(["reset"]) elif args.reset: - self.soft_reset() + mcu.run_cmd(["soft-reset", "eval", "True"]) if args.writefile: log.debug(f"{args.writefile=}") if args.new: log.warning(f"{args.new=} not implemented") - self.MCU.copy_cell_to_mcu(cell, filename=args.writefile) + mcu.copy_cell_to_mcu(cell, filename=args.writefile) return if args.readfile: log.debug(f"{args.readfile=},{args.new=}") - code = self.MCU.cell_from_mcu_file(args.readfile) + code = mcu.cell_from_mcu_file(args.readfile) # if the first line contains a magic command, replace it with this magic command but with the options commented out if code.startswith("# %%"): code = "\n".join(code.split("\n")[1:]) @@ -183,7 +212,7 @@ def micropython(self, line: str, cell: str = ""): if not cell: raise UsageError("Please specify some MicroPython code to execute") log.trace(f"{cell=}") - output = self.MCU.run_cell( + output = mcu.run_cell( cell, timeout=args.timeout, follow=args.follow, mount=args.mount ) @@ -206,6 +235,17 @@ def micropython(self, line: str, cell: str = ""): @argument( "--select", "-s", "--connect", nargs="+", help="serial port to connect to", metavar="PORT" ) + @argument( + "--backend", + choices=["serial", "docker"], + default="serial", + help="Backend to use: 'serial' for USB/serial devices or 'docker' for micropython/unix container" + ) + @argument( + "--docker-image", + default="micropython/unix:latest", + help="Docker image to use for docker backend" + ) @argument("--verify", action="store_true", help="verify that the device can be connected to") @argument("--reset", "--soft-reset", action="store_true", help="reset device.") @argument("--hard-reset", action="store_true", help="reset device.") @@ -224,6 +264,9 @@ def mpy_line(self, line: str): if not isinstance(args.timeout, float): args.timeout = float(args.timeout) # type: ignore + # Get MCU instance for the specified backend + mcu = self._get_mcu_for_backend(args.backend, args.docker_image) + # try to fixup the expression after shell and argparse mangled it if args.statement and len(args.statement) >= 1: args.statement = get_code(line, args.statement[0]) @@ -242,22 +285,25 @@ def mpy_line(self, line: str): log.warning(f"{args.select=} for multiple MCUs not yet implemented") return else: - self.select(args.select[0], verify=args.verify) + mcu.select_device(args.select[0], verify=args.verify) if args.hard_reset: - self.hard_reset() + mcu.run_cmd(["reset"]) elif args.reset: - self.soft_reset() + mcu.run_cmd(["soft-reset", "eval", "True"]) elif args.bootloader: - self.MCU.run_cmd(["bootloader"]) + mcu.run_cmd(["bootloader"]) # processing if args.list: + if args.backend == "docker": + log.info("Docker backend: container instances will be listed when available") + return ["Docker backend active"] return self.list_devices() elif args.info: - return self.get_fw_info(args.timeout) + return mcu.get_fw_info(args.timeout) elif args.eval: - return self.eval(args.eval) + return self.eval(args.eval, mcu) elif args.statement: # Assemble the command to run @@ -265,7 +311,7 @@ def mpy_line(self, line: str): cmd = ["exec", statement] log.debug(f"{cmd=}") - return self.MCU.run_cmd( + return mcu.run_cmd( cmd, stream_out=bool(args.stream), timeout=float(args.timeout), @@ -289,7 +335,7 @@ def select(self, port: Optional[str], verify: bool = False): device = port.strip() if port else "auto" return self.MCU.select_device(device, verify=verify) - def eval(self, line: str): + def eval(self, line: str, mcu: Optional[MPRemote2] = None): """ Run a Micropython expression on an attached device using mpremote. Note that the expression @@ -299,6 +345,9 @@ def eval(self, line: str): Runs the statement on the MCU and tries to convert the output to a python object. If that fails it returns the raw output as a string. """ + if mcu is None: + mcu = self.MCU + # Assemble the command to run statement = line.strip() cmd_old = ( @@ -309,7 +358,7 @@ def eval(self, line: str): f"""import json; print('{JSON_START}',json.dumps({statement}),'{JSON_END}')""", ] log.trace(repr(cmd)) - output = self.MCU.run_cmd(cmd, stream_out=False) + output = mcu.run_cmd(cmd, stream_out=False) if isinstance(output, SList): matchers = [r"^.*Error:", r"^.*Exception:"] for ln in output.l: @@ -318,7 +367,7 @@ def eval(self, line: str): raise MCUException(ln) from eval(ln.split(":")[0]) # check for json output and try to convert it if ln.startswith(JSON_START) and ln.endswith(JSON_END): - result = self.MCU.load_json_from_MCU(ln) + result = mcu.load_json_from_MCU(ln) if result != DONT_KNOW: return result return output diff --git a/tests/test_docker_backend.py b/tests/test_docker_backend.py new file mode 100644 index 0000000..db31dd7 --- /dev/null +++ b/tests/test_docker_backend.py @@ -0,0 +1,99 @@ +#!/usr/bin/env python3 +""" +Unit tests for Docker backend functionality. +""" + +import pytest +import sys +import os +from unittest.mock import Mock, patch + +# Add src to path so we can import the module +sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', 'src')) + +from micropython_magic.docker_backend import DockerMicroPython, DOCKER_AVAILABLE +from micropython_magic.mpr import MPRemote2 +from micropython_magic.octarine import MicroPythonMagic +from IPython.core.interactiveshell import InteractiveShell + +@pytest.mark.skipif(not DOCKER_AVAILABLE, reason="Docker not available") +class TestDockerBackend: + """Test suite for Docker backend functionality.""" + + def setup_method(self): + """Set up test environment.""" + self.shell = InteractiveShell.instance() + + def test_docker_backend_creation(self): + """Test that we can create a Docker backend.""" + backend = DockerMicroPython(self.shell) + assert backend is not None + assert backend.image == "micropython/unix:latest" + assert backend.container_name == "micropython-magic-session" + + def test_mpremote2_docker_backend(self): + """Test MPRemote2 with Docker backend.""" + mpr = MPRemote2(self.shell, backend="docker") + assert mpr.backend == "docker" + assert mpr.is_docker_backend == True + assert mpr._docker_backend is not None + + def test_mpremote2_serial_backend(self): + """Test MPRemote2 with serial backend (default).""" + mpr = MPRemote2(self.shell) + assert mpr.backend == "serial" + assert mpr.is_docker_backend == False + assert mpr._docker_backend is None + + def test_magic_backend_switching(self): + """Test that magic can switch between backends.""" + magic = MicroPythonMagic(self.shell) + + # Test serial backend (default) + mcu_serial = magic._get_mcu_for_backend("serial") + assert mcu_serial.backend == "serial" + + # Test docker backend + mcu_docker = magic._get_mcu_for_backend("docker") + assert mcu_docker.backend == "docker" + + @pytest.mark.integration + def test_docker_execution_integration(self): + """Integration test for Docker execution.""" + try: + backend = DockerMicroPython(self.shell) + + # Test simple execution + result = backend._exec_micropython_code("print('Hello Docker!')", 10) + assert result == ['Hello Docker!'] + + # Test mathematical expression + result = backend._exec_micropython_code("print(2 + 2)", 10) + assert result == ['4'] + + # Clean up + backend.stop() + + except Exception as e: + pytest.skip(f"Docker integration test failed: {e}") + + @pytest.mark.integration + def test_magic_docker_integration(self): + """Integration test for magic commands with Docker.""" + try: + magic = MicroPythonMagic(self.shell) + + # Test line magic with info + result = magic.mpy_line("--backend docker --info") + assert isinstance(result, dict) + assert result.get('backend') == 'docker' + + # Test simple execution + result = magic.mpy_line("--backend docker print('Magic test')") + assert result == ['Magic test'] + + except Exception as e: + pytest.skip(f"Magic Docker integration test failed: {e}") + +if __name__ == "__main__": + pytest.main([__file__, "-v"]) \ No newline at end of file