omip: install OmniPython packages from the computer, with what they need

ROADMAP.md

@@ -15,5 +15,7 @@ if something is broken or missing for you, open an
   updater would fetch and install a new version for you.
 - **Progress inside a large file.** A long transfer of one big file shows
   only elapsed time; a percentage and the bytes moved would show how far along it is.
+- **`omip` in the editor and the browser.** **Install Package** and package search
+  offer `mpftp omip` for OmniPython packages, as the command line does.
 - **Translations.** The browser interface and the extension in your
   language, with English as the fallback for anything not yet translated.

cli/src/mpftp/cli.py

@@ -1831,6 +1831,26 @@ def cmd_mip(ns: argparse.Namespace) -> None:
             client.close()
 
 
+def cmd_omip(ns: argparse.Namespace) -> None:
+    if not ns.packages and not ns.self_file:
+        raise SystemExit("mpftp omip: name a package, or pass --self to put omip.py on the board")
+    client, mode = get_client()
+    try:
+        ensure_device(client, ns.device, ns.baud)
+        params: dict[str, Any] = {
+            "packages": ns.packages,
+            "target": ns.target,
+            "mpy": not ns.no_mpy,
+            "omip_file": ns.self_file,
+        }
+        if ns.index:
+            params["index"] = ns.index
+        out(client.call("omip_install", params))
+    finally:
+        if mode.startswith("sidecar"):
+            client.close()
+
+
 def cmd_circup(ns: argparse.Namespace) -> None:
     client, mode = get_client()
     try:
@@ -2824,6 +2844,31 @@ def build_parser() -> argparse.ArgumentParser:
     )
     mip.set_defaults(func=cmd_mip)
 
