Merge pull request #33 from BrianPugh/package-manager

Package Manager MVP

README.rst

@@ -3,21 +3,24 @@
 |Python compat| |PyPi| |GHA tests| |readthedocs|
 
 
-Belay
-=====
-
 .. inclusion-marker-do-not-remove
 
-Belay is a library that enables the rapid development of projects that interact with hardware via a MicroPython or CircuitPython compatible board.
 
-Belay works by interacting with the REPL interface of a MicroPython board from Python code running on PC.
+Belay is:
+
+* A python library that enables the rapid development of projects that interact with hardware via a MicroPython or CircuitPython compatible board.
+
+* A command-line tool for developing standalone MicroPython projects.
+
+* A MicroPython package manager.
 
 Belay supports wired serial connections (USB) and wireless connections via WebREPL over WiFi.
 
 `Quick Video of Belay in 22 seconds.`_
 
 See `the documentation`_ for usage and other details.
 
+
 Who is Belay For?
 =================
 
@@ -30,8 +33,8 @@ Examples include:
 
 * Read a potentiometer to control system volume.
 
-If you have no need to run Python code on PC, then Belay is not for you.
-
+The Belay Package Manager is for people that want to use public libraries, and get them on-device in
+an easy, repeatable, dependable manner.
 
 What Problems Does Belay Solve?
 ===============================
@@ -44,11 +47,12 @@ Typically, having a python script interact with hardware involves 3 major challe
 
 3. Computer-to-device communication protocol. How are commands and results transferred? How does the device execute those commands?
 
-
 This is lot of work if you just want your computer to do something simple like turn on an LED.
 Belay simplifies all of this by merging steps 1 and 2 into the same codebase, and manages step 3 for you.
 Code is automatically synced at the beginning of script execution.
 
+The Belay Package Manager makes it easy to cache, update, and deploy third party libraries with your project.
+
 Installation
 ============
 

belay/cli/common.py

@@ -1,4 +1,23 @@
+from pathlib import Path
+from typing import Union
+
+import tomli
+
 help_port = "Port (like /dev/ttyUSB0) or WebSocket (like ws://192.168.1.100) of device."
 help_password = (  # nosec
     "Password for communication methods (like WebREPL) that require authentication."
 )
+
+
+def load_toml(path: Union[str, Path] = "pyproject.toml"):
+    path = Path(path)
+
+    with path.open("rb") as f:
+        toml = tomli.load(f)
+
+    try:
+        toml = toml["tool"]["belay"]
+    except KeyError:
+        return {}
+
+    return toml

belay/cli/exec.py

