"""
Auto-updater — compare local ``spacr`` to PyPI + the nightly branch.
Exposes a small API the Qt GUI's Help → "Check for updates" menu
entry can call. Nothing runs automatically; users always trigger a
check + confirm any upgrade.
The updater talks to two sources:
* **PyPI** — ``https://pypi.org/pypi/spacr/json`` for the latest
released version.
* **GitHub** — the nightly branch's HEAD commit hash, so nightly
users see how many commits they're behind.
It also answers Home's News panel, through :func:`fetch_release_notes`:
the published releases, cached for a day, so a running copy can show a
release newer than the one bundled in its own wheel.
Both fetches use ``urllib`` from the stdlib to avoid pulling in an
extra HTTP dependency. Timeouts are short (3 s) so a slow / offline
network doesn't block the UI. Errors are absorbed and surfaced as
"couldn't check" — never a crash.
Public API::
from spacr.updater import check_for_updates, run_pip_upgrade
info = check_for_updates() # UpdateInfo
if info.upgrade_available:
run_pip_upgrade()
"""
from __future__ import annotations
import json
import logging
import os
import re
import shutil
import subprocess
import sys
import time
from dataclasses import dataclass
from pathlib import Path
from typing import Optional, Tuple
from .logging_util import _spacr_home
LOG = logging.getLogger("spacr.updater")
PYPI_URL = "https://pypi.org/pypi/spacr/json"
GITHUB_NIGHTLY_API = (
"https://api.github.com/repos/EinarOlafsson/spacr/commits/nightly"
)
@dataclass
[docs]
class UpdateInfo:
"""Result of a version check.
:param installed_version: locally installed spaCR version, or ``"unknown"``
when neither distribution's metadata is readable.
:param latest_release: latest spaCR version returned by PyPI, or ``None``
when it is missing or unavailable.
:param nightly_sha: first seven characters of the nightly branch head
returned by GitHub, or ``None`` when unavailable.
:param error: first PyPI or GitHub request failure, prefixed by service
name, or ``None`` when neither request failed.
"""
installed_version: str
latest_release: Optional[str]
nightly_sha: Optional[str]
error: Optional[str] = None
@property
[docs]
def upgrade_available(self) -> bool:
"""Return whether PyPI advertises a version newer than this install."""
if not self.latest_release:
return False
from packaging.version import InvalidVersion, Version
try:
return Version(self.installed_version) < Version(self.latest_release)
except InvalidVersion:
return _lt(self.installed_version, self.latest_release)
def _newest_on_channel(payload, channel: str = "stable") -> str:
"""The version PyPI's JSON answer offers on ``channel``.
The stable channel is PyPI's own latest release. The nightly channel is
the newest version PyPI holds at all, pre-releases and development
builds included, skipping versions whose every file was yanked.
:param payload: the decoded ``https://pypi.org/pypi/spacr/json`` answer.
:param channel: ``"stable"`` or ``"nightly"``.
:returns: a version string, or ``""`` when the answer names none.
"""
best = str((payload or {}).get("info", {}).get("version") or "")
if channel != "nightly":
return best
from packaging.version import InvalidVersion, Version
try:
best_key = Version(best)
except InvalidVersion:
best_key = None
for text, files in ((payload or {}).get("releases") or {}).items():
files = [f for f in files or () if isinstance(f, dict)]
if not files or all(f.get("yanked") for f in files):
continue
try:
key = Version(str(text))
except InvalidVersion:
continue
if best_key is None or key > best_key:
best, best_key = str(text), key
return best
[docs]
def check_for_updates(timeout: float = 3.0) -> UpdateInfo:
"""Query PyPI + GitHub and return an :class:`UpdateInfo`.
:param timeout: per-request timeout in seconds.
"""
return _check_on_channel("stable", timeout)
def _check_on_channel(channel: str = "stable",
timeout: float = 3.0) -> UpdateInfo:
"""Query PyPI + GitHub for the versions ``channel`` offers.
:param channel: ``"stable"`` offers PyPI's latest release; ``"nightly"``
also offers pre-releases and development builds.
:param timeout: per-request timeout in seconds.
"""
_apply_network_settings()
installed = _installed_version()
latest = None
nightly = None
err = None
try:
import urllib.request
req = urllib.request.Request(
PYPI_URL, headers={"User-Agent": "spacr-updater"}
)
with urllib.request.urlopen(req, timeout=timeout) as r:
payload = json.loads(r.read())
latest = _newest_on_channel(payload, channel)
except Exception as e:
err = f"pypi: {e}"
LOG.debug("pypi check failed: %s", e)
try:
import urllib.request
req = urllib.request.Request(
GITHUB_NIGHTLY_API,
headers={"User-Agent": "spacr-updater",
"Accept": "application/vnd.github+json"},
)
with urllib.request.urlopen(req, timeout=timeout) as r:
payload = json.loads(r.read())
nightly = str(payload.get("sha") or "")[:7]
except Exception as e:
if err is None: err = f"github: {e}"
LOG.debug("nightly check failed: %s", e)
return UpdateInfo(
installed_version=installed,
latest_release=latest or None,
nightly_sha=nightly or None,
error=err,
)
GITHUB_RELEASES_API = (
"https://api.github.com/repos/EinarOlafsson/spacr/releases?per_page=100"
)
#: Redirects the news cache, for tests and for read-only homes. The same
#: escape hatch :func:`spacr.qt.space.cache_dir` and
#: :func:`spacr.qt.iconset.icon_cache_dir` offer, for the same reasons.
ENV_NEWS_CACHE = "SPACR_NEWS_CACHE"
#: One question a day, and the question is asked whether or not the last
#: one was answered. A failed attempt is stamped like a successful one
#: because the failure mode worth avoiding is a whole lab behind one NAT
#: asking api.github.com on every launch and being rate-limited together.
#: The cost of stamping a failure is that a machine that was offline at
#: launch keeps the bundled list until tomorrow, which is exactly what the
#: bundled list is for.
NEWS_MAX_AGE_S = 24 * 60 * 60
[docs]
def news_cache_path() -> Path:
"""Where the fetched release list is remembered between launches."""
override = os.environ.get(ENV_NEWS_CACHE)
root = Path(override) if override else _spacr_home() / "news"
return root / "releases.json"
def _news_links(body: str) -> list:
"""Every URL in ``body``, in order, de-duplicated.
The same shape ``tools/build_release_notes.py`` writes into the bundled
resource, so a fetched record and a bundled one are interchangeable and
nothing downstream has to know which it is holding.
"""
seen, out = set(), []
for url in re.findall(r"https?://[^\s<>)\]\"']+", body or ""):
url = url.rstrip(".,;:")
if url not in seen:
seen.add(url)
out.append(url)
return out
def _news_entries(payload) -> list:
"""Normalise the releases API's answer into bundled-resource records."""
entries = []
for release in payload or []:
if not isinstance(release, dict) or release.get("draft"):
continue
body = (release.get("body") or "").strip()
entries.append({
"tag": str(release.get("tag_name") or ""),
"name": str(release.get("name")
or release.get("tag_name") or "").strip(),
"published": str(release.get("published_at") or "")[:10],
"url": str(release.get("html_url") or ""),
"body": body,
"links": _news_links(body),
"prerelease": bool(release.get("prerelease")),
})
return entries
def _read_news_cache(max_age: float):
"""The cached release list when it is younger than ``max_age``.
:returns: the cached list, which may be empty when the last attempt
failed, or ``None`` when there is no usable cache to honour.
"""
try:
path = news_cache_path()
payload = json.loads(path.read_text(encoding="utf-8"))
fetched = float(payload.get("fetched") or 0.0)
except Exception:
return None
if not 0 < (time.time() - fetched) < max_age:
return None
releases = payload.get("releases")
if not isinstance(releases, list):
return None
return [r for r in releases if isinstance(r, dict)]
def _write_news_cache(releases: list) -> None:
"""Stamp this attempt, so the next launch does not repeat it."""
try:
path = news_cache_path()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(
json.dumps({"fetched": time.time(), "releases": releases}),
encoding="utf-8")
except Exception as e:
LOG.debug("could not write the news cache: %s", e)
[docs]
def fetch_release_notes(timeout: float = 4.0,
max_age: float = NEWS_MAX_AGE_S) -> list:
"""The repository's releases, newest first, or ``[]``.
NEVER CALL THIS ON THE GUI THREAD. It opens a socket. Home's News
panel asks for it through the same :class:`_UpdateWorker` the manual
update check runs on, after the page has been shown.
Unauthenticated, because the alternative is a token the user does not
have. That means the shared, per-address rate limit, which is why the
answer is cached for a day and why every failure is silent: the
bundled ``spacr/resources/release_notes.json`` remains the offline
source of truth and an empty list simply leaves it alone.
:param timeout: request timeout in seconds. Short on purpose.
:param max_age: how old a cached answer may be before it is asked
again, in seconds.
:returns: release records shaped like the bundled resource's, or an
empty list when the answer is unavailable for any reason at all.
"""
cached = _read_news_cache(max_age)
if cached is not None:
return cached
_apply_network_settings()
try:
import urllib.request
req = urllib.request.Request(
GITHUB_RELEASES_API,
headers={"User-Agent": "spacr-updater",
"Accept": "application/vnd.github+json"},
)
with urllib.request.urlopen(req, timeout=timeout) as r:
entries = _news_entries(json.loads(r.read()))
except Exception as e:
LOG.debug("release-notes fetch failed: %s", e)
_write_news_cache([])
return []
_write_news_cache(entries)
return entries
def _bundled_release_notes() -> list:
"""The release records shipped in this build, newest first, or ``[]``."""
try:
from importlib.resources import files
raw = files("spacr") / "resources" / "release_notes.json"
data = json.loads(raw.read_text(encoding="utf-8"))
return [r for r in data.get("releases") or [] if isinstance(r, dict)]
except Exception:
return []
def _release_version(record):
"""The :class:`packaging.version.Version` a release record names, or None."""
from packaging.version import InvalidVersion, Version
tag = str((record or {}).get("tag") or (record or {}).get("name") or "")
match = re.search(r"\d+(?:\.\d+)+\S*", tag)
if not match:
return None
try:
return Version(match.group(0))
except InvalidVersion:
return None
def _release_notes_between(old: str, new: str, releases=()) -> list:
"""The release records after ``old`` up to and including ``new``.
Fetched records come first so a newer description of the same release
replaces the bundled one; each version appears once.
:param old: the version that was running before the update.
:param new: the version running now.
:param releases: fetched release records, shaped like the bundled ones.
:returns: records newest first; empty when either version is unreadable
or nothing lies between them.
"""
from packaging.version import InvalidVersion, Version
try:
low, high = Version(str(old)), Version(str(new))
except InvalidVersion:
return []
picked = {}
for record in list(releases or ()) + _bundled_release_notes():
version = _release_version(record)
if version is None or version in picked:
continue
if low < version <= high:
picked[version] = record
return [picked[v] for v in sorted(picked, reverse=True)]
_UPDATE_CHANNELS = ("stable", "nightly")
_ENV_NETWORK_CONFIG = "SPACR_NETWORK_CONFIG"
_PROXY_VARIABLES = ("HTTPS_PROXY", "https_proxy", "HTTP_PROXY", "http_proxy")
_CA_VARIABLES = ("REQUESTS_CA_BUNDLE", "SSL_CERT_FILE", "CURL_CA_BUNDLE",
"PIP_CERT", "GIT_SSL_CAINFO", "NODE_EXTRA_CA_CERTS")
_NO_PROXY_VARIABLES = ("NO_PROXY", "no_proxy")
_LOCAL_HOSTS = "localhost,127.0.0.1,::1"
_NETWORK_EXPORTED: dict = {}
_NETWORK_DISPLACED: dict = {}
def _network_config_path() -> Path:
"""Where the proxy and certificate-bundle choice is kept for every process."""
override = os.environ.get(_ENV_NETWORK_CONFIG)
if override:
return Path(override)
return Path.home() / ".spacr" / "network.json"
def _read_network_config() -> dict:
"""The saved ``{"proxy": str, "ca_bundle": str}``, empty strings when unset."""
try:
data = json.loads(_network_config_path().read_text(encoding="utf-8"))
except Exception:
data = {}
if not isinstance(data, dict):
data = {}
return {"proxy": str(data.get("proxy") or "").strip(),
"ca_bundle": str(data.get("ca_bundle") or "").strip()}
def _write_network_config(proxy: str = "", ca_bundle: str = "") -> None:
"""Save the proxy and certificate bundle and export them at once.
Nothing is written when the choice is unchanged.
:param proxy: proxy URL such as ``http://proxy.example.org:3128``; empty
uses ``HTTPS_PROXY`` from the environment.
:param ca_bundle: path to a PEM file of trusted certificates; empty uses
``REQUESTS_CA_BUNDLE`` or ``SSL_CERT_FILE`` from the environment.
"""
wanted = {"proxy": str(proxy or "").strip(),
"ca_bundle": str(ca_bundle or "").strip()}
if wanted != _read_network_config():
path = _network_config_path()
path.parent.mkdir(parents=True, exist_ok=True)
path.write_text(json.dumps(wanted), encoding="utf-8")
_apply_network_settings()
def _undo_network_exports() -> None:
"""Put back every variable the last export replaced and nobody changed since."""
for key, value in list(_NETWORK_EXPORTED.items()):
if os.environ.get(key) == value:
original = _NETWORK_DISPLACED.get(key)
if original is None:
os.environ.pop(key, None)
else:
os.environ[key] = original
_NETWORK_EXPORTED.clear()
_NETWORK_DISPLACED.clear()
def _export(key: str, value: str) -> None:
"""Set one environment variable, remembering what it replaced."""
if os.environ.get(key) == value:
return
_NETWORK_DISPLACED[key] = os.environ.get(key)
_NETWORK_EXPORTED[key] = value
os.environ[key] = value
def _effective_network(environ=None) -> dict:
"""The proxy and certificate bundle every download will use.
A saved choice wins over the environment; otherwise ``HTTPS_PROXY`` (or
``HTTP_PROXY``) and ``REQUESTS_CA_BUNDLE`` (or ``SSL_CERT_FILE``,
``CURL_CA_BUNDLE``) are used as they are.
:param environ: the environment to read; the process environment when
omitted.
:returns: ``{"proxy", "ca_bundle", "proxy_source", "ca_source"}``.
"""
environ = os.environ if environ is None else environ
saved = _read_network_config()
proxy, proxy_source = saved["proxy"], "preferences"
if not proxy:
proxy_source = ""
for key in _PROXY_VARIABLES:
if environ.get(key):
proxy, proxy_source = environ[key], key
break
ca, ca_source = saved["ca_bundle"], "preferences"
if not ca:
ca_source = ""
for key in _CA_VARIABLES[:3]:
if environ.get(key):
ca, ca_source = environ[key], key
break
return {"proxy": proxy, "ca_bundle": ca,
"proxy_source": proxy_source, "ca_source": ca_source}
def _apply_network_settings() -> dict:
"""Export the proxy and certificate bundle to every downloader.
``requests``, ``urllib``, ``huggingface_hub``, pip, uv, conda, git and
the backend installers all run in this process or in children that
inherit its environment, and each reads a different variable: requests
and conda read ``REQUESTS_CA_BUNDLE``, urllib, httpx and uv read
``SSL_CERT_FILE``, pip reads ``PIP_CERT``, git ``GIT_SSL_CAINFO``. One
certificate bundle is therefore written to all of them, and one proxy to
both spellings of ``HTTPS_PROXY`` and ``HTTP_PROXY``, with this
machine's own addresses exempted. A certificate bundle that is not a
file is not exported. Variables an earlier call exported are restored
first, so clearing the saved choice brings back what was there before.
:returns: the :func:`_effective_network` answer that was applied.
"""
_undo_network_exports()
network = _effective_network()
if network["proxy"]:
for key in _PROXY_VARIABLES:
_export(key, network["proxy"])
if not any(os.environ.get(k) for k in _NO_PROXY_VARIABLES):
for key in _NO_PROXY_VARIABLES:
_export(key, _LOCAL_HOSTS)
if network["ca_bundle"] and os.path.isfile(network["ca_bundle"]):
for key in _CA_VARIABLES:
_export(key, network["ca_bundle"])
try:
import urllib.request
urllib.request.install_opener(None)
except Exception:
LOG.debug("could not reset urllib's opener", exc_info=True)
return network
def _installed_version() -> str:
"""Return the running ``spacr`` version, or ``"unknown"``."""
try:
from importlib.metadata import version
return version("spacr")
except Exception:
try:
from importlib.metadata import version
return version("spacr-nightly")
except Exception:
return "unknown"
def _lt(a: str, b: str) -> bool:
"""Return True iff version ``a`` is strictly less than ``b``.
Handles both 3-part and 4-part semver-ish strings, treating
missing parts as 0 (so ``1.4.1 < 1.4.1.1`` and
``1.4.1.1 < 1.4.2``).
"""
try:
pa = tuple(int(x) for x in a.split(".") if x.isdigit())
pb = tuple(int(x) for x in b.split(".") if x.isdigit())
except Exception:
return False
n = max(len(pa), len(pb))
pa = pa + (0,) * (n - len(pa))
pb = pb + (0,) * (n - len(pb))
return pa < pb
[docs]
def find_uv() -> Optional[str]:
"""The ``uv`` the desktop installers bootstrap, if this is such an install.
The native installers build their environment with ``uv venv``, which does
**not** seed ``pip``. On those installs ``python -m pip`` fails before it
starts, so the updater has to use the same tool the installer did. ``uv``
is bootstrapped one level above the venv::
<install root>/bootstrap/uv
<install root>/venv/ <- sys.prefix
:returns: an executable path, or ``None`` when this is an ordinary
pip-managed environment.
"""
bootstrap = Path(sys.prefix).parent / "bootstrap"
for name in ("uv.exe", "uv") if os.name == "nt" else ("uv", "uv.exe"):
candidate = bootstrap / name
if candidate.is_file() and os.access(candidate, os.X_OK):
return str(candidate)
found = shutil.which("uv")
return found or None
[docs]
def upgrade_command(pre_release: bool = False, *, target_version=None) -> list:
"""Upgrade this interpreter, optionally to the exact version offered.
:param pre_release: allow prerelease packages.
:param target_version: explicit version selected by the update check.
"""
if getattr(sys, "frozen", False):
raise RuntimeError(
"A frozen spaCR application is not a Python environment. "
"Update it with its installer instead of pip."
)
uv = find_uv()
if uv:
args = [uv, "pip", "install", "--upgrade",
"--python", sys.executable]
from .install_profile import read_profile
profile = read_profile()
if profile is not None:
args.extend(["--torch-backend", profile["requested_backend"]])
else:
args = [sys.executable, "-m", "pip", "install", "--upgrade"]
if pre_release:
args.append("--pre")
if target_version is not None:
from packaging.version import Version
args.append(f"spacr=={Version(str(target_version))}")
else:
args.append("spacr")
return args
[docs]
def editable_install_location() -> Optional[str]:
"""Return the active spaCR checkout, or ``None`` for a regular install.
:returns: the absolute path of the working tree, when this interpreter is
running spaCR out of one.
Detection first reads the PEP 610 ``direct_url.json`` editable-install
record. If that metadata is unavailable, it checks whether the imported
package resides outside ``site-packages`` and inside a directory containing
``.git`` or ``pyproject.toml``.
"""
try:
import importlib.metadata as md
import json
direct = md.distribution("spacr").read_text("direct_url.json")
if direct:
record = json.loads(direct)
if record.get("dir_info", {}).get("editable"):
url = str(record.get("url", ""))
if url.startswith("file://"):
from urllib.parse import unquote, urlparse
return unquote(urlparse(url).path)
if url:
return url
except Exception: # noqa: BLE001
pass
try:
import spacr as _spacr
here = os.path.abspath(os.path.dirname(
os.path.dirname(os.path.abspath(_spacr.__file__))))
except Exception: # noqa: BLE001
return None
for entry in sys.path + [getattr(sys, "prefix", "")]:
if not entry:
continue
marker = os.path.abspath(entry)
if os.path.basename(marker) in ("site-packages", "dist-packages") \
and here == marker:
return None
return here if os.path.isdir(os.path.join(here, ".git")) or \
os.path.isfile(os.path.join(here, "pyproject.toml")) else None
[docs]
def run_pip_upgrade(pre_release: bool = False, *, target_version=None):
"""Upgrade ``spacr`` in place, capturing what the packaging tool said.
:param pre_release: pass ``--pre`` so pre-releases and ``.postN``
versions are considered.
:param target_version: install this exact offered version and verify it in
a fresh process before reporting success. None retains an unpinned upgrade.
:returns: ``(exit_code, output)`` with combined stdout and stderr. Captured
output remains available to desktop installations launched without a
terminal.
"""
if getattr(sys, "frozen", False):
return 2, (
"A frozen spaCR application is not a Python environment. "
"Update it with its installer instead of pip."
)
editable = editable_install_location()
if editable:
return (2 if target_version is not None else 0, (
f"spaCR is installed in editable mode from {editable}, so there "
f"is nothing to upgrade: that folder IS the package, and pip "
f"would replace it with a release build. Update it with `git "
f"pull` there instead.\n"))
result = run_install_command(upgrade_command(pre_release, target_version=target_version))
if result[0] != 0 or target_version is None:
return result
probe = [sys.executable, "-I", "-c",
"from importlib.metadata import version; print(version('spacr'))"]
code, output = run_install_command(probe, timeout=30)
from packaging.version import InvalidVersion, Version
try:
verified = code == 0 and Version(output.strip()) == Version(str(target_version))
except InvalidVersion:
verified = False
if not verified:
return 1, (result[1] + "\nThe installer finished but the active interpreter "
f"did not verify spaCR {target_version}. Version check: {output.strip()}")
return result
[docs]
def launch_updated_app() -> int:
"""Launch the updated Qt application after the old event loop has stopped.
:returns: process identifier of the detached replacement application.
:raises OSError: the replacement could not be started.
"""
from .install_cleanup import _spawn_detached
from .restart_state import command
return _spawn_detached(command(), cwd=os.path.expanduser("~"))
[docs]
def run_install_command(args, timeout: float = 1800.0):
"""Run one packaging command, capturing everything it said.
Install offers use the same capture behavior as :func:`run_pip_upgrade`,
including when the application was launched without a terminal.
:param args: the argv to run, from :func:`upgrade_command` or
:func:`install_requirement_command`.
:param timeout: seconds before the install is given up on.
:returns: ``(exit_code, output)`` with stdout and stderr combined.
"""
args = [str(part) for part in args]
_apply_network_settings()
LOG.info("running: %s", " ".join(args))
try:
completed = subprocess.run(
args, capture_output=True, text=True, timeout=timeout)
except FileNotFoundError as exc:
LOG.exception("Packaging tool is missing")
return 1, f"Could not run {args[0]}: {exc}"
except subprocess.TimeoutExpired:
LOG.error("Command timed out after %s seconds", timeout)
return 1, (f"The command timed out after "
f"{int(timeout // 60)} minutes.")
output = "".join(part for part in
(completed.stdout or "", completed.stderr or "") if part)
if completed.returncode != 0:
LOG.error("Command failed (%s):\n%s", completed.returncode, output)
return completed.returncode, output
#: Packages where an install that MOVES OR REMOVES one changes spaCR's results
#: not to its tooling. A user pressing Install on an optional accelerator has
#: asked for a faster lasso, not for a numpy major upgrade under an
#: image-analysis stack -- so a plan that touches any of these is refused by
#: default and needs a second, explicit confirmation naming what moves.
PROTECTED_PACKAGES = ("numpy", "torch", "pandas", "scikit-learn")
[docs]
def canonical_package_name(name) -> str:
"""PEP 503 normalisation, so ``scikit_learn`` and ``Scikit-Learn`` match.
:param name: any spelling of a distribution name.
:returns: lower-case with runs of ``-``, ``_`` and ``.`` collapsed to a
single ``-``.
"""
import re
return re.sub(r"[-_.]+", "-", str(name or "").strip()).lower()
[docs]
def installed_version(name) -> Optional[str]:
"""The version of ``name`` installed here, or ``None`` if it is absent.
:param name: distribution name to query from installed package metadata.
"""
try:
from importlib.metadata import PackageNotFoundError, version
except Exception:
return None
try:
return str(version(str(name)))
except PackageNotFoundError:
return None
except Exception:
return None
[docs]
def pip_available() -> bool:
"""Is ``python -m pip`` usable in this interpreter?
``uv venv`` does not seed pip, which is the case :func:`find_uv` exists
for. It matters here because pip is the only one of the two that can
produce a machine-readable ``--report``.
"""
import importlib.util
try:
return importlib.util.find_spec("pip") is not None
except (ImportError, ValueError):
return False
[docs]
def install_requirement_command(requirement) -> list:
"""Return the command that installs ``requirement`` in this environment.
The same tool choice :func:`upgrade_command` makes -- ``uv`` when this is
a desktop install whose venv has no pip, ``python -m pip`` otherwise.
:param requirement: a pip requirement string, e.g. ``'cuml-cu12'``.
"""
uv = find_uv()
if uv and not pip_available():
return [uv, "pip", "install", "--python", sys.executable,
str(requirement)]
return [sys.executable, "-m", "pip", "install", str(requirement)]
[docs]
def dry_run_command(requirement) -> list:
"""Return the command that previews installation of ``requirement``.
pip's ``--report -`` writes a JSON document to stdout and installs
nothing; ``uv pip install --dry-run`` prints ``+ name==version`` lines.
Both are parsed by :func:`dry_run_install`, because the second is the only
one available on the desktop installs whose venv has no pip.
:param requirement: pip requirement string whose installation to preview.
"""
if pip_available():
return [sys.executable, "-m", "pip", "install", "--dry-run",
"--report", "-", str(requirement)]
uv = find_uv()
if uv:
return [uv, "pip", "install", "--dry-run", "--python", sys.executable,
str(requirement)]
return [sys.executable, "-m", "pip", "install", "--dry-run",
"--report", "-", str(requirement)]
@dataclass(frozen=True)
[docs]
class PackageChange:
"""One line of a dry-run report: what a package is now, and would be.
:param name: distribution name as reported by the resolver; its spelling
is retained.
:param current: installed version, or the version reported as removed by
uv, or ``None`` when the distribution is absent.
:param proposed: version the resolver would install, or ``None`` when it
would remove the distribution.
"""
name: str
current: Optional[str]
proposed: Optional[str]
@property
[docs]
def is_addition(self) -> bool:
"""Nothing is installed under this name today."""
return self.current is None
@property
[docs]
def is_move(self) -> bool:
"""A version already here would change."""
return (self.current is not None
and self.proposed is not None
and self.current != self.proposed)
@property
[docs]
def is_removal(self) -> bool:
"""A distribution installed today would be removed."""
return self.current is not None and self.proposed is None
@property
[docs]
def protected(self) -> bool:
"""Is this one of :data:`PROTECTED_PACKAGES`?"""
return canonical_package_name(self.name) in {
canonical_package_name(p) for p in PROTECTED_PACKAGES}
[docs]
def describe(self) -> str:
"""``'numpy 1.26.4 -> 2.2.6'`` or ``'cuml-cu12 26.8.0 (new)'``."""
if self.is_addition:
return f"{self.name} {self.proposed or '?'} (new)"
if self.is_move:
return f"{self.name} {self.current} -> {self.proposed}"
if self.is_removal:
return f"{self.name} {self.current} (removed)"
return f"{self.name} {self.current} (unchanged)"
@dataclass(frozen=True)
[docs]
class DryRun:
"""Represent a parsed ``pip install --dry-run`` result.
``ok`` is False when the resolver refused, when the tool could not be
run, or when it returned no machine-readable plan.
:param requirement: pip requirement string that was resolved.
:param ok: whether the packaging command succeeded and returned a readable
machine plan.
:param changes: parsed resolver entries, including additions, version
moves, and removals as :class:`PackageChange` records.
:param error: resolver or launch failure detail when ``ok`` is false,
otherwise ``None``.
:param raw: concatenated resolver stdout and stderr retained for
diagnostics.
"""
requirement: str
ok: bool
changes: Tuple[PackageChange, ...] = ()
error: Optional[str] = None
raw: str = ""
@property
[docs]
def additions(self) -> Tuple[PackageChange, ...]:
"""Packages that are not here at all today."""
return tuple(c for c in self.changes if c.is_addition)
@property
[docs]
def moves(self) -> Tuple[PackageChange, ...]:
"""Packages already installed whose version would change."""
return tuple(c for c in self.changes if c.is_move)
@property
[docs]
def removals(self) -> Tuple[PackageChange, ...]:
"""Packages already installed that the resolver would remove."""
return tuple(c for c in self.changes if c.is_removal)
@property
[docs]
def protected_moves(self) -> Tuple[PackageChange, ...]:
"""The moves that land on :data:`PROTECTED_PACKAGES`."""
return tuple(c for c in self.moves if c.protected)
@property
[docs]
def protected_changes(self) -> Tuple[PackageChange, ...]:
"""Protected packages whose version would change or be removed."""
return tuple(
change for change in self.changes
if change.protected and (change.is_move or change.is_removal)
)
[docs]
def summary(self) -> str:
"""Return the report shown before installation confirmation."""
if not self.ok:
return (f"Could not work out what installing {self.requirement} "
f"would change.\n{self.error or ''}".strip())
moves = self.moves
removals = self.removals
additions = self.additions
lines = [f"Installing {self.requirement} would:"]
if additions:
lines.append(f" add {len(additions)} package(s): "
+ ", ".join(f"{c.name} {c.proposed or '?'}"
for c in additions[:12])
+ (" ..." if len(additions) > 12 else ""))
if moves:
lines.append(" CHANGE the version of "
f"{len(moves)} package(s) already installed:")
lines.extend(f" {c.describe()}" for c in moves)
if removals:
lines.append(
f" REMOVE {len(removals)} package(s): "
+ ", ".join(f"{c.name} {c.current or '?'}"
for c in removals[:12])
+ (" ..." if len(removals) > 12 else "")
)
if not additions and not moves and not removals:
lines.append(" change nothing -- it is already satisfied.")
return "\n".join(lines)
[docs]
def dry_run_install(requirement, timeout: float = 600.0,
runner=None) -> DryRun:
"""Ask the packaging tool what installing ``requirement`` would change.
This function does not install packages. It resolves and reports proposed
additions and version changes so they can be reviewed before installation.
:param requirement: a pip requirement string.
:param timeout: seconds before the resolver is given up on.
:param runner: injected for tests; defaults to :func:`subprocess.run`.
:returns: a :class:`DryRun`.
"""
args = dry_run_command(requirement)
run = runner if runner is not None else subprocess.run
LOG.info("dry run: %s", " ".join(args))
try:
completed = run(args, capture_output=True, text=True, timeout=timeout)
except FileNotFoundError as exc:
return DryRun(str(requirement), False,
error=f"Could not run {args[0]}: {exc}")
except subprocess.TimeoutExpired:
return DryRun(str(requirement), False,
error=f"The resolver did not answer within "
f"{int(timeout)} seconds.")
except Exception as exc:
return DryRun(str(requirement), False, error=str(exc))
out = str(getattr(completed, "stdout", "") or "")
err = str(getattr(completed, "stderr", "") or "")
raw = out + err
if int(getattr(completed, "returncode", 1) or 0) != 0:
return DryRun(str(requirement), False,
error=_resolver_error(raw), raw=raw)
changes = _parse_pip_report(out)
if changes is None:
changes = _parse_uv_dry_run(raw)
if changes is None:
return DryRun(str(requirement), False,
error="The packaging tool produced no readable plan.",
raw=raw)
return DryRun(str(requirement), True, tuple(changes), raw=raw)
def _resolver_error(text: str) -> str:
"""The last few lines of a failed resolve -- where pip puts the reason."""
lines = [line for line in str(text).splitlines() if line.strip()]
if not lines:
return "The packaging tool failed and said nothing."
return "\n".join(lines[-6:])
def _parse_pip_report(text: str):
"""``pip install --report -`` JSON to changes, or ``None`` if it is not.
pip prints its own progress alongside the document on some versions, so
the JSON is found rather than assumed to start at character zero.
"""
body = str(text or "")
decoder = json.JSONDecoder()
start = body.find("{")
while start != -1:
try:
payload, _end = decoder.raw_decode(body, start)
except Exception:
start = body.find("{", start + 1)
continue
if not isinstance(payload, dict) or "install" not in payload:
start = body.find("{", start + 1)
continue
changes = []
for entry in payload.get("install") or []:
meta = (entry or {}).get("metadata") or {}
name = str(meta.get("name") or "").strip()
if not name:
continue
proposed = str(meta.get("version") or "") or None
changes.append(PackageChange(name, installed_version(name),
proposed))
return changes
return None
def _parse_uv_dry_run(text: str):
"""``uv pip install --dry-run`` output to changes, or ``None``.
uv prints ``+ name==version`` for what it would install and
``- name==version`` for what it would remove; a version move shows as
both, which is why the two are merged by name rather than listed twice.
"""
import re
added, removed = {}, {}
seen = False
for line in str(text or "").splitlines():
match = re.match(r"\s*([+-])\s+([A-Za-z0-9._-]+)==([^\s]+)\s*$", line)
if not match:
continue
seen = True
sign, name, version = match.groups()
(added if sign == "+" else removed)[canonical_package_name(name)] = (
name, version)
if not seen:
return None
changes = []
for key, (name, version) in added.items():
current = removed.get(key, (None, None))[1] or installed_version(name)
changes.append(PackageChange(name, current, version))
for key, (name, version) in removed.items():
if key not in added:
changes.append(PackageChange(name, version, None))
return changes
[docs]
def install_decision(dry_run: DryRun) -> dict:
"""Whether a plan may proceed, and what a second confirmation must say.
An install that would move or remove NumPy, PyTorch, pandas, or
scikit-learn is refused by default and needs a second confirmation naming
what changes.
:param dry_run: the result of :func:`dry_run_install`.
:returns: ``{allowed, needs_second_confirmation, moves, headline,
report}``. ``allowed`` is False when the dry run did not answer --
an install whose consequences are unknown is not offered.
"""
if not dry_run.ok:
return {'allowed': False, 'needs_second_confirmation': False,
'moves': (), 'headline': dry_run.summary(),
'report': dry_run.summary()}
protected = dry_run.protected_changes
if protected:
named = "; ".join(change.describe() for change in protected)
return {
'allowed': True,
'needs_second_confirmation': True,
'moves': protected,
'headline': (
"REFUSED BY DEFAULT. This install would change or remove "
"packages spaCR's "
f"own results depend on: {named}. Every measurement made "
"before and after would be made by different code. Confirm "
"again only if that is what you meant."),
'report': dry_run.summary(),
}
return {'allowed': True, 'needs_second_confirmation': False, 'moves': (),
'headline': "", 'report': dry_run.summary()}
#: The four states an installation offer can report: already available,
#: installable here, installable elsewhere, or unavailable.
OFFER_ACTIONS = ("ready", "install", "elsewhere", "impossible")
@dataclass(frozen=True)
[docs]
class InstallOffer:
"""What pressing **Install** on a greyed-out option should do.
One shape, two callers -- the regression backend picker
(:func:`spacr.regression_backends.backend_install_offer`) and the Image
UMAP's GPU acceleration (:func:`spacr.gpu_reduce.install_offer`) -- so the
panel that shows it does not have to know which asked.
:param action: offer state, normally one of :data:`OFFER_ACTIONS`; only
``"install"`` with a nonempty requirement can produce a command.
:param title: short capability heading shown by the availability interface.
:param message: primary explanation shown with the offer.
:param requirement: pip requirement used to build a local install command
for an install action, or ``None`` when no local command is available.
:param recipe: optional setup or external-environment instructions appended
to the message.
:param runs_anything: informational local-install marker set by
:func:`offer_install`; :attr:`command`, not this flag, controls
execution.
"""
action: str
title: str
message: str
requirement: Optional[str] = None
recipe: str = ""
runs_anything: bool = False
@property
[docs]
def command(self) -> Optional[list]:
"""The install command, or ``None`` when nothing may be run."""
if self.action != "install" or not self.requirement:
return None
return install_requirement_command(self.requirement)
[docs]
def as_text(self) -> str:
"""Message and recipe as one block, for a dialog or a log."""
parts = [self.message.strip()]
if self.recipe.strip():
parts.append(self.recipe.strip())
return "\n\n".join(part for part in parts if part)
[docs]
def offer_ready(title: str, message: str) -> InstallOffer:
"""An offer for something that is already available.
:param title: short heading shown for the available capability.
:param message: explanation shown with the offer.
"""
return InstallOffer("ready", title, message)
[docs]
def offer_install(title: str, message: str, requirement: str,
recipe: str = "") -> InstallOffer:
"""An offer that may run pip here, after a dry run and a confirmation.
:param title: short heading shown for the optional capability.
:param message: explanation shown with the install offer.
:param requirement: pip requirement string that can satisfy the feature.
"""
return InstallOffer("install", title, message, str(requirement), recipe,
runs_anything=True)
[docs]
def offer_elsewhere(title: str, message: str, recipe: str) -> InstallOffer:
"""An offer that names another environment and runs nothing.
:param title: short heading shown for the optional capability.
:param message: explanation of why installation must happen elsewhere.
:param recipe: instructions for preparing the external environment.
"""
return InstallOffer("elsewhere", title, message, None, recipe)
[docs]
def offer_impossible(title: str, message: str, recipe: str = "") -> InstallOffer:
"""An offer that says installing cannot help, and why.
:param title: short heading shown for the unavailable capability.
:param message: explanation of why installation cannot satisfy it.
"""
return InstallOffer("impossible", title, message, None, recipe)