+    omip = sub.add_parser(
+        "omip",
+        parents=[device_opts],
+        help="Install OmniPython packages with what they need from micropython-lib, or omip.py itself",
+    )
+    omip.add_argument("packages", nargs="*")
+    omip.add_argument(
+        "--self",
+        dest="self_file",
+        action="store_true",
+        help="Copy omip.py to the board, so it can install for itself (MicroPython and CircuitPython)",
+    )
+    omip.add_argument(
+        "--target",
+        default="/lib",
+        help="Board install directory (default: /lib)",
+    )
+    omip.add_argument("--no-mpy", action="store_true")
+    omip.add_argument(
+        "--index",
+        default=None,
+        help="Look for the packages on this index alone (default: the OmniPython index, then micropython-lib)",
+    )
+    omip.set_defaults(func=cmd_omip)
+
     circ = sub.add_parser(
         "circup",
         parents=[device_opts],

cli/src/mpftp/omip.py

@@ -0,0 +1,129 @@
+# SPDX-FileCopyrightText: 2026 Brad Barnett
+#
+# SPDX-License-Identifier: MIT
+
+"""Work out what an OmniPython package needs, from the computer.
+
+A board with no network can't run ``omip`` on itself, so the host does the same
+walk and copies the files. A package is looked for on the OmniPython index and
+then on micropython-lib, and whatever it names under ``upstream`` in its JSON
+comes from micropython-lib. ``mip`` and mpremote ignore that key.
+
+Nothing here touches a board: `walk` returns the files, and the sidecar copies
+them.
+"""
+
+from __future__ import annotations
+
+import json
+import urllib.error
+import urllib.request
+from dataclasses import dataclass, field
+from typing import Optional
+
+INDEX = "https://lib.omnipython.org"
+UPSTREAM = "https://micropython.org/pi/v2"
+#: The one file that gives a board ``omip.install``, the bootstrap ``mip`` uses too.
+OMIP_FILE = "github:PyDevices/pydevices/lib/omip.py"
+
+_URL_PREFIXES = ("http://", "https://", "github:", "gitlab:", "codeberg:")
+_HOSTS = {
+    "codeberg:": "codeberg.org/api/v1/repos/{}/{}/raw/{p}?ref={}",
+    "github:": "raw.githubusercontent.com/{}/{}/{}/{p}",
+    "gitlab:": "gitlab.com/{}/{}/-/raw/{}/{p}",
+}
+_TIMEOUT = 30
+
+
+class OmipError(RuntimeError):
+    """A package or a file it needs can't be had."""
+
+
+@dataclass
+class PlannedFile:
+    path: str  # relative to the install target
+    url: str
+    short_hash: Optional[str] = None  # set when the file comes from an index
+
+
+@dataclass
+class Plan:
+    channel: str  # "py", or the board's .mpy version
+    files: list[PlannedFile] = field(default_factory=list)
+    seen: set[str] = field(default_factory=set)
+
+
+def fetch(url: str) -> Optional[bytes]:
+    """The bytes at ``url``, or None when the server says it isn't there."""
+    # Some hosts refuse urllib's own user agent.
+    request = urllib.request.Request(url, headers={"User-Agent": "mpftp"})
+    try:
+        with urllib.request.urlopen(request, timeout=_TIMEOUT) as response:
+            return response.read()
+    except urllib.error.HTTPError as e:
+        if e.code == 404:
+            return None
+        raise OmipError(f"HTTP {e.code} from {url}") from e
+    except (urllib.error.URLError, OSError) as e:
+        raise OmipError(f"Can't reach {url}: {getattr(e, 'reason', e)}") from e
+
+
+def _is_url(spec: str) -> bool:
+    return spec.startswith(_URL_PREFIXES)
+
+
+def _rewrite_url(url: str, version: Optional[str]) -> str:
+    for host, form in _HOSTS.items():
+        if url.startswith(host):
+            parts = url[len(host) :].split("/")
+            return "https://" + form.format(parts[0], parts[1], version or "HEAD", p="/".join(parts[2:]))
+    return url
+
+
+def _read_json(url: str) -> Optional[dict]:
+    body = fetch(url)
+    if body is None:
+        return None
+    try:
+        return json.loads(body)
+    except ValueError as e:
+        raise OmipError(f"Not JSON at {url}: {e}") from e
+
+
+def walk(plan: Plan, spec: str, version: Optional[str], indexes: list[str]) -> None:
+    """Add the files of ``spec``, and of everything it needs, to ``plan``."""
+    if spec in plan.seen:
+        return
+    plan.seen.add(spec)
+    source = indexes[0]
+    base = None
+    if _is_url(spec):
+        url = _rewrite_url(spec, version)
+        if spec.endswith((".py", ".mpy")):
+            plan.files.append(PlannedFile(spec.rsplit("/", 1)[-1], url))
+            return
+        if not spec.endswith(".json"):
+            url = url.rstrip("/") + "/package.json"
+        body = _read_json(url)
+        if body is None:
+            raise OmipError(f"Package not found: {url}")
+        base = url.rpartition("/")[0]
+    else:
+        for index in indexes:
+            body = _read_json(f"{index}/package/{plan.channel}/{spec}/{version or 'latest'}.json")
+            if body is not None:
+                source = index
+                break
+        else:
+            raise OmipError(f"Package not found: {spec} (looked on {', '.join(indexes)})")
+        indexes = [source]
+    for path, short_hash in body.get("hashes", ()):
+        plan.files.append(PlannedFile(path, f"{source}/file/{short_hash[:2]}/{short_hash}", short_hash))
+    for path, url in body.get("urls", ()):
+        if base and not _is_url(url):
+            url = f"{base}/{url}"
+        plan.files.append(PlannedFile(path, _rewrite_url(url, version)))
+    for dep, dep_version in body.get("deps", ()):
+        walk(plan, dep, dep_version, indexes)
+    for dep, dep_version in body.get("upstream", ()):
+        walk(plan, dep, dep_version, [UPSTREAM])

cli/src/mpftp/sidecar.py

@@ -3357,6 +3357,102 @@ def mip_install(
                 elif not self.transport:
                     self._repl_mode = False
 
+    def omip_install(
+        self,
+        packages: list[str],
+        target: Optional[str] = None,
+        mpy: bool = True,
+        index: Optional[str] = None,
+        omip_file: bool = False,
+    ) -> dict[str, Any]:
+        """Install packages the way ``omip`` does on a board, from the host.
+
+        Finds each package on the OmniPython index and then micropython-lib, or
+        on ``index`` alone, and follows its ``upstream`` key to micropython-lib.
+        Everything is downloaded before the first write, so a package that can't
+        be found leaves the board as it was. ``omip_file`` also copies ``omip.py``
+        itself, which is the one thing this does on CircuitPython.
+        """
+        from . import omip
+
+        resolved_target = (target or "/lib").rstrip("/") or "/lib"
+        if not resolved_target.startswith("/"):
+            raise RuntimeError(f"omip target must be an absolute board path (got {resolved_target!r})")
+        if (self.interpreter or "micropython") == "circuitpython":
+            if packages or not omip_file:
+                raise RuntimeError(
+                    "omip installs packages on MicroPython; use mpftp circup on CircuitPython"
+                )
+            return self._omip_file_to_circuitpython(resolved_target)
+
+        specs = ([omip.OMIP_FILE] if omip_file else []) + list(packages)
+        if not specs:
+            raise RuntimeError("omip_install: packages required")
+        indexes = [index.rstrip("/")] if index else [omip.INDEX, omip.UPSTREAM]
+        from mpremote import mip as mp_mip
+
+        def op(t):
+            channel: Any = "py"
+            if mpy:
+                t.exec("import sys")
+                channel = t.eval("getattr(sys.implementation, '_mpy', 0) & 0xFF") or "py"
+            plan = omip.Plan(str(channel))
+            for spec in specs:
+                name, _, version = spec.partition("@")
+                _notify("mip_progress", {"package": spec, "target": resolved_target, "phase": "start"})
+                omip.walk(plan, name, version or None, indexes)
+
+            logs: list[str] = []
+            downloads: list[tuple[str, bytes]] = []
+            for planned in plan.files:
+                dest = f"{resolved_target}/{planned.path}"
+                if planned.short_hash and mp_mip._check_exists(t, dest, planned.short_hash):
+                    logs.append(f"Exists: {dest}")
+                    continue
+                data = omip.fetch(planned.url)
+                if data is None:
+                    raise omip.OmipError(f"File not found: {planned.url}")
+                logs.append(f"Copying: {dest} from {planned.url}")
+                downloads.append((dest, data))
+            for dest, data in downloads:
+                mp_mip._ensure_path_exists(t, dest)
+                t.fs_writefile(dest, data, progress_callback=_progress_notifier(dest))
+            self._sync_remote_fs(t)
+            for spec in specs:
+                _notify("mip_progress", {"package": spec, "target": resolved_target, "phase": "done"})
+            return {
+                "output": "\n".join(logs),
+                "packages": specs,
+                "target": resolved_target,
+                "indexes": indexes,
+                "mpy": mpy,
+            }
+
+        return self.with_raw(self._transfer(op))
+
+    def _omip_file_to_circuitpython(self, target: str) -> dict[str, Any]:
+        """Copy ``omip.py`` on, through the drive when there is one."""
+        from . import omip
+
+        planned = omip.Plan("py")
+        omip.walk(planned, omip.OMIP_FILE, None, [omip.INDEX])
+        file = planned.files[0]
+        data = omip.fetch(file.url)
+        if data is None:
+            raise omip.OmipError(f"File not found: {file.url}")
+        dest = f"{target}/{file.path}"
+        if self._circuitpy_host_path(dest) is None:
+            from mpremote import mip as mp_mip
+
+            self.with_raw(lambda t: (self._ensure_cp_writable(t), mp_mip._ensure_path_exists(t, dest)))
+        written = self.fs_write(dest, base64.b64encode(data).decode())
+        return {
+            "output": f"Copying: {dest} from {file.url}",
+            "packages": [omip.OMIP_FILE],
+            "target": target,
+            "via": written.get("via", "serial"),
+        }
+
     def _find_circuitpy_host_roots(self) -> list[str]:
         """Host paths for a mounted CIRCUITPY drive (Windows letters or /media)."""
         roots: list[str] = []
@@ -4866,6 +4962,13 @@ def _debug_tee_loop(
         bool(p.get("mpy", True)),
         p.get("index"),
     ),
+    "omip_install": lambda p: SESSION.omip_install(
+        list(p.get("packages") or []),
+        p.get("target"),
+        bool(p.get("mpy", True)),
+        p.get("index"),
+        bool(p.get("omip_file", False)),
+    ),
     "circup_install": lambda p: SESSION.circup_install(
         list(p.get("packages") or []),
         py=bool(p.get("py", False)),