naneos.uploader_update

Automatic update of a Raspberry Pi installation to the newest release.

naneos-uploader-update runs from the naneos_uploader_update.timer that the installer writes (once a day, randomized). It is cheap when there is nothing to do: one request to the PyPI JSON of the package, whose info.version is the newest release without pre-releases, compared with the installed version. The service is not touched in that case.

When PyPI has a newer release, the installer of exactly that release (git tag vX.Y.Z) is downloaded and run with --version X.Y.Z. The installer is the updater because it also carries the unit files and config drop-ins, which a plain pip install --upgrade would miss. It restarts the service once.

Never touched: an installation from a git ref (--ref, hardware testing) and pre-releases. The timer is switched with --auto-update / --no-auto-update on the installer line, with AUTO_UPDATE=on|off in the change file on the SD card, or with systemctl enable|disable --now naneos_uploader_update.timer.

make_plan(installed=__version__, source=None, fetch=fetch_json)

Decide without changing anything.

Source code in src/naneos/uploader_update.py

def make_plan(
    installed: str = __version__,
    source: str | None = None,
    fetch: Callable[[str], dict] = fetch_json,
) -> Plan:
    """Decide without changing anything."""
    plan = Plan(installed=installed, source=source or installed_from())
    if plan.source != "PyPI":
        plan.reason = "installed from a git ref, not updated automatically"
        return plan
    try:
        plan.latest = str(fetch(PYPI_URL)["info"]["version"])
    except Exception as e:  # network, JSON shape: try again tomorrow
        plan.reason = f"could not read the newest version from PyPI ({e})"
        return plan
    try:
        newer = version_key(plan.latest) > version_key(installed)
    except ValueError as e:
        plan.reason = str(e)
        return plan
    if not newer:
        plan.reason = f"up to date, newest release is {plan.latest}"
    return plan

run_installer(script, args)

Run the installer with its output going to our stdout (the journal).

Source code in src/naneos/uploader_update.py

def run_installer(script: Path, args: list[str]) -> int:
    """Run the installer with its output going to our stdout (the journal)."""
    sys.stdout.flush()
    return subprocess.run(["bash", str(script), *args], check=False).returncode

update(user, plan, fetch_text_=fetch_text, run=run_installer)

Download the installer of the newest release and run it. Returns its exit code.

Source code in src/naneos/uploader_update.py

def update(
    user: str,
    plan: Plan,
    fetch_text_: Callable[[str], str] = fetch_text,
    run: Callable[[Path, list[str]], int] = run_installer,
) -> int:
    """Download the installer of the newest release and run it. Returns its exit code."""
    assert plan.latest is not None
    ref = f"v{plan.latest}"
    try:
        script_text = fetch_text_(INSTALLER_URL.format(ref=ref))
    except Exception as e:
        # The release workflow creates the tag after the release is on PyPI, so for a
        # short while there is a release without its installer. The installer of
        # master is no substitute: it may already belong to the next release and
        # write unit files the installed version does not understand.
        print(f"could not download the installer of {ref} ({e}), trying again tomorrow")
        return 1
    print(f"installer from {ref}")
    with tempfile.TemporaryDirectory(prefix="naneos-update-") as tmp:
        script = Path(tmp) / "install.sh"
        script.write_text(script_text, encoding="utf-8")
        return run(script, ["--version", plan.latest, "--user", user])

version_key(version)

A sortable key for the version formats this project publishes (2.0.6, 2.0.6rc2).

Source code in src/naneos/uploader_update.py

def version_key(version: str) -> tuple[tuple[int, ...], int, int]:
    """A sortable key for the version formats this project publishes (2.0.6, 2.0.6rc2)."""
    match = _VERSION.match(version.strip())
    if match is None:
        raise ValueError(f"cannot compare version {version!r}")
    release = [int(part) for part in match.group("release").split(".")]
    while len(release) > 1 and release[-1] == 0:
        release.pop()
    return tuple(release), _STAGES[match.group("stage")], int(match.group("num") or 0)