@@ -14,3 +14,4 @@ def exec(
     """Execute python statement on-device."""
     device = Device(port, password=password)
     device(statement)
+    device.close()

belay/cli/identify.py

@@ -15,7 +15,11 @@ def identify(
     neopixel: bool = Option(False, help="Indicator is a neopixel."),
 ):
     """Display device firmware information and blink an LED."""
-    device = info(port=port, password=password)
+    device = Device(port, password=password)
+    version_str = "v" + ".".join(str(x) for x in device.implementation.version)
+    print(
+        f"{device.implementation.name} {version_str} - {device.implementation.platform}"
+    )
 
     if device.implementation.name == "circuitpython":
         device(f"led = digitalio.DigitalInOut(board.GP{pin})")
@@ -54,8 +58,11 @@ def set_led(state):
             def set_led(state):
                 led.value(state)  # noqa: F821
 
-    while True:
-        set_led(True)
-        sleep(0.5)
-        set_led(False)
-        sleep(0.5)
+    try:
+        while True:
+            set_led(True)
+            sleep(0.5)
+            set_led(False)
+            sleep(0.5)
+    finally:
+        device.close()

belay/cli/info.py

@@ -14,4 +14,4 @@ def info(
     print(
         f"{device.implementation.name} {version_str} - {device.implementation.platform}"
     )
-    return device
+    device.close()

belay/cli/install.py

@@ -0,0 +1,55 @@
+from pathlib import Path
+from typing import List, Optional
+
+from rich.progress import Progress
+from typer import Argument, Option
+
+from belay import Device
+from belay.cli.common import help_password, help_port, load_toml
+from belay.cli.run import run as run_cmd
+from belay.cli.sync import sync
+
+
+def install(
+    port: str = Argument(..., help=help_port),
+    password: str = Option("", help=help_password),
+    mpy_cross_binary: Optional[Path] = Option(
+        None, help="Compile py files with this executable."
+    ),
+    run: Optional[Path] = Option(None, help="Run script on-device after installing."),
+    main: Optional[Path] = Option(
+        None, help="Sync script to /main.py after installing."
+    ),
+):
+    """Sync dependencies and project itself."""
+    if run and run.suffix != ".py":
+        raise ValueError("Run script MUST be a python file.")
+    if main and main.suffix != ".py":
+        raise ValueError("Main script MUST be a python file.")
+    toml = load_toml()
+    pkg_name = toml["name"]
+
+    sync(
+        port=port,
+        folder=Path(".belay-lib"),
+        dst="/lib",
+        password=password,
+        keep=None,
+        ignore=None,
+        mpy_cross_binary=mpy_cross_binary,
+    )
+    sync(
+        port=port,
+        folder=Path(pkg_name),
+        dst=f"/{pkg_name}",
+        password=password,
+        keep=None,
+        ignore=None,
+        mpy_cross_binary=mpy_cross_binary,
+    )
+    if main:
+        with Device(port, password=password) as device:
+            device.sync(main, keep=True, mpy_cross_binary=mpy_cross_binary)
+
+    if run:
+        run_cmd(port=port, file=run, password=password)

belay/cli/main.py

@@ -5,12 +5,16 @@
 from belay.cli.exec import exec
 from belay.cli.identify import identify
 from belay.cli.info import info
+from belay.cli.install import install
 from belay.cli.run import run
 from belay.cli.sync import sync
+from belay.cli.update import update
 
 app = typer.Typer()
 app.command()(sync)
 app.command()(run)
 app.command()(exec)
 app.command()(info)
 app.command()(identify)
+app.command()(update)
+app.command()(install)

belay/cli/run.py

@@ -15,3 +15,4 @@ def run(
     device = Device(port, password=password)
     content = file.read_text()
     device(content)
+    device.close()

belay/cli/sync.py

@@ -42,3 +42,4 @@ def sync(
         )
 
         progress_update(description="Sync complete.")
+        device.close()

belay/cli/update.py

@@ -0,0 +1,20 @@
+from typing import List
+
+from rich.console import Console
+from typer import Argument
+
+from belay.packagemanager import download_dependencies
+
+from .common import load_toml
+
+
+def update(packages: List[str] = Argument(None, help="Specific package(s) to update.")):
+    console = Console()
+    toml = load_toml()
+
+    try:
+        dependencies = toml["dependencies"]
+    except KeyError:
+        return
+
+    download_dependencies(dependencies, packages=packages, console=console)

belay/device.py

@@ -737,6 +737,12 @@ def _preprocess_src_file_hash_helper(src_file):
             progress_update(description="Cleaning up...")
         self._exec_snippet("sync_end")
 
+    def __enter__(self):
+        return self
+
+    def __exit__(self, exc_type, exc_value, exc_tb):
+        self.close()
+
     def close(self) -> None:
         """Close the connection to device."""
         return self._board.close()

belay/packagemanager.py

@@ -0,0 +1,127 @@
+import ast
+from contextlib import nullcontext
+from pathlib import Path
+from typing import Dict, List, Optional, Union
+from urllib.parse import urlparse
+
+import httpx
+from rich.console import Console
+
+
+class NonMatchingURL(Exception):
+    pass
+
+
+def _strip_www(url: str):
+    if url.startswith("www."):
+        url = url[4:]
+    return url
+
+
+def _process_url_github(url: str):
+    """Transforms github-like url into githubusercontent."""
+    url = str(url)
+    parsed = urlparse(url)
+    netloc = _strip_www(parsed.netloc)
+    if netloc == "github.com":
+        # Transform to raw.githubusercontent
+        _, user, project, mode, branch, *path = parsed.path.split("/")
+        return f"https://raw.githubusercontent.com/{user}/{project}/{branch}/{'/'.join(path)}"
+    elif netloc == "raw.githubusercontent.com":
+        return f"https://raw.githubusercontent.com{parsed.path}"
+    else:
+        # TODO: Try and be a little helpful if url contains github.com
+        raise NonMatchingURL
+
+
+def _process_url(url: str):
+    parsers = [
+        _process_url_github,
+    ]
+    for parser in parsers:
+        try:
+            return parser(url)
+        except NonMatchingURL:
+            pass
+
+    # Unmodified URL
+    return url
+
+
+def _get_text(url: Union[str, Path]):
+    url = str(url)
+    if url.startswith(("https://", "http://")):
+        res = httpx.get(url)
+        res.raise_for_status()
+        return res.text
+    else:
+        # Assume local file
+        return Path(url).read_text()
+
+
+def download_dependencies(
+    dependencies: Dict[str, Union[str, Dict]],
+    packages: Optional[List[str]] = None,
+    local_dir: Union[str, Path] = ".belay-lib",
+    console: Optional[Console] = None,
+):
+    """Download dependencies.
+
+    Parameters
+    ----------
+    dependencies: dict
+        Dependencies to install (probably parsed from TOML file).
+    packages: Optional[List[str]]
+        Only download this package.
+    local_dir: Union[str, Path]
+        Download dependencies to this directory.
+        Will create directories as necessary.
+    console: Optional[Console]
+        Print progress out to console.
+    """
+    local_dir = Path(local_dir)
+    if not packages:
+        # Update all packages
+        packages = list(dependencies.keys())
+
+    if console:
+        cm = console.status("[bold green]Updating Dependencies")
+    else:
+        cm = nullcontext()
+
+    def log(*args, **kwargs):
+        if console:
+            console.log(*args, **kwargs)
+
+    with cm:
+        for pkg_name in packages:
+            dep = dependencies[pkg_name]
+            if isinstance(dep, str):
+                dep = {"path": dep}
+            elif not isinstance(dep, dict):
+                raise ValueError(f"Invalid value for key {pkg_name}.")
+
+            log(f"{pkg_name}: Updating...")
+
+            url = _process_url(dep["path"])
+            ext = Path(url).suffix
+
+            # Single file
+            dst = local_dir / (pkg_name + ext)
+            dst.parent.mkdir(parents=True, exist_ok=True)
+
+            new_code = _get_text(url)
+
+            if ext == ".py":
+                ast.parse(new_code)  # Check for valid python code
+
+            try:
+                old_code = dst.read_text()
+            except FileNotFoundError:
+                old_code = ""
+
+            if new_code == old_code:
+                log(f"{pkg_name}: No changes detected.")
+            else:
+                log(f"[bold green]{pkg_name}: Updated.")
+                dst.write_text(new_code)