"""Textual app for managing Proxmox guests through cv4pve-cli.""" import subprocess import sys from datetime import datetime from rich.text import Text from textual import work from textual.app import App, ComposeResult, SuspendNotSupported from textual.binding import Binding from textual.containers import Horizontal, Vertical, VerticalScroll from textual.screen import ModalScreen from textual.widgets import ( Button, Checkbox, DataTable, Footer, Header, Input, Label, Log, OptionList, Select, Static, ) from textual.widgets._data_table import RowDoesNotExist from textual.widgets.option_list import Option from cv4pve_tui import cli def _message(exc: Exception) -> str: return ( exc.summary if isinstance(exc, cli.CliError) else f"{type(exc).__name__}: {exc}" ) class DialogScreen(ModalScreen): """A centred, titled box (#dialog) with labelled rows and right-aligned buttons.""" # every widget in a row gets an explicit width: Horizontal alone squeezes Select to 1 # and stretches Input to 100 %, pushing the rest off-screen DEFAULT_CSS = """ DialogScreen { align: center middle; } DialogScreen #dialog { width: 80; max-width: 100%; height: auto; max-height: 100%; overflow-y: auto; border: round $accent; border-title-style: bold; padding: 0 1; background: $surface; } DialogScreen .row, DialogScreen .buttons { height: auto; } DialogScreen .buttons { align-horizontal: right; } DialogScreen .row Label { width: 14; height: 3; content-align: left middle; } DialogScreen .row Select, DialogScreen .row Input { width: 1fr; } DialogScreen .row Select.mode { width: 16; } DialogScreen .sub { height: auto; border: round $accent; border-title-style: bold; padding: 0 1; } """ def box(self, title: str) -> Vertical: """The #dialog container; use as `with self.box(title):` inside compose.""" box = Vertical(id="dialog") # titles carry guest names: never parse them as markup box.border_title = Text(title) return box class ConfirmScreen(DialogScreen): BINDINGS = [("escape", "dismiss(False)", "No")] # noqa: RUF012 - Textual idiom def __init__(self, message: str, title: str = "Confirm") -> None: super().__init__() self.message, self.title_text = message, title def compose(self) -> ComposeResult: with self.box(self.title_text): yield Static(self.message, markup=False) with Horizontal(classes="buttons"): yield Button("Yes", id="yes", variant="error") yield Button("No", id="no") def on_button_pressed(self, event: Button.Pressed) -> None: self.dismiss(event.button.id == "yes") class EditScreen(DialogScreen): """Dismisses with a dict of validated values, or None when cancelled.""" BINDINGS = [("escape", "dismiss(None)", "Cancel")] # noqa: RUF012 - Textual idiom def __init__(self, guest: dict, config: dict, disks: dict[str, str]) -> None: super().__init__() self.guest, self.config, self.disks = guest, config, disks def compose(self) -> ComposeResult: with self.box(f"Edit resources: {self.guest['name']} ({self.guest['vmid']})"): yield Static("Empty = unchanged") with Horizontal(classes="row"): yield Label("Cores") yield Input( placeholder=f"now {self.config.get('cores', '?')}", id="cores" ) with Horizontal(classes="row"): yield Label("Memory (MiB)") yield Input( placeholder=f"now {self.config.get('memory', '?')}", id="memory" ) with Horizontal(classes="row"): yield Label("Disk") yield Select( [(f"{k} ({v})", k) for k, v in self.disks.items()], prompt="disk", id="disk", ) with Horizontal(classes="row"): yield Label("Grow by (GiB)") yield Input(placeholder="grow by GiB", id="grow") with Horizontal(classes="buttons"): yield Button("Apply", id="apply", variant="primary") yield Button("Cancel", id="cancel") def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "cancel": self.dismiss(None) return try: values = {name: self._number(name) for name in ("cores", "memory", "grow")} except ValueError as exc: self.app.notify(str(exc), severity="error") return disk = self.query_one("#disk", Select).value if values["grow"] is not None and disk is Select.NULL: self.app.notify("Choose a disk to grow", severity="error") return values["disk"] = None if disk is Select.NULL else disk self.dismiss(values) def _number(self, name: str) -> int | None: text = self.query_one(f"#{name}", Input).value.strip() if not text: return None if not text.isdecimal() or int(text) <= 0: raise ValueError(f"{name} must be a positive integer") return int(text) class ContextScreen(DialogScreen): """Dismisses with the chosen context name, or None when cancelled.""" BINDINGS = [("escape", "dismiss(None)", "Cancel")] # noqa: RUF012 - Textual idiom def __init__(self, contexts: list[dict]) -> None: super().__init__() self.contexts = contexts def compose(self) -> ComposeResult: with self.box("Switch context"): yield OptionList( *( Option( f"{'*' if c['current'] else ' '} {c['name']} {c['host']}", id=c["name"], ) for c in self.contexts ), id="contexts", ) def on_option_list_option_selected(self, event: OptionList.OptionSelected) -> None: self.dismiss(event.option.id) class BackupScreen(DialogScreen): """Dismisses with the chosen storage, or None when cancelled.""" BINDINGS = [("escape", "dismiss(None)", "Cancel")] # noqa: RUF012 - Textual idiom def __init__(self, guest: dict, storages: list[str], chosen: str) -> None: super().__init__() self.guest, self.storages, self.chosen = guest, storages, chosen def compose(self) -> ComposeResult: with self.box(f"Back up {self.guest['name']} ({self.guest['vmid']})"): yield Static( f"Back up {self.guest['name']} ({self.guest['vmid']}) now: mode snapshot, zstd", markup=False, ) with Horizontal(classes="row"): yield Label("Storage") yield Select( [(s, s) for s in self.storages], value=self.chosen, allow_blank=False, id="storage", ) with Horizontal(classes="buttons"): yield Button("Yes", id="yes", variant="primary") yield Button("No", id="no") def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "yes": self.dismiss(self.query_one("#storage", Select).value) else: self.dismiss(None) class SnapshotScreen(DialogScreen): """Dismisses with {name, description, vmstate}, or None when cancelled.""" BINDINGS = [("escape", "dismiss(None)", "Cancel")] # noqa: RUF012 - Textual idiom def __init__(self, guest: dict) -> None: super().__init__() self.guest = guest def compose(self) -> ComposeResult: with self.box(f"Take snapshot: {self.guest['name']} ({self.guest['vmid']})"): with Horizontal(classes="row"): yield Label("Name") yield Input(datetime.now().strftime("tui_%Y%m%d_%H%M"), id="snapname") with Horizontal(classes="row"): yield Label("Description") yield Input(placeholder="optional", id="snapdesc") if cli.kind(self.guest) == "vm": yield Checkbox("Include RAM (vmstate)", id="vmstate") with Horizontal(classes="buttons"): yield Button("Yes", id="yes", variant="primary") yield Button("No", id="no") def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id != "yes": self.dismiss(None) return try: name = cli.valid_snapname(self.query_one("#snapname", Input).value) description = cli.valid_description( self.query_one("#snapdesc", Input).value.strip() ) except ValueError as exc: self.app.notify(str(exc), severity="error") return vmstate = next(iter(self.query("#vmstate").results(Checkbox)), None) self.dismiss( { "name": name, "description": description, "vmstate": bool(vmstate and vmstate.value), } ) _IPV4_MODES = [("DHCP", "dhcp"), ("Static", "static"), ("None", "none")] _IPV6_MODES = [("SLAAC", "auto"), *_IPV4_MODES] class NetScreen(DialogScreen): """Edit bridge, VLAN, firewall and IP of one interface; confirms, then applies.""" BINDINGS = [("escape", "dismiss", "Cancel")] # noqa: RUF012 - Textual idiom def __init__( self, guest: dict, config: dict, bridges: list[str], running: bool ) -> None: super().__init__() self.guest, self.config, self.running = guest, config, running self.nets = cli.nets(config) self.is_vm = cli.kind(guest) == "vm" self.ip_enabled = not self.is_vm or cli.has_cloudinit(config) # a bridge the node no longer lists must stay selectable, or Select rejects the prefill used = {n.get("bridge") for n in self.nets.values()} - {None} self.bridges = sorted(set(bridges) | used) def compose(self) -> ComposeResult: with self.box(f"Edit network: {self.guest['name']} ({self.guest['vmid']})"): with Horizontal(classes="row"): yield Label("Interface") yield Select( [(k, k) for k in self.nets], value=next(iter(self.nets)), allow_blank=False, id="iface", ) with Horizontal(classes="row"): yield Label("Bridge") yield Select( [(b, b) for b in self.bridges], allow_blank=False, id="bridge" ) with Horizontal(classes="row"): yield Label("VLAN") yield Input(placeholder="VLAN tag (empty = none)", id="tag") with Horizontal(classes="row"): yield Label("Firewall") yield Checkbox("", id="firewall") if not self.ip_enabled: yield Static( "IP is configured inside the VM (no cloud-init drive)", id="ip_note" ) for fam, modes in (("ipv4", _IPV4_MODES), ("ipv6", _IPV6_MODES)): with Vertical(id=f"{fam}_section", classes="sub") as section: section.border_title = fam.replace("ip", "IP") with Horizontal(classes="row"): yield Select(modes, allow_blank=False, id=fam, classes="mode") yield Input(placeholder="address/prefix", id=f"{fam}_cidr") yield Input(placeholder="gateway", id=f"{fam}_gw") with Horizontal(classes="buttons"): yield Button("Apply", id="apply", variant="primary") yield Button("Cancel", id="cancel") def on_mount(self) -> None: for wid in ("ipv4", "ipv4_cidr", "ipv4_gw", "ipv6", "ipv6_cidr", "ipv6_gw"): self.query_one(f"#{wid}").disabled = not self.ip_enabled self._prefill() def on_select_changed(self, event: Select.Changed) -> None: if event.select.id == "iface": self._prefill() def _ip_cfg(self, iface: str) -> dict: """The dict holding the IP keys: the VM's ipconfigN, or the LXC net itself.""" if self.is_vm: return cli.parse_net(str(self.config.get(f"ipconfig{iface[3:]}", ""))) return self.nets[iface] def _prefill(self) -> None: iface = self.query_one("#iface", Select).value net = self.nets[iface] self.query_one("#bridge", Select).value = net.get("bridge", self.bridges[0]) self.query_one("#tag", Input).value = net.get("tag", "") self.query_one("#firewall", Checkbox).value = net.get("firewall") == "1" for fam, family in (("ipv4", 4), ("ipv6", 6)): mode, cidr, gw = cli.ip_mode(self._ip_cfg(iface), family) self.query_one(f"#{fam}", Select).value = mode self.query_one(f"#{fam}_cidr", Input).value = cidr self.query_one(f"#{fam}_gw", Input).value = gw def _changes(self) -> tuple[dict, dict]: """(changes, old) as raw netN/ipconfigN strings, None = absent; ValueError on bad input.""" iface = self.query_one("#iface", Select).value text = self.query_one("#tag", Input).value.strip() if text and not text.isdecimal(): raise ValueError("VLAN tag must be 1-4094") current = self.nets[iface] link = ( self.query_one("#bridge", Select).value, int(text) if text else None, self.query_one("#firewall", Checkbox).value, ) was = (current.get("bridge"), int(current["tag"]) if current.get("tag") else None, current.get("firewall") == "1") # fmt: skip # untouched fields are not rewritten (e.g. firewall=0 stays as it is) net = cli.apply_link(current, *link) if link != was else dict(current) ip_key = f"ipconfig{iface[3:]}" ip_old = self._ip_cfg(iface) if self.is_vm else {} ip = ip_old if self.is_vm else net # the dict that holds the IP keys if self.ip_enabled: for fam, family in (("ipv4", 4), ("ipv6", 6)): mode, cidr, gw = ( self.query_one(f"#{fam}{s}").value for s in ("", "_cidr", "_gw") ) wanted = (mode, cidr.strip(), gw.strip()) if wanted != cli.ip_mode( ip, family ): # untouched: e.g. ip6=manual stays ip = cli.apply_ip(ip, family, *wanted) if not self.is_vm: net = ip changes, old = {}, {} if net != self.nets[iface]: changes[iface] = cli.format_net(net) old[iface] = str(self.config[iface]) if self.is_vm and ip != ip_old: changes[ip_key] = cli.format_net(ip) or None old[ip_key] = self.config.get(ip_key) return changes, old def on_button_pressed(self, event: Button.Pressed) -> None: if event.button.id == "cancel": self.dismiss() return try: changes, old = self._changes() except ValueError as exc: self.app.notify(str(exc), severity="error") return if not changes: self.app.notify("No change") return lines = [ f"{k}\n from {old[k] or '(not set)'}\n to {v or '(removed)'}" for k, v in changes.items() ] notes = [] if self.running and any(k.startswith("net") for k in changes): notes.append( "Link changes apply immediately and can cut the guest's network." ) if ( self.running and not self.is_vm and any(_ip_changed(old[k], v) for k, v in changes.items()) ): # seen live: static -> dhcp left the CT without IPv4 until it was rebooted notes.append( "IP changes on a running container may need a reboot of the container " "(e.g. DHCP is not restarted)." ) if any(k.startswith("ipconfig") for k in changes): notes.append("IP settings (cloud-init) apply at next boot.") who = f"{self.guest['name']} ({self.guest['vmid']})" self.app.push_screen( ConfirmScreen( "\n".join([f"Change network of {who}?", *lines, *notes]), "Network change", ), lambda ok: ( self.app.start_net_change(self.guest, changes, old, self) if ok else None ), ) class TaskLogScreen(DialogScreen): """The log of one task, as plain text (task output is never parsed as markup).""" BINDINGS = [("escape", "dismiss", "Close")] # noqa: RUF012 - Textual idiom DEFAULT_CSS = """ TaskLogScreen #dialog { width: 120; height: 90%; } TaskLogScreen Log { height: 1fr; } """ def __init__(self, entry: dict) -> None: super().__init__() self.entry = entry def compose(self) -> ComposeResult: t = self.entry with self.box(f"Log: {t.get('type', '')} {t.get('id') or t.get('node', '')}"): yield Log(id="tasklog") def on_mount(self) -> None: self._load_worker() @work(thread=True) def _load_worker(self) -> None: try: text = cli.task_log(self.entry["upid"]) except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.app.call_from_thread(self.app.notify, _message(exc), severity="error") return self.app.call_from_thread(self.query_one("#tasklog", Log).write, text) def _duration(task: dict) -> str: end, start = task.get("endtime"), task.get("starttime") if not end: return "running" if not start: return "-" # finished, but the start is unknown secs = max(0, int(end - start)) if secs >= 3600: return f"{secs // 3600}h {secs % 3600 // 60}m" return f"{secs // 60}m {secs % 60}s" if secs >= 60 else f"{secs}s" class TasksScreen(ModalScreen): """Recent cluster tasks, or those of one guest (vmid); Enter shows a task's log.""" BINDINGS = [ # noqa: RUF012 - Textual idiom ("escape", "dismiss", "Close"), ("g", "reload", "Refresh"), ] DEFAULT_CSS = """ TasksScreen { align: center middle; } TasksScreen #tasks_section { width: 100%; height: 100%; border: round $accent; border-title-style: bold; padding: 0 1; background: $surface; } TasksScreen DataTable { height: 1fr; } """ def __init__(self, guest: dict | None = None) -> None: super().__init__() self.guest = guest def compose(self) -> ComposeResult: title = "Tasks" if self.guest: title += f": {self.guest['name']} ({self.guest['vmid']})" with Vertical(id="tasks_section") as section: section.border_title = Text(title) # guest names: never markup yield DataTable(id="tasks", cursor_type="row") yield Static("", id="tasks_note", markup=False) yield Footer() def on_mount(self) -> None: self.rows: list[dict] = [] self.query_one("#tasks", DataTable).add_columns( "", "Start", "Duration", "Node", "Type", "VMID", "User", "Status" ) self.action_reload() def action_reload(self) -> None: self._load_worker() @work(thread=True) def _load_worker(self) -> None: try: tasks = cli.tasks() except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.app.call_from_thread(self.app.notify, _message(exc), severity="error") return if self.guest: tasks = [t for t in tasks if t.get("id") == str(self.guest["vmid"])] self.app.call_from_thread(self._fill, tasks) def _fill(self, tasks: list[dict]) -> None: self.rows = tasks table = self.query_one("#tasks", DataTable) table.clear() for t in tasks: # running = no endtime yet; a finished task without a status text is "unknown" status = "running" if not t.get("endtime") else t.get("status") or "unknown" _add_plain( table, "!" if status not in ("OK", "running") else "", _when(t.get("starttime")), _duration(t), t.get("node", ""), t.get("type", ""), t.get("id") or "-", t.get("user", ""), status, ) self.query_one("#tasks_note", Static).update("" if tasks else "No tasks") def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: self.app.push_screen(TaskLogScreen(self.rows[event.cursor_row])) def _collect_updates() -> tuple[list[dict], list[str]]: """Runs in a worker thread: (rows over all online nodes, one note per failed node).""" rows, notes = [], [] for node in cli.nodes(): try: rows += [{**p, "node": node} for p in cli.package_updates(node)] except Exception as exc: # noqa: BLE001 - a failing node must not hide the others notes.append(f"{node}: unavailable: {_message(exc)}") return rows, notes class UpdatesScreen(ModalScreen): """Pending package updates of every online node (read-only).""" BINDINGS = [ # noqa: RUF012 - Textual idiom ("escape", "dismiss", "Close"), ("g", "reload", "Refresh"), ] DEFAULT_CSS = """ UpdatesScreen { align: center middle; } UpdatesScreen #updates_section { width: 100%; height: 100%; border: round $accent; border-title-style: bold; padding: 0 1; background: $surface; } UpdatesScreen DataTable { height: 1fr; } """ def compose(self) -> ComposeResult: with Vertical(id="updates_section") as section: section.border_title = "Package updates" yield DataTable(id="updates", cursor_type="row") yield Static("", id="updates_note", markup=False) yield Footer() def on_mount(self) -> None: self.query_one("#updates", DataTable).add_columns( "Node", "Package", "Current", "New", "Origin", "Priority" ) self.action_reload() def action_reload(self) -> None: self._load_worker() @work(thread=True) def _load_worker(self) -> None: gen = self.app.generation try: rows, notes = _collect_updates() except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.app.call_from_thread(self.app.notify, _message(exc), severity="error") return self.app.call_from_thread(self._fill, rows, notes) self.app.call_from_thread( self.app._set_count, None if notes else len(rows), gen ) def _fill(self, rows: list[dict], notes: list[str]) -> None: table = self.query_one("#updates", DataTable) table.clear() for p in rows: _add_plain( table, p["node"], p.get("Package", ""), p.get("OldVersion", ""), p.get("Version", ""), p.get("Origin", ""), p.get("Priority", ""), ) lines = [] if rows else ["No pending updates"] lines += notes lines.append("Lists are as of each node's last apt update; it is not run here.") self.query_one("#updates_note", Static).update("\n".join(lines)) class InfoScreen(ModalScreen): """Backups, backup job and network of one guest (read-only), plus backup now.""" BINDINGS = [ # noqa: RUF012 - Textual idiom ("escape", "dismiss", "Close"), ("g", "reload", "Refresh"), ("b", "backup", "Backup now"), ("a", "toggle_all", "All backups"), ("n", "net", "Edit network"), ("u", "undo", "Undo network"), ("e", "edit", "Edit resources"), ("p", "snapshot", "Take snapshot"), ("R", "rollback", "Roll back"), ("D", "delete_snapshot", "Delete snapshot"), ("l", "tasks", "Tasks"), ] DEFAULT_CSS = """ InfoScreen DataTable { height: auto; max-height: 12; } InfoScreen .section { height: auto; border: round $accent; border-title-style: bold; padding: 0 1; margin-bottom: 1; } InfoScreen .label { text-style: bold; margin-top: 1; } InfoScreen .label.first { margin-top: 0; } """ def __init__(self, guest: dict) -> None: super().__init__() self.guest = guest self.storages: list[str] = [] self.jobs: list[dict] = [] self.backups: list[dict] = [] self.show_all = False self.config: dict = {} self.running = False self.snaps: list[dict] = [] def compose(self) -> ComposeResult: with VerticalScroll(id="info_body"): # four sections do not fit short terminals yield Static(f"{self.guest['name']} ({self.guest['vmid']})", markup=False) with Vertical(id="resources_section", classes="section") as res: res.border_title = "Resources" yield DataTable(id="resources", cursor_type="none") with Vertical(id="backup_section", classes="section") as backup: backup.border_title = "Backups" yield Static("", id="jobs", markup=False) yield DataTable(id="backups") yield Static("", id="hidden", markup=False) with Vertical(id="snapshots_section", classes="section") as snaps: snaps.border_title = "Snapshots" yield DataTable(id="snapshots", cursor_type="row") yield Static("", id="snap_note", markup=False) with Vertical(id="network_section", classes="section") as network: network.border_title = "Network" yield Static("Configured", classes="label first") yield DataTable(id="nets") yield Static("In use", classes="label") yield Static("", id="unavailable", markup=False) yield DataTable(id="addresses") yield Footer() def on_mount(self) -> None: self.query_one("#resources", DataTable).add_columns( "Resource", "Configured", "In use" ) self.query_one("#backups", DataTable).add_columns( "Date", "Storage", "Size (GiB)", "Format", "Notes" ) self.query_one("#snapshots", DataTable).add_columns( "Name", "Date", "Description", *(["RAM"] if cli.kind(self.guest) == "vm" else []), ) self.query_one("#nets", DataTable).add_columns( "Iface", "Bridge", "VLAN", "MAC", "IPv4", "IPv6", "Firewall" ) self.query_one("#addresses", DataTable).add_columns( "Interface", "MAC", "IPv4", "IPv6" ) self.action_reload() def action_reload(self) -> None: self._load_worker() def action_tasks(self) -> None: self.app.push_screen(TasksScreen(self.guest)) @work(thread=True) def _load_worker(self) -> None: self.load() def load(self) -> None: """Runs in a worker thread.""" guest = self.guest try: config = cli.guest_config(guest) nets = cli.nets(config) storages = cli.backup_storages(guest) jobs = cli.backup_jobs() backups = cli.backups(guest, storages) except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.app.call_from_thread(self.app.notify, _message(exc), severity="error") return addresses, unavailable, usage = [], "", {} try: # live status: the guest table's status may be stale usage = cli.current(guest) running = usage.get("status") == "running" except Exception as exc: # noqa: BLE001 - shown inline, the rest of the screen still works running, unavailable = False, f"unavailable: {_message(exc)}" if running: try: addresses = cli.interfaces(guest) except Exception as exc: # noqa: BLE001 - shown inline, the rest of the screen still works unavailable = f"unavailable: {_message(exc)}" snaps, snap_note = [], "" try: snaps = cli.snapshots(guest) except Exception as exc: # noqa: BLE001 - shown inline, the rest of the screen still works snap_note = f"unavailable: {_message(exc)}" self.running = running self.app.call_from_thread( self._show, (config, nets, storages, jobs, backups, addresses, unavailable, usage), snaps, snap_note, ) def _show(self, loaded: tuple, snaps: list[dict], snap_note: str) -> None: if not self.is_attached: # closed while loading return try: self._fill_screen(*loaded) self._show_snapshots(snaps, snap_note) except Exception as exc: # noqa: BLE001 - odd API data must not kill the app self.app.notify(f"info: {_message(exc)}", severity="error") def _fill_screen( self, config, nets, storages, jobs, backups, addresses, unavailable, usage ) -> None: self.storages, self.jobs, self.config = storages, jobs, config covering = cli.covering_jobs(jobs, self.guest) self.query_one("#jobs", Static).update( "; ".join(_job_text(j) for j in covering) or "No backup job covers this guest" ) table = self.query_one("#resources", DataTable).clear() in_use = _in_use(usage, cli.kind(self.guest)) if self.running else {} for name, value in _resources(config, cli.kind(self.guest)): _add_plain(table, name, value, in_use.get(name, "-")) self.backups = backups self._show_backups() table = self.query_one("#nets", DataTable).clear() for name, net in nets.items(): if cli.kind(self.guest) == "vm": # IP settings are cloud-init's ipconfigN mac = next(iter(net.values()), "-") ip = cli.parse_net(str(config.get(f"ipconfig{name[3:]}", ""))) else: mac, ip = net.get("hwaddr", "-"), net _add_plain( table, name, net.get("bridge", "-"), net.get("tag", "-"), mac, ip.get("ip", "-"), ip.get("ip6", "-"), "yes" if net.get("firewall") == "1" else "no", ) note = self.query_one("#unavailable", Static) note.update(unavailable) note.display = bool(unavailable) # no empty line when the addresses are known table = self.query_one("#addresses", DataTable).clear() for i in addresses: _add_plain( table, i["name"], i["mac"] or "-", ", ".join(i["ipv4"]), ", ".join(i["ipv6"]), ) def _show_snapshots(self, snaps: list[dict], note: str) -> None: self.snaps = snaps table = self.query_one("#snapshots", DataTable).clear() vm = cli.kind(self.guest) == "vm" for s in snaps: _add_plain( table, s["name"], _when(s.get("snaptime")), s.get("description") or "", *(["yes" if s.get("vmstate") else "no"] if vm else []), ) self.query_one("#snap_note", Static).update( note or ("" if snaps else "No snapshots") ) def _show_backups(self) -> None: name = self.guest["name"] shown = [ b for b in self.backups if self.show_all or not _earlier_guest(b, name) ] table = self.query_one("#backups", DataTable).clear() for b in shown: _add_plain( table, datetime.fromtimestamp(b["ctime"]).strftime("%Y-%m-%d %H:%M") if b.get("ctime") else "-", b["storage"], f"{b['size'] / 2**30:.1f}" if b.get("size") is not None else "-", b.get("format") or "-", (b.get("notes") or "") + (" (earlier guest)" if _earlier_guest(b, name) else ""), ) hidden = len(self.backups) - len(shown) self.query_one("#hidden", Static).update( f"{hidden} backup(s) of an earlier guest with this VMID hidden - press a to show" if hidden else "" ) def action_edit(self) -> None: self.app.edit_guest(self.guest) def action_toggle_all(self) -> None: self.show_all = not self.show_all self._show_backups() def _refuse_if_busy(self) -> bool: busy = self.guest["vmid"] in self.app.busy if busy: self.app.notify( f"{self.guest['name']} ({self.guest['vmid']}) is busy", severity="warning", ) return busy def action_net(self) -> None: if cli.nets(self.config) and not self._refuse_if_busy(): self._open_net() @work(thread=True) def _open_net(self) -> None: try: bridges = cli.bridges(self.guest) except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.app.call_from_thread(self.app.notify, _message(exc), severity="error") return self.app.call_from_thread( self.app.push_screen, NetScreen(self.guest, self.config, bridges, self.running), ) def action_undo(self) -> None: old = self.app.undo.get(self.guest["vmid"]) if old is None: self.app.notify("No network change to undo") return if self._refuse_if_busy(): return lines = [f"{k} = {v or '(remove)'}" for k, v in old.items()] self.app.push_screen( ConfirmScreen( "\n".join(["Restore the previous network settings?", *lines]), "Undo network change", ), lambda ok: self.app.start_net_change(self.guest, old, None) if ok else None, ) def _snapshot_under_cursor(self) -> dict | None: """The snapshot at the cursor; None (with a notify) when there is none or the guest is busy.""" row = self.query_one("#snapshots", DataTable).cursor_row if not 0 <= row < len(self.snaps): self.app.notify("No snapshots") return None return None if self._refuse_if_busy() else self.snaps[row] def action_snapshot(self) -> None: if self._refuse_if_busy(): return self.app.push_screen(SnapshotScreen(self.guest), self._take_snapshot) def _take_snapshot(self, values: dict | None) -> None: if values: self.app.start_job( self.guest, "Snapshot", lambda: cli.take_snapshot( self.guest, values["name"], values["description"], values["vmstate"], ), ) def action_rollback(self) -> None: snap = self._snapshot_under_cursor() if snap is None: return text = ( f"Roll back to snapshot {snap['name']} ({_when(snap.get('snaptime'))})? " "Everything changed since then is lost. " "If the guest is running, it is started again afterwards." ) def job() -> None: # status read right before the rollback (not at screen load); if it cannot be # read the rollback does not run, rather than leave a running guest stopped start = cli.status(self.guest) == "running" cli.rollback_snapshot(self.guest, snap["name"], start) self.app.push_screen( ConfirmScreen(text, "Roll back"), lambda ok: ( self.app.start_job( self.guest, "Rollback", job, ) if ok else None ), ) def action_delete_snapshot(self) -> None: snap = self._snapshot_under_cursor() if snap is None: return self.app.push_screen( ConfirmScreen(f"Delete snapshot {snap['name']}?", "Delete snapshot"), lambda ok: ( self.app.start_job( self.guest, "Delete snapshot", lambda: cli.delete_snapshot(self.guest, snap["name"]), ) if ok else None ), ) def action_backup(self) -> None: if not self.storages: self.app.notify("No backup storage available", severity="warning") return chosen = cli.job_storage(self.jobs, self.guest) if chosen not in self.storages: chosen = self.storages[0] self.app.push_screen( BackupScreen(self.guest, self.storages, chosen), lambda storage: ( self.app.start_backup(self.guest, storage) if storage else None ), ) def _when(epoch: int | None) -> str: return datetime.fromtimestamp(epoch).strftime("%Y-%m-%d %H:%M") if epoch else "-" def _ip_changed(old: str | None, new: str | None) -> bool: before, after = cli.parse_net(old or ""), cli.parse_net(new or "") return any(before.get(k) != after.get(k) for k in ("ip", "gw", "ip6", "gw6")) def _earlier_guest(backup: dict, name: str) -> bool: """VMIDs are reused: notes (the job's {{guestname}}) naming another guest mean a deleted one.""" notes = backup.get("notes") or "" return bool(notes) and notes != name and not notes.startswith(f"{name} ") def _job_text(job: dict) -> str: keep = (job.get("prune-backups") or {}).get("keep-last") return f"{job['schedule']} to {job['storage']}" + ( f", keep-last {keep}" if keep else "" ) _TYPES = {"lxc": "lxc", "ct": "lxc", "vm": "qemu", "qemu": "qemu"} def guest_matches(guest: dict, query: str) -> bool: """True if the guest matches every space-separated term (case-insensitive). Plain term: substring of the name, the VMID or one tag. `is:running|stopped` matches the status, `type:lxc|ct|vm|qemu` the type; an unknown value matches nothing. """ tags = (guest.get("tags") or "").lower().split(";") for term in query.lower().split(): key, _, value = term.partition(":") if key == "is" and value: ok = (guest.get("status") or "").lower() == value elif key == "type" and value: ok = _TYPES.get(value) == (guest.get("type") or "").lower() else: ok = ( term in str(guest.get("name", "")).lower() or term in str(guest.get("vmid", "")) or any(term in t for t in tags) ) if not ok: return False return True class GuestTable(DataTable): """The guest table; shows its Enter (row selected -> info screen) in the footer.""" BINDINGS = [Binding("enter", "select_cursor", "Info")] # noqa: RUF012 - Textual idiom class FilterInput(Input): """The filter box; Escape clears and hides it.""" BINDINGS = [Binding("escape", "app.clear_filter", "Clear filter")] # noqa: RUF012 - Textual idiom class GuestApp(App): BINDINGS = [ # noqa: RUF012 - Textual idiom ("s", "act('start')", "Start"), ("S", "act('shutdown')", "Shutdown"), ("x", "act('stop')", "Stop"), ("r", "act('reboot')", "Reboot"), ("t", "shell", "Shell"), ("g", "reload", "Refresh"), ("c", "context", "Context"), ("slash", "filter", "Filter"), ("l", "tasks", "Tasks"), ("U", "updates", "Updates"), ("q", "quit", "Quit"), # also on the table, once a filter was applied with Enter (focus left the input) Binding("escape", "clear_filter", "Clear filter", show=False), ] def __init__(self) -> None: super().__init__() self.guests: dict[str, dict] = {} self.busy: set[int] = set() self.undo: dict[int, dict] = {} # vmid -> values before the last network change # bumped on every context switch; replies started earlier come from the old cluster self.generation = 0 self.switching = False self.context_name = "" self.unprotected: set[int] = set() # VMIDs no backup job covers self.package_count: int | None = None # pending package updates; None = unknown CSS = """ #guests_section { border: round $accent; border-title-style: bold; padding: 0 1; } #guests_section DataTable { height: 1fr; } #filter { display: none; } """ def compose(self) -> ComposeResult: yield Header() with Vertical(id="guests_section") as section: section.border_title = "Guests" yield GuestTable(id="guests", cursor_type="row") yield FilterInput( id="filter", placeholder="name, vmid, tag · is:running · type:lxc" ) yield Footer() def on_mount(self) -> None: # one place for every sub_title change: mount, switch and failed switch self.watch(self, "sub_title", self._show_context_frame) self._check_version() table = self.query_one("#guests", DataTable) table.add_columns( "", # "!" = no backup job covers the guest "VMID", "Name", "Type", "Node", "Status", "Cores", "Memory (MiB)", "Disk (GiB)", "CPU %", "Mem %", ) self._show_context() self.action_reload() self._count_worker() def action_tasks(self) -> None: self.push_screen(TasksScreen()) def action_updates(self) -> None: self.push_screen(UpdatesScreen()) @work(thread=True) def _count_worker(self) -> None: self._refresh_count() def _refresh_count(self) -> None: """Runs in a worker thread: any failure leaves the count unknown, silently.""" gen = self.generation try: rows, notes = _collect_updates() count = None if notes else len(rows) except Exception: # noqa: BLE001 - the count is optional; no notify on failure count = None self.call_from_thread(self._set_count, count, gen) def _set_count(self, count: int | None, gen: int) -> None: if gen != self.generation: # fetched before a context switch: old cluster return self.package_count = count self._update_frame() @work(thread=True) def _check_version(self) -> None: try: found = cli.version() except Exception as exc: # noqa: BLE001 - every later call fails too; say why once self.call_from_thread( self.notify, f"cv4pve-cli is not runnable: {_message(exc)}", severity="error", ) return if found != cli.TESTED_VERSION: self.call_from_thread( self.notify, f"cv4pve-cli {found} found; this app was tested with {cli.TESTED_VERSION}. " "Some screens may misread its output.", severity="warning", ) def _show_context_frame(self, sub_title: str) -> None: self.context_name = sub_title.removeprefix("context: ") self._update_frame() def _update_frame(self) -> None: n = len(self.unprotected & {g["vmid"] for g in self.guests.values()}) count = f" · {n} without backup job" if n else "" query = self.query_one("#filter", Input).value.strip() if query: shown = self.query_one("#guests", DataTable).row_count count = f" · filter: {query} · {shown} of {len(self.guests)}" + count if self.package_count: n = self.package_count count += f" · {n} package update{'s' if n != 1 else ''}" # plain text: the filter is typed by the user and must never be parsed as markup self.query_one("#guests_section").border_subtitle = Text( self.context_name + count ) @work(thread=True) def _show_context(self) -> None: self._read_context() def _read_context(self) -> None: """Runs in a worker thread: show the CLI's current context in the Header.""" gen = self.generation try: name = next(c["name"] for c in cli.contexts() if c["current"]) except StopIteration: name = "?" except Exception as exc: # noqa: BLE001 - any failure must reach the user, not kill the app self.call_from_thread(self.notify, _message(exc), severity="error") name = "?" self.call_from_thread(self._set_context_title, name, gen) def _set_context_title(self, name: str, gen: int) -> None: if gen == self.generation: self.sub_title = f"context: {name}" def _selected(self) -> dict | None: table = self.query_one("#guests", DataTable) if not table.row_count: return None return self.guests[ table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value ] def _fill( self, guests: list[dict], gen: int, unprotected: set[int] | None = None ) -> None: if gen != self.generation: # listed before a context switch: old cluster return self.unprotected = unprotected or set() self.guests = {str(g["vmid"]): g for g in guests} self._show_guests() def _show_guests(self) -> None: """Rebuild the table from self.guests: only the guests matching the filter.""" table = self.query_one("#guests", DataTable) current = ( # the key, not _selected(): self.guests may already be the new listing table.coordinate_to_cell_key(table.cursor_coordinate).row_key.value if table.row_count else None ) table.clear() query = self.query_one("#filter", Input).value for g in self.guests.values(): if not guest_matches(g, query): continue _add_plain(table, "!" if g["vmid"] in self.unprotected else "", g["vmid"], g["name"], g["type"], g["node"], g["status"], g.get("maxcpu", "-"), _scaled(g.get("maxmem"), 2**20), _scaled(g.get("maxdisk"), 2**30), *_percents(g), key=str(g["vmid"]), ) # fmt: skip if current is not None: try: table.move_cursor(row=table.get_row_index(current)) except RowDoesNotExist: # gone or filtered out: keep the default cursor pass self._update_frame() def action_filter(self) -> None: box = self.query_one("#filter", Input) box.display = True box.focus() def action_clear_filter(self) -> None: box = self.query_one("#filter", Input) box.value = "" # Input.Changed rebuilds the table box.display = False self.query_one("#guests", DataTable).focus() def on_input_changed(self, event: Input.Changed) -> None: if event.input.id == "filter": self._show_guests() def on_input_submitted(self, event: Input.Submitted) -> None: if event.input.id == "filter": self.query_one("#guests", DataTable).focus() def _reload(self, acted: dict | None = None) -> None: """Runs in a worker thread. `acted` gets its live status: the listing lags.""" gen = self.generation try: guests = cli.list_guests() for g in guests: if acted and g["vmid"] == acted["vmid"]: live = cli.current(g) # status and usage of the acted guest lag too g.update( { k: live[k] for k in ("status", "cpu", "mem", "maxmem") if k in live } ) except Exception as exc: # noqa: BLE001 - any failure must reach the user, not kill the app self.call_from_thread(self.notify, _message(exc), severity="error") return unprotected = None try: unprotected = cli.not_backed_up() except Exception as exc: # noqa: BLE001 - the listing still works without it self.call_from_thread( self.notify, f"backup coverage unknown: {_message(exc)}", severity="warning", ) self.call_from_thread(self._fill, guests, gen, unprotected) def action_reload(self) -> None: if self.switching: self.notify("Switching context; refresh after it ends", severity="warning") return self._reload_worker() @work(thread=True) def _reload_worker(self) -> None: self._reload() def action_context(self) -> None: if self.busy or self.switching: self.notify( "Busy (guest action or context switch); try again when it ends", severity="warning", ) return self._open_context() @work(thread=True) def _open_context(self) -> None: try: contexts = cli.contexts() except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.call_from_thread(self.notify, _message(exc), severity="error") return current = next((c["name"] for c in contexts if c["current"]), None) self.call_from_thread( self.push_screen, ContextScreen(contexts), lambda name: self._context_chosen(name, current), ) def _context_chosen(self, name: str | None, current: str | None) -> None: if name is None or name == current: return self.query_one( "#guests", DataTable ).clear() # old cluster's guests: not actionable self.guests = {} self.undo.clear() # VMIDs repeat across clusters: never undo onto another cluster self.package_count = None self._update_frame() self.generation += 1 self.switching = True self._switch_context(name) @work(thread=True) def _switch_context(self, name: str) -> None: try: try: cli.use_context(name) except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.call_from_thread(self.notify, _message(exc), severity="error") self._read_context() else: self.call_from_thread(setattr, self, "sub_title", f"context: {name}") self._reload() self._refresh_count() finally: self.call_from_thread(setattr, self, "switching", False) def start_job(self, guest: dict, label: str, job) -> None: if guest["vmid"] in self.busy: self.notify( f"{guest['name']} ({guest['vmid']}) is busy", severity="warning" ) return self.busy.add(guest["vmid"]) self.notify(f"{label} {guest['name']} ({guest['vmid']}) ...") self._run_job(guest, label, job) @work(thread=True) def _run_job(self, guest: dict, label: str, job) -> None: try: job() except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.call_from_thread( self.notify, f"{label} failed: {_message(exc)}", severity="error" ) else: self.call_from_thread( self.notify, f"{label} done: {guest['name']} ({guest['vmid']})" ) finally: self._reload_infos( guest ) # e.g. edited cores show up in an open info screen # before un-busying, so no context switch can start mid-reload self._reload(acted=guest) self.busy.discard(guest["vmid"]) def action_act(self, name: str) -> None: guest = self._selected() if guest is None: return run = lambda: cli.action(name, guest) if name == "start": self.start_job(guest, name, run) return who = f"{guest['name']} ({guest['vmid']})" text = f"{name.capitalize()} {who}?" if name == "stop": text = f"Hard power-off {who}? The guest is killed without shutting down." self.push_screen( ConfirmScreen(text, name.capitalize()), lambda ok: self.start_job(guest, name, run) if ok else None, ) def edit_guest(self, guest: dict) -> None: """Opened from the info screen (key e).""" if guest["vmid"] in self.busy: self.notify( f"{guest['name']} ({guest['vmid']}) is busy", severity="warning" ) return self._open_edit(guest) def action_info(self) -> None: guest = self._selected() if guest is not None: self.push_screen(InfoScreen(guest)) def on_data_table_row_selected(self, event: DataTable.RowSelected) -> None: if event.data_table.id == "guests": # Enter on a guest row self.action_info() def start_backup(self, guest: dict, storage: str) -> None: who = f"{guest['name']} ({guest['vmid']})" if guest["vmid"] in self.busy: self.notify(f"{who} is busy", severity="warning") return self.busy.add(guest["vmid"]) self.notify(f"Backup started: {who} to {storage}") self._run_backup(guest, storage) @work(thread=True) def _run_backup(self, guest: dict, storage: str) -> None: who = f"{guest['name']} ({guest['vmid']})" try: cli.backup_now(guest, storage) except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.call_from_thread( self.notify, f"Backup failed: {_message(exc)}", severity="error" ) else: self.call_from_thread(self.notify, f"Backup done: {who}") finally: self._reload_infos(guest) self.busy.discard(guest["vmid"]) def _reload_infos(self, guest: dict) -> None: """Runs in a worker thread: reload the guest's open InfoScreens (call while still busy).""" for screen in list(self.screen_stack): if isinstance(screen, InfoScreen) and screen.guest["vmid"] == guest["vmid"]: screen.load() def start_net_change( self, guest: dict, changes: dict, entry: dict | None, screen=None ) -> None: """`entry` becomes the guest's undo entry on success (None clears it). `screen` (the NetScreen) is closed on success and stays open on failure. """ who = f"{guest['name']} ({guest['vmid']})" if guest["vmid"] in self.busy: self.notify(f"{who} is busy", severity="warning") return self.busy.add(guest["vmid"]) self.notify(f"Network change started: {who}") self._run_net(guest, changes, entry, screen) @work(thread=True) def _run_net(self, guest: dict, changes: dict, entry: dict | None, screen) -> None: who = f"{guest['name']} ({guest['vmid']})" done = False try: cli.set_net(guest, changes) except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.call_from_thread( self.notify, f"Network change failed: {_message(exc)}", severity="error" ) else: done = True if entry is None: self.undo.pop(guest["vmid"], None) else: self.undo[guest["vmid"]] = entry self.call_from_thread(self.notify, f"Network change done: {who}") finally: self._reload_infos(guest) if done and screen is not None: self.call_from_thread(self._close_net, screen) self.busy.discard(guest["vmid"]) def _close_net(self, screen) -> None: if self.screen is screen: # the user may have left it already screen.dismiss() def action_shell(self) -> None: guest = self._selected() if guest is None: return who = f"{guest['name']} ({guest['vmid']})" if guest["type"] != "lxc": self.notify(f"{who} is not a container", severity="warning") return if guest["status"] != "running": self.notify(f"{who} is not running", severity="warning") return if guest["vmid"] in self.busy: self.notify(f"{who} is busy", severity="warning") return cmd = ["ssh", "-t", f"root@{guest['node']}", "pct", "enter", str(guest["vmid"])] error = None try: with self.suspend(): try: # inside: Textual's suspend() skips the resume if its body raises _reset_terminal() code = subprocess.run(cmd, check=False).returncode except OSError as exc: error = _message(exc) except SuspendNotSupported as exc: error = _message(exc) else: if ( error is None and code == 255 ): # other codes are the shell's own last status error = "ssh failed (exit 255)" if error: self.notify(f"Shell in {who} failed: {error}", severity="error") self.action_reload() @work(thread=True) def _open_edit(self, guest: dict) -> None: try: config = cli.guest_config(guest) disks = cli.disks(config, cli.kind(guest)) except Exception as exc: # noqa: BLE001 - surface every failure, never kill the app self.call_from_thread(self.notify, _message(exc), severity="error") return self.call_from_thread( self.push_screen, EditScreen(guest, config, disks), lambda v: self._edit_done(guest, v), ) def _edit_done(self, guest: dict, values: dict | None) -> None: if not values or not any(values[k] for k in ("cores", "memory", "grow")): return if values["grow"]: who = f"{guest['name']} ({guest['vmid']})" self.push_screen( ConfirmScreen( f"Grow {values['disk']} of {who} by {values['grow']} GiB? This cannot be undone.", "Grow disk", ), lambda ok: ( self._apply(guest, values) if ok else None ), # "No" cancels the whole edit ) else: self._apply(guest, values) def _apply(self, guest: dict, values: dict) -> None: cores, memory, grow = values["cores"], values["memory"], values["grow"] def job() -> None: if cores or memory: cli.set_resources(guest, cores=cores, memory_mb=memory) if grow: cli.resize_disk(guest, values["disk"], grow) if cores or memory: try: pending = cli.pending_keys(guest) except Exception as exc: # noqa: BLE001 - only a notice; the change itself succeeded self.call_from_thread( self.notify, f"pending check failed: {_message(exc)}", severity="warning", ) return if {"cores", "memory"} & set(pending): self.call_from_thread(self.notify, "Change applies at next reboot") self.start_job(guest, "edit", job) def _reset_terminal() -> None: """Full terminal reset (ESC c, what `reset` sends): clean screen, no leftover TUI modes.""" sys.stdout.write("\033c") sys.stdout.flush() def _resources(config: dict, kind: str) -> list[tuple[str, str]]: """Configured CPU, memory (and CT swap) and disks, as (name, value) rows.""" cores = config.get("cores") if cores is None and kind == "ct": cpu = "unlimited" # no cores limit: the container may use all host cores else: cores = int(cores or 1) cpu = f"{cores} core{'s' if cores != 1 else ''}" if kind == "vm": sockets = int(config.get("sockets") or 1) cpu += f" × {sockets} socket{'s' if sockets != 1 else ''}" rows = [("CPU", cpu)] if config.get("memory") is not None: rows.append(("Memory", f"{config['memory']} MiB")) if kind == "ct" and config.get("swap") is not None: rows.append(("Swap", f"{config['swap']} MiB")) rows += [(f"Disk {name}", size) for name, size in cli.disks(config, kind).items()] return rows def _pct(part, whole) -> float | None: return part / whole * 100 if part is not None and whole else None def _percents(g: dict) -> list[str]: """CPU % and Mem % cells of the listing; "-" unless running and the field is known.""" if g["status"] != "running": return ["-", "-"] cpu = g["cpu"] * 100 if g.get("cpu") is not None else None mem = _pct(g.get("mem"), g.get("maxmem")) return ["-" if v is None else f"{v:.1f}" for v in (cpu, mem)] def _in_use(usage: dict, kind: str) -> dict[str, str]: """In-use cells of a running guest by Resource name; missing fields stay absent ("-").""" out = {} if usage.get("cpu") is not None: out["CPU"] = f"{usage['cpu'] * 100:.1f} %" if usage.get("mem") is not None: pct = _pct(usage["mem"], usage.get("maxmem")) out["Memory"] = f"{usage['mem'] / 2**20:.0f} MiB" + ( f" ({pct:.0f} %)" if pct is not None else "" ) if kind == "ct": if usage.get("swap") is not None: out["Swap"] = f"{usage['swap'] / 2**20:.0f} MiB" if usage.get("disk") is not None: pct = _pct(usage["disk"], usage.get("maxdisk")) out["Disk rootfs"] = f"{usage['disk'] / 2**30:.1f} GiB" + ( f" ({pct:.0f} %)" if pct is not None else "" ) return out def _add_plain(table: DataTable, *cells, **kwargs) -> None: """DataTable parses str cells as markup; names and notes like "a[/b]" must stay literal.""" table.add_row(*(Text(c) if isinstance(c, str) else c for c in cells), **kwargs) def _scaled(value, unit: int): return "-" if value is None else round(value / unit) def main() -> None: GuestApp().run()