"""CachyOS Update - Decky plugin backend. Runs as root (plugin.json declares the "root" flag) and drives pacman, yay, flatpak and fwupdmgr non-interactively, streaming every output line to the frontend. All user-facing wording lives in the frontend (src/i18n.ts). This module only ever returns stable identifiers - phase ids, hint ids, reason ids - so the UI can render them in the user's language. Raw command output is passed through untranslated on purpose: it is the tool's own output. This file is also runnable standalone for development on a machine without Decky installed: python3 main.py --selftest python3 main.py --check python3 main.py --update --dry-run """ import asyncio import json import os import re import shutil import sys import time from collections import deque from pathlib import Path try: import decky # type: ignore except ImportError: # standalone / development decky = None # -------------------------------------------------------------------------- # decky stub for standalone runs # -------------------------------------------------------------------------- if decky is None: class _StubLogger: def _p(self, level, msg, *a): print(f"[{level}] {msg % a if a else msg}", file=sys.stderr) def debug(self, msg, *a): self._p("debug", msg, *a) def info(self, msg, *a): self._p("info", msg, *a) def warning(self, msg, *a): self._p("warn", msg, *a) def error(self, msg, *a): self._p("error", msg, *a) def exception(self, msg, *a): self._p("error", msg, *a) class _StubDecky: """Mimics the subset of the decky module that this plugin uses.""" def __init__(self): base = Path(os.environ.get("TMPDIR", "/tmp")) / "decky-cachyos-update-dev" (base / "settings").mkdir(parents=True, exist_ok=True) (base / "runtime").mkdir(parents=True, exist_ok=True) self.DECKY_PLUGIN_SETTINGS_DIR = str(base / "settings") self.DECKY_PLUGIN_RUNTIME_DIR = str(base / "runtime") self.DECKY_USER = os.environ.get("USER", "root") self.DECKY_USER_HOME = os.path.expanduser("~") self.DECKY_PLUGIN_VERSION = "dev" self.logger = _StubLogger() async def emit(self, event, *args): if event == "cachyos_update_log": print(args[0]) else: print(f"<{event}> {args}", file=sys.stderr) decky = _StubDecky() # type: ignore STANDALONE = True else: STANDALONE = False # -------------------------------------------------------------------------- # constants # -------------------------------------------------------------------------- LOG_MAX_LINES = 500 PACMAN_LOCK = "/var/lib/pacman/db.lck" # Weight of each phase in the progress bar, roughly proportional to how long # each step usually takes on a handheld. PHASE_WEIGHTS = { "keyring": 5, "pacman": 40, "aur": 30, "flatpak_system": 12, "flatpak_user": 5, "fwupd": 8, } DEFAULT_SETTINGS = { "enable_aur": True, "enable_flatpak": True, "enable_fwupd": False, "auto_check": True, "check_interval_hours": 6, "notify_on_updates": True, } # Packages whose upgrade means the machine should be restarted. Mirrors the # trigger list of /usr/share/libalpm/hooks/cachyos-reboot-required.hook. REBOOT_TRIGGER_PATTERNS = [ re.compile(r"^linux(-.*)?$"), re.compile(r"^systemd(-.*)?$"), re.compile(r"^mesa$"), re.compile(r"^nvidia(-.*)?$"), re.compile(r"^wayland$"), re.compile(r"^xorg-server(-.*)?$"), re.compile(r"^mkinitcpio(-.*)?$"), re.compile(r"^dracut(-.*)?$"), re.compile(r"^(amd|intel)-ucode$"), re.compile(r"^linux-firmware(-.*)?$"), re.compile(r"^glibc$"), ] # Output markers that mean "this needs a human", mapped to a hint id that the # frontend turns into localized advice. Order matters: first match wins. ERROR_HINTS = [ ("unresolvable package conflicts", "pkg_conflict"), ("conflicting files", "file_conflict"), ("exists in filesystem", "file_conflict"), ("unresolvable dependencies", "unresolvable_deps"), ("invalid or corrupted package", "corrupt_package"), ("signature from", "bad_signature"), ("failed to synchronize", "sync_failed"), ("could not resolve host", "no_network"), ("failed to obtain auth", "fwupd_auth"), ("not authorized", "polkit_denied"), ("ac power required", "ac_power_required"), ("battery level is too low", "battery_too_low"), ] PROGRESS_RE = re.compile(r"\(\s*(\d+)\s*/\s*(\d+)\s*\)") UPGRADING_RE = re.compile(r"upgrading\s+([a-zA-Z0-9@._+-]+)") # -------------------------------------------------------------------------- # state # -------------------------------------------------------------------------- class State: def __init__(self): self.status = "idle" # idle | checking | updating | done | error self.phase = "" self.progress = 0.0 self.log = deque(maxlen=LOG_MAX_LINES) self.counts = {"pacman": 0, "aur": 0, "flatpak": 0, "fwupd": 0} self.details = {"pacman": [], "aur": [], "flatpak": [], "fwupd": []} self.last_check = 0.0 self.last_update = 0.0 self.reboot_required = False self.pacnew = [] self.failed_phases = [] self.hint_id = "" self.warnings = [] self.error_text = "" # only for unexpected exceptions def to_dict(self): return { "status": self.status, "phase": self.phase, "progress": self.progress, "log": list(self.log), "counts": self.counts, "details": self.details, "total": sum(self.counts.values()), "last_check": self.last_check, "last_update": self.last_update, "reboot_required": self.reboot_required, "pacnew": self.pacnew, "failed_phases": self.failed_phases, "hint_id": self.hint_id, "warnings": self.warnings, "error_text": self.error_text, } # -------------------------------------------------------------------------- # helpers # -------------------------------------------------------------------------- def _which(binary): return shutil.which(binary) def _base_env(): """Deterministic, non-interactive environment for every child process.""" env = dict(os.environ) # C.UTF-8 keeps pacman's error strings parseable regardless of the user's # locale - the ERROR_HINTS matching below relies on the English wording. env["LC_ALL"] = "C.UTF-8" env["LANG"] = "C.UTF-8" env["TERM"] = "dumb" env["PYTHONUNBUFFERED"] = "1" # Suppresses flatpak's ANSI progress bars and OSC escape sequences, which # would otherwise end up as garbage in the log panel. env["FLATPAK_FANCY_OUTPUT"] = "0" return env def _yay_env(): """yay must not see SUDO_USER/DOAS_USER when running as root. With either set, yay picks /tmp/yay as its build directory and creates it root-owned, which the de-elevated (systemd DynamicUser) build then cannot write to. Unset, it uses /var/cache/yay, which systemd creates and chowns itself via CacheDirectory=yay. """ env = _base_env() for key in ("SUDO_USER", "DOAS_USER", "SUDO_UID", "SUDO_GID"): env.pop(key, None) return env def _flatpak_user_cmd(flatpak, user, home, args): """Build a `flatpak --user` invocation for the desktop user. `runuser` sets HOME/USER/LOGNAME but inherits everything else. If Decky's service environment happens to carry XDG_DATA_HOME, FLATPAK_USER_DIR or friends, flatpak would silently operate on the wrong tree - so the environment is rebuilt from scratch with `env -i` instead of scrubbed. """ return [ "runuser", "-u", user, "--", "env", "-i", f"HOME={home}", f"USER={user}", f"LOGNAME={user}", "PATH=/usr/local/bin:/usr/bin:/bin", "LC_ALL=C.UTF-8", "FLATPAK_FANCY_OUTPUT=0", flatpak, ] + args def _is_root(): return os.geteuid() == 0 def _pacman_busy(): return os.path.exists(PACMAN_LOCK) _installed_cache = None def _installed(pkg): """True if a package is installed, without shelling out to pacman. The local database has thousands of entries, so the name set is built once and reused; it only changes across an update run, which invalidates it. """ global _installed_cache if _installed_cache is None: local = Path("/var/lib/pacman/local") try: _installed_cache = { p.name.rsplit("-", 2)[0] for p in local.iterdir() if p.is_dir() } except OSError: _installed_cache = set() return pkg in _installed_cache def _invalidate_installed_cache(): global _installed_cache _installed_cache = None # -------------------------------------------------------------------------- # plugin # -------------------------------------------------------------------------- class Plugin: def __init__(self): self.state = State() self.settings = dict(DEFAULT_SETTINGS) self.lock = asyncio.Lock() self._auto_task = None self._dry_run = False # -- lifecycle --------------------------------------------------------- async def _main(self): self._load_settings() self._restore_state() decky.logger.info("CachyOS Update backend started (root=%s)", _is_root()) if not _is_root(): decky.logger.error( 'Backend is NOT running as root - plugin.json is missing the ' '"root" flag. Updates will fail.' ) self._auto_task = asyncio.create_task(self._auto_check_loop()) async def _unload(self): if self._auto_task: self._auto_task.cancel() self._save_state() decky.logger.info("CachyOS Update backend stopped") # -- persistence ------------------------------------------------------- @property def _settings_file(self): return Path(decky.DECKY_PLUGIN_SETTINGS_DIR) / "settings.json" @property def _state_file(self): return Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / "state.json" def _load_settings(self): try: if self._settings_file.exists(): stored = json.loads(self._settings_file.read_text()) for key in DEFAULT_SETTINGS: if key in stored: self.settings[key] = stored[key] except Exception as exc: decky.logger.warning("Could not read settings: %s", exc) def _save_settings(self): try: self._settings_file.parent.mkdir(parents=True, exist_ok=True) self._settings_file.write_text(json.dumps(self.settings, indent=2)) except Exception as exc: decky.logger.warning("Could not save settings: %s", exc) def _restore_state(self): """Carry the last check result across plugin restarts.""" try: if not self._state_file.exists(): return stored = json.loads(self._state_file.read_text()) self.state.counts = stored.get("counts", self.state.counts) self.state.details = stored.get("details", self.state.details) self.state.last_check = stored.get("last_check", 0.0) self.state.last_update = stored.get("last_update", 0.0) except Exception as exc: decky.logger.warning("Could not read stored state: %s", exc) def _save_state(self): try: self._state_file.parent.mkdir(parents=True, exist_ok=True) self._state_file.write_text( json.dumps( { "counts": self.state.counts, "details": self.state.details, "last_check": self.state.last_check, "last_update": self.state.last_update, }, indent=2, ) ) except Exception as exc: decky.logger.warning("Could not save state: %s", exc) # -- emitting ---------------------------------------------------------- async def _log(self, line): line = line.rstrip() self.state.log.append(line) await decky.emit("cachyos_update_log", line) async def _set_phase(self, phase, done_weight, total_weight): self.state.phase = phase self.state.progress = done_weight / total_weight if total_weight else 0.0 await decky.emit("cachyos_update_progress", self.state.progress, phase) async def _set_status(self, status): self.state.status = status await decky.emit("cachyos_update_state", status) # -- process runner ---------------------------------------------------- async def _run( self, cmd, env=None, stream=True, phase_base=0.0, phase_span=0.0, merge_stderr=True, ): """Run a command, streaming stdout(+stderr) line by line. Returns (returncode, list_of_lines). `merge_stderr=False` is for probes whose output gets parsed - fwupdmgr in particular interleaves progress ticks on stderr that would otherwise corrupt the result. """ if self._dry_run: await self._log("$ " + " ".join(cmd)) return 0, [] if stream: await self._log("$ " + " ".join(cmd)) try: proc = await asyncio.create_subprocess_exec( *cmd, stdout=asyncio.subprocess.PIPE, stderr=( asyncio.subprocess.STDOUT if merge_stderr else asyncio.subprocess.DEVNULL ), stdin=asyncio.subprocess.DEVNULL, env=env or _base_env(), ) except FileNotFoundError: if stream: await self._log(f"!! {cmd[0]} not found") return 127, [] lines = [] assert proc.stdout is not None while True: raw = await proc.stdout.readline() if not raw: break line = raw.decode("utf-8", "replace").rstrip("\n") lines.append(line) if stream: await self._log(line) # pacman/yay print "(12/34) upgrading foo" - use it to move the # bar inside a phase instead of only between phases. if phase_span: match = PROGRESS_RE.search(line) if match: cur, total = int(match.group(1)), int(match.group(2)) if total: frac = min(cur / total, 1.0) self.state.progress = phase_base + phase_span * frac await decky.emit( "cachyos_update_progress", self.state.progress, self.state.phase, ) rc = await proc.wait() return rc, lines # -- checking ---------------------------------------------------------- async def _check_pacman(self): if not _which("checkupdates"): return [], "missing_checkupdates" env = _base_env() # A stable database path: the default lives in TMPDIR and would be # re-downloaded on every call. db = Path(decky.DECKY_PLUGIN_RUNTIME_DIR) / "checkupdates-db" db.mkdir(parents=True, exist_ok=True) env["CHECKUPDATES_DB"] = str(db) rc, lines = await self._run(["checkupdates"], env=env, stream=False) # checkupdates is inverted: 0 = updates found, 2 = nothing to do. if rc == 2: return [], "" if rc != 0: return [], "checkupdates_failed" return [l for l in lines if l.strip()], "" async def _check_aur(self): if not self.settings["enable_aur"]: return [], "" yay = _which("yay") if not yay: return [], "missing_yay" rc, lines = await self._run( [yay, "-Qua", "--color", "never"], env=_yay_env(), stream=False ) entries = [l for l in lines if l.strip() and "->" in l] if rc != 0 and not entries: return [], "" return entries, "" async def _check_flatpak(self): if not self.settings["enable_flatpak"]: return [], "" flatpak = _which("flatpak") if not flatpak: return [], "" entries = [] # Refreshing appstream data is what makes --updates accurate. await self._run( [flatpak, "update", "--system", "--appstream", "--noninteractive"], stream=False, ) rc, lines = await self._run( [ flatpak, "remote-ls", "--system", "--updates", "--columns=application,version", ], stream=False, ) if rc == 0: entries += [l for l in lines if l.strip()] user = getattr(decky, "DECKY_USER", None) home = getattr(decky, "DECKY_USER_HOME", "") if user and home and _which("runuser"): rc, lines = await self._run( _flatpak_user_cmd( flatpak, user, home, ["remote-ls", "--user", "--updates", "--columns=application,version"], ), stream=False, ) if rc == 0: entries += [l for l in lines if l.strip()] return entries, "" async def _check_fwupd(self): if not self.settings["enable_fwupd"]: return [], "" fwupdmgr = _which("fwupdmgr") if not fwupdmgr: return [], "" rc, lines = await self._run( [fwupdmgr, "get-updates", "--json"], stream=False, merge_stderr=False ) # With --json, "nothing to do" is exit 0 plus an empty Devices array; # older fwupd returns 2 (EXIT_NOTHING_TO_DO). Anything else is unusable. if rc not in (0, 2): return [], "" try: data = json.loads("\n".join(lines)) except ValueError: return [], "" entries = [] for dev in data.get("Devices", []): name = dev.get("Name", "?") for rel in dev.get("Releases", []): entries.append(f"{name} -> {rel.get('Version', '?')}") return entries, "" async def check_updates(self): """Look for updates without installing anything.""" if self.lock.locked(): return self.state.to_dict() async with self.lock: await self._set_status("checking") warnings = [] try: results = {} for key, coro in ( ("pacman", self._check_pacman()), ("aur", self._check_aur()), ("flatpak", self._check_flatpak()), ("fwupd", self._check_fwupd()), ): entries, warning = await coro results[key] = entries if warning: warnings.append(warning) self.state.details = results self.state.counts = {k: len(v) for k, v in results.items()} self.state.last_check = time.time() self.state.warnings = warnings self.state.error_text = "" self._save_state() except Exception as exc: decky.logger.exception("Update check failed") self.state.error_text = str(exc) finally: await self._set_status("idle") return self.state.to_dict() # -- updating ---------------------------------------------------------- def _enabled_phases(self): phases = ["keyring", "pacman"] if self.settings["enable_aur"] and _which("yay"): phases.append("aur") if self.settings["enable_flatpak"] and _which("flatpak"): phases.append("flatpak_system") if getattr(decky, "DECKY_USER", None) and _which("runuser"): phases.append("flatpak_user") if self.settings["enable_fwupd"] and _which("fwupdmgr"): phases.append("fwupd") return phases async def _phase_keyring(self, **kw): # Upgrading the keyring before the rest is the documented cure for the # classic "invalid or corrupted package (PGP signature)" failure. Only # valid as one atomic sequence with the -Su right after it, otherwise # it would be a partial upgrade. pkgs = [p for p in ("archlinux-keyring", "cachyos-keyring") if _installed(p)] cmd = ["pacman", "-Sy", "--noconfirm", "--color", "never"] if pkgs: cmd += ["--needed"] + pkgs rc, lines = await self._run(cmd, **kw) return rc == 0, lines async def _phase_pacman(self, **kw): rc, lines = await self._run( [ "pacman", "-Su", "--noconfirm", "--noprogressbar", "--color", "never", ], **kw, ) return rc == 0, lines async def _phase_aur(self, **kw): rc, lines = await self._run( [ _which("yay") or "yay", "-Syu", "--noconfirm", "--removemake", "--noprogressbar", "--color", "never", ], env=_yay_env(), **kw, ) return rc == 0, lines async def _phase_flatpak_system(self, **kw): # --system is not optional here: without an explicit scope flatpak # updates the system installation *and the calling user's* one, which # as root means it would start populating /root/.local/share/flatpak. # --noninteractive already implies -y and picks the quiet transaction. rc, lines = await self._run( [ _which("flatpak") or "flatpak", "update", "--system", "--noninteractive", ], **kw, ) return rc == 0, lines async def _phase_flatpak_user(self, **kw): rc, lines = await self._run( _flatpak_user_cmd( _which("flatpak") or "flatpak", decky.DECKY_USER, decky.DECKY_USER_HOME, ["update", "--user", "--noninteractive"], ), **kw, ) return rc == 0, lines async def _phase_fwupd(self, **kw): fwupdmgr = _which("fwupdmgr") or "fwupdmgr" # No --force: it would re-download the metadata on every single run, # which is wasteful on a battery-powered handheld. Fresh metadata # simply exits 2 (EXIT_NOTHING_TO_DO), which is a success here. rc, lines = await self._run([fwupdmgr, "refresh", "--assume-yes"], **kw) if rc == 101: # ENETUNREACH await self._log("!! Firmware server unreachable - skipped.") return True, lines rc, update_lines = await self._run( [ fwupdmgr, "update", "--assume-yes", "--no-reboot-check", "--no-unreported-check", "--no-remote-check", ], **kw, ) lines += update_lines # 0 = done, 2 = EXIT_NOTHING_TO_DO, 101 = network unreachable. return rc in (0, 2, 101), lines async def start_update(self, dry_run=False): if self.lock.locked(): return {"started": False, "reason_id": "already_running"} if not _is_root() and not dry_run: return {"started": False, "reason_id": "not_root"} if _pacman_busy() and not dry_run: return {"started": False, "reason_id": "pacman_busy"} asyncio.create_task(self._do_update(dry_run)) return {"started": True, "reason_id": ""} async def _do_update(self, dry_run=False): async with self.lock: self._dry_run = dry_run self.state.log.clear() self.state.error_text = "" self.state.hint_id = "" self.state.warnings = [] self.state.failed_phases = [] self.state.progress = 0.0 await self._set_status("updating") phases = self._enabled_phases() total_weight = sum(PHASE_WEIGHTS[p] for p in phases) done_weight = 0 all_output = [] handlers = { "keyring": self._phase_keyring, "pacman": self._phase_pacman, "aur": self._phase_aur, "flatpak_system": self._phase_flatpak_system, "flatpak_user": self._phase_flatpak_user, "fwupd": self._phase_fwupd, } try: for phase in phases: weight = PHASE_WEIGHTS[phase] await self._set_phase(phase, done_weight, total_weight) ok, lines = await handlers[phase]( phase_base=done_weight / total_weight, phase_span=weight / total_weight, ) all_output += lines done_weight += weight if not ok: self.state.failed_phases.append(phase) await self._log(f"!! Phase failed: {phase}") # The repo upgrade is the foundation - if it breaks, # building AUR packages on top of it is a bad idea. if phase in ("keyring", "pacman"): break self.state.progress = 1.0 _invalidate_installed_cache() self.state.reboot_required = self._needs_reboot(all_output) if "fwupd" in phases and not self.state.reboot_required: self.state.reboot_required = await self._firmware_reboot_pending() self.state.pacnew = await self._find_pacnew() self.state.last_update = time.time() if self.state.failed_phases: self.state.status = "error" self.state.hint_id = self._hint_for(all_output) else: self.state.status = "done" except Exception as exc: decky.logger.exception("Update failed") self.state.status = "error" self.state.error_text = str(exc) finally: self._dry_run = False await decky.emit("cachyos_update_state", self.state.status) await decky.emit( "cachyos_update_finished", self.state.status == "done", self.state.failed_phases, self.state.reboot_required, ) # Only a successful run means "nothing pending" - after a failure # the old counts stay, otherwise the panel would claim the system # is up to date while the updates are still sitting there. if not dry_run and self.state.status == "done": self.state.details = {k: [] for k in self.state.details} self.state.counts = {k: 0 for k in self.state.counts} self._save_state() # -- post-update analysis --------------------------------------------- def _hint_for(self, lines): blob = "\n".join(lines).lower() for marker, hint_id in ERROR_HINTS: if marker in blob: return hint_id return "" def _needs_reboot(self, output_lines): """True if the running kernel no longer matches what is installed. Primary signal is the same one cachy-update uses: the module tree of the running kernel still has its vmlinuz. Skipped when kernel-modules-hook or mkmm is installed, because those keep old module trees around and would make the test always pass. """ try: if not (_installed("kernel-modules-hook") or _installed("mkmm")): release = os.uname().release if not Path(f"/usr/lib/modules/{release}/vmlinuz").is_file(): return True except Exception as exc: decky.logger.warning("Kernel check failed: %s", exc) # Cross-check: did this run upgrade anything from the reboot list? upgraded = { match.group(1) for line in output_lines for match in [UPGRADING_RE.search(line)] if match } return any( pattern.match(pkg) for pkg in upgraded for pattern in REBOOT_TRIGGER_PATTERNS ) async def _firmware_reboot_pending(self): """Firmware is staged to the ESP and flashed on the next boot. `--no-reboot-check` suppresses the prompt but not the need, so ask fwupd explicitly. Exit 0 means a reboot IS pending, 2 means clean. """ fwupdmgr = _which("fwupdmgr") if not fwupdmgr or self._dry_run: return False rc, _ = await self._run( [fwupdmgr, "check-reboot-needed"], stream=False, merge_stderr=False ) return rc == 0 async def _find_pacnew(self): if not _which("pacdiff"): return [] env = _base_env() env["DIFFPROG"] = "/bin/true" rc, lines = await self._run(["pacdiff", "--output"], env=env, stream=False) if rc != 0: return [] return [l.strip() for l in lines if l.strip().startswith("/")] # -- frontend API ------------------------------------------------------ async def get_state(self): return self.state.to_dict() async def get_settings(self): return self.settings async def set_settings(self, settings): for key in DEFAULT_SETTINGS: if key in settings: self.settings[key] = settings[key] self._save_settings() return self.settings async def reboot(self): if self._dry_run: return True await self._run(["systemctl", "reboot"], stream=False) return True async def self_test(self): tools = {} # (binary, version args, substrings that identify the relevant line) for name, args, match in ( ("pacman", ["-Q", "pacman"], ()), ("yay", ["--version"], ()), ("paru", ["--version"], ()), ("checkupdates", None, ()), ("pacdiff", None, ()), ("flatpak", ["--version"], ()), # fwupdmgr prints a compile/runtime table, not a single line. ("fwupdmgr", ["--version"], ("runtime", "org.freedesktop.fwupd")), ("runuser", None, ()), ("systemd-run", ["--version"], ()), ("fakeroot", None, ()), ): path = _which(name) entry = {"found": bool(path), "path": path or "", "version": ""} if path and args: try: rc, lines = await self._run( [path] + args, stream=False, merge_stderr=False ) if rc == 0: for line in lines: text = " ".join(line.split()) if not text: continue if match: if all(m in text for m in match): entry["version"] = text[:80] break elif any(c.isdigit() for c in text): entry["version"] = text[:80] break except Exception: pass tools[name] = entry return { "root": _is_root(), "uid": os.geteuid(), "standalone": STANDALONE, "decky_user": getattr(decky, "DECKY_USER", ""), "decky_user_home": getattr(decky, "DECKY_USER_HOME", ""), "kernel": os.uname().release, "kernel_vmlinuz_present": Path( f"/usr/lib/modules/{os.uname().release}/vmlinuz" ).is_file(), "pacman_busy": _pacman_busy(), "tools": tools, "settings": self.settings, } # -- background check -------------------------------------------------- async def _auto_check_loop(self): # Let the session settle before hitting the network. await asyncio.sleep(30) while True: try: if self.settings["auto_check"]: await self.check_updates() total = sum(self.state.counts.values()) if total and self.settings["notify_on_updates"]: await decky.emit("cachyos_update_available", total) except asyncio.CancelledError: raise except Exception as exc: decky.logger.warning("Automatic check failed: %s", exc) hours = max(1, int(self.settings.get("check_interval_hours", 6))) await asyncio.sleep(hours * 3600) # -------------------------------------------------------------------------- # standalone entry point (development only) # -------------------------------------------------------------------------- async def _cli(): plugin = Plugin() plugin._load_settings() args = sys.argv[1:] if "--selftest" in args: result = await plugin.self_test() print(json.dumps(result, indent=2, ensure_ascii=False)) if not result["root"]: print("\nNote: not running as root - expected during development.") return if "--check" in args: state = await plugin.check_updates() print(json.dumps( { "counts": state["counts"], "total": state["total"], "warnings": state["warnings"], "details": state["details"], }, indent=2, ensure_ascii=False, )) return if "--update" in args: dry = "--dry-run" in args if not dry and not _is_root(): print("A real update needs root. Use --dry-run to test.") return if dry: # Enable every phase so the dry run shows all command lines. plugin.settings["enable_fwupd"] = True await plugin._do_update(dry_run=dry) print(f"\nStatus: {plugin.state.status}") if plugin.state.failed_phases: print(f"Failed phases: {', '.join(plugin.state.failed_phases)}") if plugin.state.error_text: print(f"Error: {plugin.state.error_text}") return print(__doc__) if __name__ == "__main__": asyncio.run(_cli())