diff --git a/modalapi/archive.py b/modalapi/archive.py
new file mode 100644
index 000000000..7aea9867f
--- /dev/null
+++ b/modalapi/archive.py
@@ -0,0 +1,169 @@
+# This file is part of pi-stomp.
+#
+# pi-stomp is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# pi-Stomp is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with pi-stomp. If not, see .
+
+"""Backup/restore archiving on a worker thread, with progress the UI can poll.
+
+zip and unzip both name each entry as they finish it. Weighting those lines by
+the entry's uncompressed size gives a bar that tracks wall-clock rather than
+file count — 306 NAM models at ~295KB each dominate the run, while hundreds of
+small .ttl files would otherwise sprint the bar to nowhere.
+"""
+
+import enum
+import logging
+import os
+import re
+import signal
+import subprocess
+import threading
+import zipfile
+
+# " adding: path/to/file (deflated 12%)" — the path may contain spaces and
+# parens, so anchor on the trailing method/ratio rather than splitting.
+_ZIP_ENTRY = re.compile(r"^\s*(?:adding|updating):\s+(.*?)\s+\((?:stored|deflated)\s+\d+%\)\s*$")
+_UNZIP_ENTRY = re.compile(r"^\s*(?:extracting|inflating|linking|creating):\s+(.+?)\s*$")
+
+
+class JobState(enum.Enum):
+ RUNNING = "running"
+ DONE = "done"
+ FAILED = "failed"
+ CANCELLED = "cancelled"
+
+
+class ArchiveJob:
+ """Runs one archiving subprocess off of the UI thread. Poll progress()/state
+ from the main thread; nothing here touches the LCD."""
+
+ def __init__(self, argv: list[str], weights: dict[str, int], entry_re: re.Pattern[str]) -> None:
+ self._argv = argv
+ self._weights = weights
+ self._entry_re = entry_re
+ self._total = sum(weights.values()) or 1
+ self._done = 0
+ self._entry = ""
+ self._lock = threading.Lock()
+ self._state = JobState.RUNNING
+ self._error = ""
+ self._proc: subprocess.Popen[str] | None = None
+ self._cancelled = False
+ self._thread = threading.Thread(target=self._run, daemon=True, name="archive-job")
+ self._thread.start()
+
+ @staticmethod
+ def backup(script: str, dest: str, src_dir: str) -> "ArchiveJob":
+ weights = {}
+ for root, dirs, files in os.walk(src_dir):
+ if os.path.relpath(root, src_dir) == ".":
+ dirs[:] = [d for d in dirs if d != ".lv2"]
+ for name in files:
+ path = os.path.join(root, name)
+ rel = os.path.relpath(path, src_dir)
+ try:
+ weights[rel] = os.stat(path, follow_symlinks=True).st_size
+ except OSError:
+ weights[rel] = 0
+ return ArchiveJob([script, dest, src_dir], weights, _ZIP_ENTRY)
+
+ @staticmethod
+ def restore(script: str, username: str, archive: str, target_dir: str) -> "ArchiveJob":
+ with zipfile.ZipFile(archive) as zf:
+ weights = {i.filename: i.file_size for i in zf.infolist() if not i.is_dir()}
+ return ArchiveJob(["sudo", "-u", username, script, archive, target_dir], weights, _UNZIP_ENTRY)
+
+ def progress(self) -> float:
+ with self._lock:
+ return min(1.0, self._done / self._total)
+
+ @property
+ def current_entry(self) -> str:
+ with self._lock:
+ return self._entry
+
+ @property
+ def done_bytes(self) -> int:
+ with self._lock:
+ return min(self._done, self._total)
+
+ @property
+ def total_bytes(self) -> int:
+ return self._total
+
+ @property
+ def state(self) -> JobState:
+ with self._lock:
+ return self._state
+
+ @property
+ def error(self) -> str:
+ with self._lock:
+ return self._error
+
+ def cancel(self) -> None:
+ with self._lock:
+ self._cancelled = True
+ proc = self._proc
+ if proc is None or proc.poll() is not None:
+ return
+ # The script is a bash parent of zip; signal the group or only bash dies.
+ try:
+ os.killpg(proc.pid, signal.SIGTERM)
+ except (ProcessLookupError, PermissionError) as e:
+ logging.warning("archive cancel: %s", e)
+
+ def _run(self) -> None:
+ try:
+ proc = subprocess.Popen(
+ self._argv,
+ stdout=subprocess.PIPE,
+ stderr=subprocess.STDOUT,
+ text=True,
+ start_new_session=True,
+ )
+ except OSError as e:
+ with self._lock:
+ self._state = JobState.FAILED
+ self._error = str(e)
+ return
+
+ with self._lock:
+ self._proc = proc
+
+ tail: list[str] = []
+ assert proc.stdout is not None
+ for line in proc.stdout:
+ m = self._entry_re.match(line)
+ if m is None:
+ tail.append(line.rstrip())
+ del tail[:-8]
+ continue
+ name = m.group(1)
+ weight = self._weights.get(name)
+ if weight is None:
+ continue
+ with self._lock:
+ self._done += weight
+ self._entry = name
+
+ rc = proc.wait()
+ with self._lock:
+ if self._cancelled:
+ self._state = JobState.CANCELLED
+ elif rc == 0:
+ self._state = JobState.DONE
+ self._done = self._total
+ else:
+ self._state = JobState.FAILED
+ self._error = "\n".join(tail) or f"exited {rc}"
diff --git a/modalapi/modhandler.py b/modalapi/modhandler.py
index 0993a984a..dfaf824bb 100755
--- a/modalapi/modhandler.py
+++ b/modalapi/modhandler.py
@@ -18,6 +18,7 @@
from modalapi.sync import SyncMode, SyncModeSetter
import bisect
+import datetime
import json
import logging
import os
@@ -1603,6 +1604,29 @@ def _human_size(num_bytes: int) -> str:
value /= 1000
return f"{value:.1f}GB"
+ def _drive_detail(self, backup_dir: str) -> str:
+ """Drive name plus free space — the thing that decides whether a backup fits."""
+ mount = os.path.dirname(backup_dir)
+ name = os.path.basename(mount)
+ try:
+ usage = shutil.disk_usage(mount)
+ except OSError:
+ return name
+ return f"{name} · {self._human_size(usage.free)} free of {self._human_size(usage.total)}"
+
+ def _archive_detail(self, backup_dir: str) -> str:
+ """Drive name plus the archive's size and age — restore overwrites data/,
+ so which vintage is about to land matters more than free space."""
+ mount = os.path.dirname(backup_dir)
+ name = os.path.basename(mount)
+ path = os.path.join(backup_dir, self.backup_file)
+ try:
+ st = os.stat(path)
+ except OSError:
+ return name
+ when = datetime.datetime.fromtimestamp(st.st_mtime).strftime("%d %b %H:%M")
+ return f"{name} · {self._human_size(st.st_size)} from {when}"
+
def _drive_label(self, backup_dir: str) -> str:
mount = os.path.dirname(backup_dir)
try:
@@ -1627,17 +1651,22 @@ def user_backup_data(self, arg):
self._choose_usb_drive(self.check_usb(), self._do_backup_data)
def _do_backup_data(self, backup_dir: str):
- self.lcd.draw_info_message("Backing up, please wait...", refresh=True)
+ from modalapi.archive import ArchiveJob
+ from ui.archive_panel import ArchiveProgressPanel
+
logging.info("Data backup...")
cmd = os.path.join(self.homedir, "util", "data-backup.sh")
- try:
- subprocess.check_output([cmd, os.path.join(backup_dir, self.backup_file), self.data_dir])
- self.lcd.draw_message_dialog("Backup complete", "Info")
- logging.info("Backup complete")
- except subprocess.CalledProcessError as e:
- logging.error("user_backup_data:" + str(e.output))
- finally:
- self.lcd.draw_info_message("", refresh=True)
+ job = ArchiveJob.backup(cmd, os.path.join(backup_dir, self.backup_file), self.data_dir)
+ self.lcd.pstack.push_panel(
+ ArchiveProgressPanel(
+ title="Backing up",
+ noun="Backup",
+ subtitle=self._drive_detail(backup_dir),
+ job=job,
+ on_dismiss=self._dismiss_archive_panel,
+ cancellable=True,
+ )
+ )
def user_restore_data(self, arg, on_success=None):
# Only offer drives that actually hold a backup — no point asking the
@@ -1646,24 +1675,44 @@ def user_restore_data(self, arg, on_success=None):
self._choose_usb_drive(restorable, lambda d: self._do_restore_data(d, on_success=on_success))
def _do_restore_data(self, backup_dir: str, on_success=None):
- self.lcd.draw_info_message("Restoring, please wait...", refresh=True)
+ from modalapi.archive import ArchiveJob
+ from ui.archive_panel import ArchiveProgressPanel
+
logging.info("Restoring data backup...")
cmd = os.path.join(self.homedir, "util", "data-restore.sh")
- try:
- subprocess.check_output(
- ["sudo", "-u", self.username, cmd, os.path.join(backup_dir, self.backup_file), self.data_dir]
- )
- logging.info("Restore complete")
- if on_success is not None:
- on_success()
- self.lcd.draw_message_dialog(
- "Restore complete. Press OK to restart.", "Info", on_dismiss=lambda: self.system_menu_restart_sound(None)
+ job = ArchiveJob.restore(cmd, self.username, os.path.join(backup_dir, self.backup_file), self.data_dir)
+ self.lcd.pstack.push_panel(
+ ArchiveProgressPanel(
+ title="Restoring",
+ noun="Restore",
+ subtitle=self._archive_detail(backup_dir),
+ job=job,
+ on_dismiss=lambda: self._dismiss_restore_panel(on_success),
+ cancellable=False,
+ done_label="Restart to continue",
)
- except subprocess.CalledProcessError as e:
- self.lcd.draw_message_dialog(e.output.decode("utf-8"))
- logging.error("user_restore_data: " + e.output.decode("utf-8"))
- finally:
- self.lcd.draw_info_message("", refresh=True)
+ )
+
+ def _dismiss_archive_panel(self) -> None:
+ from ui.archive_panel import ArchiveProgressPanel
+
+ panel = self.lcd.pstack.find_panel_type(ArchiveProgressPanel)
+ if panel is not None:
+ self.lcd.pstack.pop_panel(panel)
+ self.lcd.draw_main_panel()
+
+ def _dismiss_restore_panel(self, on_success=None) -> None:
+ from modalapi.archive import JobState
+ from ui.archive_panel import ArchiveProgressPanel
+
+ panel = self.lcd.pstack.find_panel_type(ArchiveProgressPanel)
+ restored = panel is not None and panel.job_state is JobState.DONE
+ self._dismiss_archive_panel()
+ if not restored:
+ return
+ if on_success is not None:
+ on_success()
+ self.restart_ui_stack()
def system_menu_save_current_pb(self, _arg: None):
if self._current is None:
@@ -1697,6 +1746,17 @@ def system_menu_reload(self, arg):
logging.info("Exiting main process, systemctl should restart if enabled")
sys.exit(0)
+ def restart_ui_stack(self) -> None:
+ logging.info("Restarting mod-ui + deps")
+ try:
+ subprocess.Popen(
+ ["sudo", "systemctl", "--no-block", "restart", "mod-ui"],
+ stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL,
+ )
+ except OSError as e:
+ logging.error("restart_ui_stack: %s", e)
+
def system_menu_restart_sound(self, arg):
self.lcd.splash_show()
logging.info("Restart sound engine (jack)")
diff --git a/pistomp/nam/panel.py b/pistomp/nam/panel.py
index 8211af172..8cadb0ebe 100644
--- a/pistomp/nam/panel.py
+++ b/pistomp/nam/panel.py
@@ -45,6 +45,7 @@
from uilib.label import Label
from uilib.misc import TextHAlign, get_text_bbox, get_text_size
from uilib.paint import PaintContext
+from uilib.progress_bar import ProgressBarWidget, fmt_time
from uilib.pygame_init import font as _make_font
from uilib.text import Button, TextWidget
from uilib.widget import Widget
@@ -114,14 +115,6 @@
# ── Colour palette ────────────────────────────────────────────────────────────
# Progress bar colour stops (position 0.0–1.0 along bar width)
-_BAR_STOPS: list[tuple[float, tuple[int, int, int]]] = [
- (0.00, (0, 200, 75)),
- (0.35, (120, 215, 0)),
- (0.65, (230, 148, 0)),
- (1.00, (215, 55, 10)),
-]
-_BAR_DIM = 0.13 # brightness of unfilled segments
-
# Status LED
_LED_IDLE = (70, 70, 78)
_LED_CAPTURING = (0, 200, 80)
@@ -158,11 +151,6 @@
# ── Helpers ───────────────────────────────────────────────────────────────────
-def _fmt_time(seconds: float) -> str:
- s = max(0, int(seconds))
- return f"{s // 60}:{s % 60:02d}"
-
-
def _centred_x(text: str, font, width: int) -> int:
bb = get_text_bbox(text, font)
return (width - (bb[2] - bb[0])) // 2 - bb[0]
@@ -200,104 +188,6 @@ def symbol_for(self, role: ParamRole) -> str | None:
return self.symbol
-class ProgressBarWidget(Widget):
- """Segmented colour-gradient progress bar with elapsed/remaining time labels."""
-
- _MARGIN = 12 # left/right inset
- _BAR_Y = 30 # top of bar within widget
- _BAR_H = 30 # bar height
- _LABEL_GAP = 10 # gap between bar bottom and label top
- _N_SEGS = 40 # number of colour segments
- _SEG_GAP = 2 # gap between segments in pixels
-
- def __init__(self, box: Box, total_seconds: float, font, caption_font, parent: Widget) -> None:
- super().__init__(box=box, bkgnd_color=(0, 0, 0), parent=parent)
- self._total = total_seconds
- self._progress = 0.0
- self._frozen = False
- self._elapsed = 0.0
- self._remaining = total_seconds
- self._font = font
- self._caption_font = caption_font
- inner_w = box.width - 2 * self._MARGIN
- self._seg_w = max(1, (inner_w - (self._N_SEGS - 1) * self._SEG_GAP) // self._N_SEGS)
-
- def set_progress(self, progress: float) -> None:
- if self._frozen:
- return
- p = max(0.0, min(1.0, progress))
- old_filled = int(self._progress * self._N_SEGS)
- self._progress = p
- self._elapsed = p * self._total
- self._remaining = self._total - self._elapsed
- if int(p * self._N_SEGS) != old_filled:
- self.refresh()
-
- def freeze(self) -> None:
- self._frozen = True
-
- def set_done(self) -> None:
- self._progress = 1.0
- self._elapsed = self._total
- self._remaining = 0.0
- self._frozen = True
-
- def reset(self) -> None:
- self._progress = 0.0
- self._elapsed = 0.0
- self._remaining = self._total
- self._frozen = False
-
- def advance_rotation(self, dt: float) -> None:
- pass
-
- @staticmethod
- def _color_at(t: float) -> tuple[int, int, int]:
- stops = _BAR_STOPS
- if t <= stops[0][0]:
- return stops[0][1]
- if t >= stops[-1][0]:
- return stops[-1][1]
- for i in range(len(stops) - 1):
- t0, c0 = stops[i]
- t1, c1 = stops[i + 1]
- if t0 <= t <= t1:
- f = (t - t0) / (t1 - t0)
- return (
- int(c0[0] + f * (c1[0] - c0[0])),
- int(c0[1] + f * (c1[1] - c0[1])),
- int(c0[2] + f * (c1[2] - c0[2])),
- )
- return stops[-1][1]
-
- def _draw(self, ctx: PaintContext) -> None:
- n = self._N_SEGS
- filled = int(self._progress * n)
- sw = self._seg_w
- bx = self._MARGIN
- by = self._BAR_Y
- ctx.fill((0, 0, 0))
-
- for i in range(n):
- t = i / (n - 1) if n > 1 else 0.0
- r, g, b = self._color_at(t)
- if i < filled:
- color: tuple[int, int, int] = (r, g, b)
- else:
- color = (int(r * _BAR_DIM), int(g * _BAR_DIM), int(b * _BAR_DIM))
- ctx.draw_rectangle(Box.xywh(bx + i * (sw + self._SEG_GAP), by, sw, self._BAR_H), fill=color)
-
- label_y = by + self._BAR_H + self._LABEL_GAP
- right_x = ctx.width - self._MARGIN
-
- elapsed_str = _fmt_time(self._elapsed)
- ctx.draw_text((bx, label_y), elapsed_str, fill=(130, 118, 80), font=self._font)
-
- remaining_str = f"−{_fmt_time(self._remaining)}"
- rw, _ = get_text_size(remaining_str, self._font)
- ctx.draw_text((right_x - rw, label_y), remaining_str, fill=(205, 180, 110), font=self._font)
-
-
class LevelMeter(Widget):
"""Segmented horizontal VU meter with dB readout and clip indicator."""
@@ -505,7 +395,7 @@ def __init__(
)
self._btn_start = Button(
box=Box.xywh(_BTN_X_ACTION, _BTN_Y, _BTN_W, _BTN_H),
- text=f"Start ({_fmt_time(self._duration)})",
+ text=f"Start ({fmt_time(self._duration)})",
font=font,
outline_radius=4,
parent=self,
diff --git a/tests/archive_fake.py b/tests/archive_fake.py
new file mode 100644
index 000000000..11b1d84d6
--- /dev/null
+++ b/tests/archive_fake.py
@@ -0,0 +1,51 @@
+"""Test double for ArchiveJob — no thread, no subprocess, state driven by hand."""
+
+from contextlib import contextmanager
+from unittest.mock import patch
+
+from modalapi.archive import JobState
+
+
+class FakeArchiveJob:
+ def __init__(self, argv: list[str], total_bytes: int = 458_633_729, state: JobState = JobState.RUNNING) -> None:
+ self.argv = argv
+ self.total_bytes = total_bytes
+ self.done_bytes = 0
+ self.current_entry = ""
+ self.state = state
+ self.error = ""
+ self.cancelled = False
+
+ def progress(self) -> float:
+ return self.done_bytes / self.total_bytes
+
+ def cancel(self) -> None:
+ self.cancelled = True
+ self.state = JobState.CANCELLED
+
+ def advance(self, fraction: float, entry: str = "") -> None:
+ self.done_bytes = int(self.total_bytes * fraction)
+ if entry:
+ self.current_entry = entry
+
+ def finish(self, state: JobState = JobState.DONE, error: str = "") -> None:
+ self.done_bytes = self.total_bytes
+ self.state = state
+ self.error = error
+
+
+@contextmanager
+def fake_jobs(state: JobState = JobState.RUNNING):
+ """Patch both ArchiveJob factories. Yields the list of jobs handed out."""
+ created: list[FakeArchiveJob] = []
+
+ def make(*argv):
+ job = FakeArchiveJob(list(argv), state=state)
+ created.append(job)
+ return job
+
+ with (
+ patch("modalapi.archive.ArchiveJob.backup", side_effect=make),
+ patch("modalapi.archive.ArchiveJob.restore", side_effect=make),
+ ):
+ yield created
diff --git a/tests/integration/test_system_menu.py b/tests/integration/test_system_menu.py
index 3704aa760..0580d6efc 100644
--- a/tests/integration/test_system_menu.py
+++ b/tests/integration/test_system_menu.py
@@ -3,7 +3,10 @@
import os
from unittest.mock import patch
+from modalapi.archive import JobState
+from tests.archive_fake import fake_jobs
from tests.types import SystemFixture
+from ui.archive_panel import ArchiveProgressPanel
def test_system_menu_shutdown(modhandler_system: SystemFixture):
@@ -95,19 +98,18 @@ def test_check_usb_ignores_unmounted_media_dirs(modhandler_system: SystemFixture
assert handler.check_usb() == []
-def test_backup_with_usb_runs_script(modhandler_system: SystemFixture):
- """user_backup_data() invokes data-backup.sh with the discovered mount's backups dir."""
+def test_backup_with_usb_starts_job_behind_progress_panel(modhandler_system: SystemFixture):
+ """user_backup_data() starts the job against the discovered mount and shows progress
+ rather than blocking the UI thread until zip finishes."""
handler = modhandler_system.handler
with (
patch.object(handler, "check_usb", return_value=["/media/MYSTICK/backups"]),
- patch("subprocess.check_output") as mock_backup,
- patch.object(handler.lcd, "draw_message_dialog") as mock_dialog,
+ fake_jobs() as jobs,
):
handler.user_backup_data(None)
- args = mock_backup.call_args[0][0]
- assert args[1] == os.path.join("/media/MYSTICK/backups", handler.backup_file)
- mock_dialog.assert_called_once_with("Backup complete", "Info")
+ assert jobs[0].argv[1] == os.path.join("/media/MYSTICK/backups", handler.backup_file)
+ assert handler.lcd.pstack.find_panel_type(ArchiveProgressPanel) is not None
class _FakeUsage:
@@ -122,12 +124,12 @@ def test_backup_with_multiple_usb_shows_selection_menu(modhandler_system: System
with (
patch.object(handler, "check_usb", return_value=dirs),
patch("shutil.disk_usage", return_value=_FakeUsage(32_000_000_000)),
- patch("subprocess.check_output") as mock_backup,
+ fake_jobs() as jobs,
patch.object(handler.lcd, "draw_selection_menu") as mock_menu,
):
handler.user_backup_data(None)
- mock_backup.assert_not_called()
+ assert jobs == []
mock_menu.assert_called_once()
args, kwargs = mock_menu.call_args
assert args[1] == "Choose USB drive"
@@ -136,9 +138,9 @@ def test_backup_with_multiple_usb_shows_selection_menu(modhandler_system: System
# Picking the second item runs the backup against that stick's dir.
_label, callback, arg = items[1]
- with patch("subprocess.check_output") as mock_backup, patch.object(handler.lcd, "draw_message_dialog"):
+ with fake_jobs() as jobs:
callback(arg)
- assert mock_backup.call_args[0][0][1] == os.path.join(dirs[1], handler.backup_file)
+ assert jobs[0].argv[1] == os.path.join(dirs[1], handler.backup_file)
def test_restore_only_offers_drives_with_a_backup(modhandler_system: SystemFixture):
@@ -148,36 +150,81 @@ def test_restore_only_offers_drives_with_a_backup(modhandler_system: SystemFixtu
with (
patch.object(handler, "check_usb", return_value=dirs),
patch("os.path.exists", side_effect=lambda p: p == os.path.join(dirs[1], handler.backup_file)),
- patch("subprocess.check_output") as mock_restore,
+ fake_jobs() as jobs,
patch.object(handler.lcd, "draw_selection_menu") as mock_menu,
- patch.object(handler, "system_menu_restart_sound"),
- patch.object(handler.lcd, "draw_message_dialog"),
+ patch.object(handler, "restart_ui_stack"),
):
handler.user_restore_data(None)
mock_menu.assert_not_called()
- assert mock_restore.call_args[0][0][-2] == os.path.join(dirs[1], handler.backup_file)
+ assert jobs[0].argv[-2] == os.path.join(dirs[1], handler.backup_file)
-def test_restore_success_defers_restart_until_ok_pressed(modhandler_system: SystemFixture):
- """A successful restore must not restart anything until the user presses OK — the restart
- cascades (jack -> mod-host -> mod-ui -> pi-stomp), so firing it immediately would tear down
- the process before the user ever sees the confirmation dialog."""
+def test_restore_defers_restart_until_the_button_is_pressed(modhandler_system: SystemFixture):
+ """A successful restore must not restart anything on its own — the restart cascades
+ (jack -> mod-host -> mod-ui -> pi-stomp), so firing it the moment unzip exits would tear
+ the process down under the user. The button press is the consent; there is no dialog."""
handler = modhandler_system.handler
with (
- patch("subprocess.check_output", return_value=b""),
+ fake_jobs() as jobs,
patch.object(handler.lcd, "draw_message_dialog") as mock_dialog,
- patch.object(handler, "system_menu_restart_sound") as mock_restart,
+ patch.object(handler, "restart_ui_stack") as mock_restart,
):
handler._do_restore_data("/media/MYSTICK/backups")
+ panel = handler.lcd.pstack.find_panel_type(ArchiveProgressPanel)
+ assert panel is not None
+ jobs[0].finish(JobState.DONE)
+ panel.tick()
mock_restart.assert_not_called()
- assert "OK" in mock_dialog.call_args[0][0]
+ assert panel._btn.text == "Restart to continue"
+
+ panel._on_button()
+
+ mock_restart.assert_called_once_with()
+ mock_dialog.assert_not_called()
+
+
+def test_failed_restore_offers_close_and_never_restarts(modhandler_system: SystemFixture):
+ """A restore that failed leaves data/ half-written; restarting into that is worse than
+ staying put, so the terminal button must not invite it."""
+ handler = modhandler_system.handler
+ with (
+ fake_jobs() as jobs,
+ patch.object(handler, "restart_ui_stack") as mock_restart,
+ ):
+ handler._do_restore_data("/media/MYSTICK/backups")
+ panel = handler.lcd.pstack.find_panel_type(ArchiveProgressPanel)
+ assert panel is not None
+
+ jobs[0].finish(JobState.FAILED, error="unzip: cannot find zipfile directory")
+ panel.tick()
+ assert panel._btn.text == "Close"
+
+ panel._on_button()
+
+ mock_restart.assert_not_called()
+
+
+def test_backup_completion_shows_no_dialog(modhandler_system: SystemFixture):
+ """Backup reports success in the panel itself — no popup to dismiss afterwards."""
+ handler = modhandler_system.handler
+ with (
+ patch.object(handler, "check_usb", return_value=["/media/MYSTICK/backups"]),
+ fake_jobs() as jobs,
+ patch.object(handler.lcd, "draw_message_dialog") as mock_dialog,
+ ):
+ handler.user_backup_data(None)
+ panel = handler.lcd.pstack.find_panel_type(ArchiveProgressPanel)
+ assert panel is not None
- on_dismiss = mock_dialog.call_args.kwargs["on_dismiss"]
- on_dismiss()
+ jobs[0].finish(JobState.DONE)
+ panel.tick()
+ assert panel._btn.text == "Close"
+ panel._on_button()
- mock_restart.assert_called_once_with(None)
+ mock_dialog.assert_not_called()
+ assert handler.lcd.pstack.find_panel_type(ArchiveProgressPanel) is None
def test_restore_with_no_backups_shows_no_usb_dialog(modhandler_system: SystemFixture):
@@ -193,3 +240,54 @@ def test_restore_with_no_backups_shows_no_usb_dialog(modhandler_system: SystemFi
mock_dialog.assert_called_once()
assert "USB" in mock_dialog.call_args[0][0]
+
+
+def test_drive_detail_reports_free_space(modhandler_system: SystemFixture):
+ """The backup subtitle names the drive and how much room is left on it."""
+ handler = modhandler_system.handler
+
+ class _Usage:
+ total = 58_000_000_000
+ free = 57_000_000_000
+
+ with patch("shutil.disk_usage", return_value=_Usage()):
+ assert handler._drive_detail("/media/STAGE_LEFT/backups") == "STAGE_LEFT · 57.0GB free of 58.0GB"
+
+
+def test_drive_detail_falls_back_to_name_when_unreadable(modhandler_system: SystemFixture):
+ """A stick yanked between menu and panel must not take the panel down with it."""
+ handler = modhandler_system.handler
+ with patch("shutil.disk_usage", side_effect=OSError):
+ assert handler._drive_detail("/media/GONE/backups") == "GONE"
+
+
+def test_archive_detail_reports_size_and_age(modhandler_system: SystemFixture):
+ """Restore overwrites data/, so the subtitle says which vintage is about to land."""
+ handler = modhandler_system.handler
+
+ class _Stat:
+ st_size = 344_307_247
+ st_mtime = 1_754_942_700.0 # 2025-08-11 18:45 local
+
+ with patch("os.stat", return_value=_Stat()):
+ detail = handler._archive_detail("/media/STAGE_LEFT/backups")
+
+ assert detail.startswith("STAGE_LEFT · 344.3MB from ")
+
+
+def test_archive_detail_falls_back_to_name_when_missing(modhandler_system: SystemFixture):
+ handler = modhandler_system.handler
+ with patch("os.stat", side_effect=OSError):
+ assert handler._archive_detail("/media/GONE/backups") == "GONE"
+
+
+def test_restart_ui_stack_is_non_blocking_and_skips_jack(modhandler_system: SystemFixture):
+ """Restoring data/ invalidates what mod-ui and pi-stomp cache, not the audio engine.
+ Restarting jack would drag six other units down with it, and os.system would block the
+ 10ms loop while waiting on the restart that kills this very process."""
+ handler = modhandler_system.handler
+ with patch("subprocess.Popen") as mock_popen:
+ handler.restart_ui_stack()
+
+ argv = mock_popen.call_args[0][0]
+ assert argv == ["sudo", "systemctl", "--no-block", "restart", "mod-ui"]
diff --git a/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_closed.png b/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_closed.png
index 742d88283..0b7492e4c 100644
Binary files a/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_closed.png and b/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_closed.png differ
diff --git a/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_edited.png b/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_edited.png
index 08af9c26f..ea93d3db7 100644
Binary files a/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_edited.png and b/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_edited.png differ
diff --git a/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_open.png b/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_open.png
index aa934ec7d..e3a98e23a 100644
Binary files a/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_open.png and b/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_dialog_open.png differ
diff --git a/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_focused.png b/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_focused.png
index b7783a4a9..b1bd1d67e 100644
Binary files a/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_focused.png and b/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_edits_gain/gain_focused.png differ
diff --git a/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_initial_render/idle.png b/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_initial_render/idle.png
index d8dc28b5b..7b5e06fc1 100644
Binary files a/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_initial_render/idle.png and b/tests/snapshots/v2/test_nam_panel/test_nam_nav_only_initial_render/idle.png differ
diff --git a/tests/snapshots/v3/test_nam_panel/test_aborted_state/aborted.png b/tests/snapshots/v3/test_nam_panel/test_aborted_state/aborted.png
index ab2fb939d..616dd2082 100644
Binary files a/tests/snapshots/v3/test_nam_panel/test_aborted_state/aborted.png and b/tests/snapshots/v3/test_nam_panel/test_aborted_state/aborted.png differ
diff --git a/tests/snapshots/v3/test_nam_panel/test_capturing_state/capturing.png b/tests/snapshots/v3/test_nam_panel/test_capturing_state/capturing.png
index 2aa237933..2f2134321 100644
Binary files a/tests/snapshots/v3/test_nam_panel/test_capturing_state/capturing.png and b/tests/snapshots/v3/test_nam_panel/test_capturing_state/capturing.png differ
diff --git a/tests/snapshots/v3/test_nam_panel/test_done_state/done.png b/tests/snapshots/v3/test_nam_panel/test_done_state/done.png
index 09f3e7e6a..9945b63a5 100644
Binary files a/tests/snapshots/v3/test_nam_panel/test_done_state/done.png and b/tests/snapshots/v3/test_nam_panel/test_done_state/done.png differ
diff --git a/tests/snapshots/v3/test_nam_panel/test_failed_state/failed.png b/tests/snapshots/v3/test_nam_panel/test_failed_state/failed.png
index ff22ddb64..385172fbe 100644
Binary files a/tests/snapshots/v3/test_nam_panel/test_failed_state/failed.png and b/tests/snapshots/v3/test_nam_panel/test_failed_state/failed.png differ
diff --git a/tests/snapshots/v3/test_nam_panel/test_idle_state/idle.png b/tests/snapshots/v3/test_nam_panel/test_idle_state/idle.png
index d8dc28b5b..7b5e06fc1 100644
Binary files a/tests/snapshots/v3/test_nam_panel/test_idle_state/idle.png and b/tests/snapshots/v3/test_nam_panel/test_idle_state/idle.png differ
diff --git a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_cancel_leaves_previous_archive/cancelled.png b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_cancel_leaves_previous_archive/cancelled.png
new file mode 100644
index 000000000..3afb5df08
Binary files /dev/null and b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_cancel_leaves_previous_archive/cancelled.png differ
diff --git a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_completes_after_drive_chosen/complete.png b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_completes_after_drive_chosen/complete.png
index bd81d5c5e..79edd1b72 100644
Binary files a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_completes_after_drive_chosen/complete.png and b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_completes_after_drive_chosen/complete.png differ
diff --git a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_completes_after_drive_chosen/in_progress.png b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_completes_after_drive_chosen/in_progress.png
new file mode 100644
index 000000000..0792f3c67
Binary files /dev/null and b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_completes_after_drive_chosen/in_progress.png differ
diff --git a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_completes_after_drive_chosen/partway.png b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_completes_after_drive_chosen/partway.png
new file mode 100644
index 000000000..4c3b8761d
Binary files /dev/null and b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_completes_after_drive_chosen/partway.png differ
diff --git a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_failure_shows_error/failed.png b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_failure_shows_error/failed.png
new file mode 100644
index 000000000..2f000ab62
Binary files /dev/null and b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_backup_failure_shows_error/failed.png differ
diff --git a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_completes_after_drive_chosen/complete.png b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_completes_after_drive_chosen/complete.png
index 3dc4756d0..11cff8b59 100644
Binary files a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_completes_after_drive_chosen/complete.png and b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_completes_after_drive_chosen/complete.png differ
diff --git a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_completes_after_drive_chosen/in_progress.png b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_completes_after_drive_chosen/in_progress.png
new file mode 100644
index 000000000..3961fb1fc
Binary files /dev/null and b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_completes_after_drive_chosen/in_progress.png differ
diff --git a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_skips_menu_when_only_one_drive_has_a_backup/0.png b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_skips_menu_when_only_one_drive_has_a_backup/0.png
index 3dc4756d0..3961fb1fc 100644
Binary files a/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_skips_menu_when_only_one_drive_has_a_backup/0.png and b/tests/snapshots/v3/test_usb_backup_restore_snapshot/test_v3_restore_skips_menu_when_only_one_drive_has_a_backup/0.png differ
diff --git a/tests/snapshots/v3/test_welcome_panel/test_restore_success/restore_success.png b/tests/snapshots/v3/test_welcome_panel/test_restore_success/restore_success.png
index 3dc4756d0..0f8597b07 100644
Binary files a/tests/snapshots/v3/test_welcome_panel/test_restore_success/restore_success.png and b/tests/snapshots/v3/test_welcome_panel/test_restore_success/restore_success.png differ
diff --git a/tests/v3/test_usb_backup_restore_snapshot.py b/tests/v3/test_usb_backup_restore_snapshot.py
index ec613cefa..39d753ef0 100644
--- a/tests/v3/test_usb_backup_restore_snapshot.py
+++ b/tests/v3/test_usb_backup_restore_snapshot.py
@@ -3,16 +3,24 @@
from unittest.mock import patch
+import time
+
+from pistomp.input.event import SwitchEvent, SwitchEventKind
+from tests.v3.nav_helpers import nav_click, nav_encoder
from uilib.misc import InputEvent
+from modalapi.archive import JobState
+from tests.archive_fake import fake_jobs
from tests.types import SystemFixture
+from ui.archive_panel import ArchiveProgressPanel
_TWO_DRIVES = ["/media/STAGE_LEFT/backups", "/media/STAGE_RIGHT/backups"]
class _FakeUsage:
- def __init__(self, total: int):
+ def __init__(self, total: int, free: int = 31_000_000_000):
self.total = total
+ self.free = free
def _setup_main_panel(v3_system: SystemFixture):
@@ -35,6 +43,11 @@ def _navigate_to_drive(handler, backup_dir: str):
raise AssertionError(f"no menu item for {backup_dir}")
+def _tick(handler):
+ """Drive one UI poll so the progress panel picks up the job's state."""
+ handler.lcd.poll_updates()
+
+
def _click_selected(handler):
"""Simulate a real click on the open selection menu's currently-highlighted item,
driving the same dismiss-then-callback path a physical encoder click would."""
@@ -69,10 +82,58 @@ def test_v3_backup_completes_after_drive_chosen(v3_system: SystemFixture, snapsh
_navigate_to_drive(handler, _TWO_DRIVES[1])
snapshot("selection")
- with patch("subprocess.check_output", return_value=b""):
+ with fake_jobs() as jobs, patch("shutil.disk_usage", return_value=_FakeUsage(58_000_000_000, 57_000_000_000)):
_click_selected(handler)
+ _tick(handler)
+ snapshot("in_progress")
+
+ jobs[0].advance(0.62, "user-files/NAM Models/Boost Pedal Pack/FORTIN GRIND.nam")
+ _tick(handler)
+ snapshot("partway")
- snapshot("complete")
+ jobs[0].finish(JobState.DONE)
+ _tick(handler)
+ snapshot("complete")
+
+
+def test_v3_backup_cancel_leaves_previous_archive(v3_system: SystemFixture, snapshot):
+ """Cancel is offered while a backup runs; the panel then reports the old archive is intact."""
+ handler = v3_system.handler
+ _setup_main_panel(v3_system)
+
+ with (
+ patch.object(handler, "check_usb", return_value=[_TWO_DRIVES[0]]),
+ patch("shutil.disk_usage", return_value=_FakeUsage(58_000_000_000, 57_000_000_000)),
+ fake_jobs() as jobs,
+ ):
+ handler.user_backup_data(None)
+ panel = handler.lcd.pstack.find_panel_type(ArchiveProgressPanel)
+ assert panel is not None
+
+ jobs[0].advance(0.4, "user-files/NAM Models/DDE - Life Droner (NoCab).nam")
+ _tick(handler)
+ panel._on_button()
+ _tick(handler)
+
+ assert jobs[0].cancelled
+ snapshot("cancelled")
+
+
+def test_v3_backup_failure_shows_error(v3_system: SystemFixture, snapshot):
+ """A failed backup surfaces the script's last output line instead of dying silently."""
+ handler = v3_system.handler
+ _setup_main_panel(v3_system)
+
+ with (
+ patch.object(handler, "check_usb", return_value=[_TWO_DRIVES[0]]),
+ patch("shutil.disk_usage", return_value=_FakeUsage(58_000_000_000, 57_000_000_000)),
+ fake_jobs() as jobs,
+ ):
+ handler.user_backup_data(None)
+ jobs[0].finish(JobState.FAILED, error="zip I/O error: No space left on device")
+ _tick(handler)
+
+ snapshot("failed")
def test_v3_restore_shows_usb_drive_selection_menu(v3_system: SystemFixture, snapshot):
@@ -104,10 +165,14 @@ def test_v3_restore_completes_after_drive_chosen(v3_system: SystemFixture, snaps
_navigate_to_drive(handler, _TWO_DRIVES[1])
snapshot("selection")
- with patch("subprocess.check_output", return_value=b""), patch("os.system"):
+ with fake_jobs() as jobs, patch("os.system"):
_click_selected(handler)
+ _tick(handler)
+ snapshot("in_progress")
- snapshot("complete")
+ jobs[0].finish(JobState.DONE)
+ _tick(handler)
+ snapshot("complete")
def test_v3_restore_skips_menu_when_only_one_drive_has_a_backup(v3_system: SystemFixture, snapshot):
@@ -118,9 +183,83 @@ def test_v3_restore_skips_menu_when_only_one_drive_has_a_backup(v3_system: Syste
with (
patch.object(handler, "check_usb", return_value=_TWO_DRIVES),
patch("os.path.exists", side_effect=lambda p: p.startswith(_TWO_DRIVES[1])),
- patch("subprocess.check_output", return_value=b""),
+ fake_jobs(),
patch("os.system"),
):
handler.user_restore_data(None)
+ _tick(handler)
snapshot()
+
+
+def test_v3_restore_offers_no_way_out_while_running(v3_system: SystemFixture):
+ """A running restore is deliberately undismissable — unzip is overwriting data/ in
+ place and a half-restored tree cannot be undone, so there is no button to press."""
+ handler = v3_system.handler
+ _setup_main_panel(v3_system)
+
+ with (
+ patch.object(handler, "check_usb", return_value=[_TWO_DRIVES[0]]),
+ patch("os.path.exists", return_value=True),
+ fake_jobs(),
+ ):
+ handler.user_restore_data(None)
+ panel = handler.lcd.pstack.find_panel_type(ArchiveProgressPanel)
+ assert panel is not None
+
+ assert panel.sel_ref is None
+ assert not panel._btn.visible
+ nav_click(handler)
+
+ assert handler.lcd.pstack.find_panel_type(ArchiveProgressPanel) is panel
+
+
+def test_v3_running_job_swallows_footswitches(v3_system: SystemFixture):
+ """The old blocking implementation froze the UI thread, so nothing could act mid-run.
+ Now that the job is off-thread the panel must swallow input itself — a footswitch
+ reaching the cascade would toggle a bypass (mod-ui writes data/) during the restore."""
+ handler = v3_system.handler
+ _setup_main_panel(v3_system)
+
+ with (
+ patch.object(handler, "check_usb", return_value=[_TWO_DRIVES[0]]),
+ patch("os.path.exists", return_value=True),
+ fake_jobs() as jobs,
+ ):
+ handler.user_restore_data(None)
+ fs = SwitchEvent(controller=v3_system.hw.footswitches[0], kind=SwitchEventKind.PRESS, timestamp=time.monotonic())
+ nav = SwitchEvent(controller=nav_encoder(handler), kind=SwitchEventKind.PRESS, timestamp=time.monotonic())
+
+ assert handler.lcd.handle(fs) is True
+ assert handler.lcd.handle(nav) is True
+
+ # Once finished the panel stops hoarding input.
+ jobs[0].finish(JobState.DONE)
+ _tick(handler)
+ assert handler.lcd.handle(fs) is False
+
+
+def test_v3_restore_button_is_nav_reachable_once_finished(v3_system: SystemFixture):
+ """When the job ends the button must be both visible and *selected* — it is the panel's
+ only selectable, so if add_sel_widget did not auto-select it the user would be trapped."""
+ handler = v3_system.handler
+ _setup_main_panel(v3_system)
+
+ with (
+ patch.object(handler, "check_usb", return_value=[_TWO_DRIVES[0]]),
+ patch("os.path.exists", return_value=True),
+ patch.object(handler, "restart_ui_stack") as mock_restart,
+ fake_jobs() as jobs,
+ ):
+ handler.user_restore_data(None)
+ panel = handler.lcd.pstack.find_panel_type(ArchiveProgressPanel)
+ assert panel is not None
+
+ jobs[0].finish(JobState.DONE)
+ _tick(handler)
+
+ assert panel.sel_ref is panel._btn
+ nav_click(handler)
+
+ mock_restart.assert_called_once_with()
+ assert handler.lcd.pstack.find_panel_type(ArchiveProgressPanel) is None
diff --git a/tests/v3/test_welcome_panel.py b/tests/v3/test_welcome_panel.py
index 6301c91a3..da11eb0bc 100644
--- a/tests/v3/test_welcome_panel.py
+++ b/tests/v3/test_welcome_panel.py
@@ -10,7 +10,10 @@
from unittest.mock import patch
import common.token as Token
+from modalapi.archive import JobState
+from tests.archive_fake import fake_jobs
from tests.types import SystemFixture
+from ui.archive_panel import ArchiveProgressPanel
from tests.v3.nav_helpers import nav_step, nav_click
from ui.welcome import WelcomePanel
@@ -72,7 +75,8 @@ def test_setup_noop(v3_system: SystemFixture, snapshot):
def test_restore_success(v3_system: SystemFixture, snapshot):
- """Restore calls load_settings before set_setting and pops welcome."""
+ """Restore calls load_settings before set_setting and pops welcome — but only once the
+ user closes the progress panel, since the restore now runs off the UI thread."""
handler = v3_system.handler
_open_welcome(v3_system)
@@ -80,12 +84,55 @@ def test_restore_success(v3_system: SystemFixture, snapshot):
with (
patch.object(handler, "check_usb", return_value=["/media/USB/backups"]),
patch("os.path.exists", return_value=True),
- patch("subprocess.check_output", return_value=b""),
+ patch.object(handler, "restart_ui_stack") as mock_restart,
+ fake_jobs() as jobs,
):
nav_step(handler, 1)
nav_click(handler)
handler.poll_lcd_updates()
+ # Welcome survives behind the progress panel until the restore finishes.
+ assert handler.lcd.pstack.find_panel_type(WelcomePanel) is not None
+
+ panel = handler.lcd.pstack.find_panel_type(ArchiveProgressPanel)
+ assert panel is not None
+ jobs[0].finish(JobState.DONE)
+ handler.poll_lcd_updates()
+ panel._on_button()
+ handler.poll_lcd_updates()
+ mock_restart.assert_called_once_with()
+
snapshot("restore_success")
assert handler.lcd.pstack.find_panel_type(WelcomePanel) is None
+
+
+def test_restore_failure_keeps_welcome_and_never_restarts(v3_system: SystemFixture):
+ """A failed restore from the welcome screen must leave the user where they started:
+ welcome still up, WELCOME_SEEN unset so it reappears, and no restart into a
+ half-written data/."""
+ handler = v3_system.handler
+
+ _open_welcome(v3_system)
+
+ with (
+ patch.object(handler, "check_usb", return_value=["/media/USB/backups"]),
+ patch("os.path.exists", return_value=True),
+ patch.object(handler, "restart_ui_stack") as mock_restart,
+ fake_jobs() as jobs,
+ ):
+ nav_step(handler, 1)
+ nav_click(handler)
+ handler.poll_lcd_updates()
+
+ panel = handler.lcd.pstack.find_panel_type(ArchiveProgressPanel)
+ assert panel is not None
+ jobs[0].finish(JobState.FAILED, error="unzip: cannot find zipfile directory")
+ handler.poll_lcd_updates()
+ panel._on_button()
+ handler.poll_lcd_updates()
+
+ mock_restart.assert_not_called()
+ assert handler.lcd.pstack.find_panel_type(WelcomePanel) is not None
+ calls = handler.settings.set_setting.call_args_list # pyright: ignore[reportAttributeAccessIssue]
+ assert Token.WELCOME_SEEN not in [c.args[0] for c in calls]
diff --git a/ui/archive_panel.py b/ui/archive_panel.py
new file mode 100644
index 000000000..06290dd64
--- /dev/null
+++ b/ui/archive_panel.py
@@ -0,0 +1,205 @@
+# This file is part of pi-stomp.
+#
+# pi-stomp is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# pi-Stomp is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with pi-stomp. If not, see .
+
+"""Modal progress panel for USB backup and restore.
+
+ ┌───────────────────────────────────────────────┐
+ │ Backing up │
+ │ STAGE_LEFT │
+ │ ▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓░░░░░░░░░░░░░░░░░░░░░░░░░░░ │
+ │ 0:52 -2:14 │
+ │ 268MB / 344MB │
+ │ [ Cancel ] │
+ └───────────────────────────────────────────────┘
+
+A backup cancels cleanly — the script builds a temp archive and renames it,
+so the previous good backup survives. A restore unzips over data/ in place and
+therefore offers no cancel; a half-restored tree cannot be undone.
+"""
+
+import os
+import time
+
+from uilib.box import Box
+from uilib.config import Config
+from uilib.label import Label
+from uilib.panel import Panel
+from uilib.misc import get_text_size
+from uilib.progress_bar import STEEL_LABELS, STEEL_STOPS, ProgressBarWidget
+from uilib.text import Button
+
+from pistomp.input.event import ControllerEvent
+
+from modalapi.archive import ArchiveJob, JobState
+
+_W, _H = 320, 240
+_BTN_W, _BTN_H, _BTN_GAP, _BTN_PAD = 92, 28, 6, 14
+_BTN_Y = _H - _BTN_H - _BTN_GAP
+_BAR_Y = 72
+_BAR_H = 88
+_MARGIN = 12
+_FILE_Y = _BAR_Y + _BAR_H + 2
+_STATUS_Y = _FILE_Y + 18
+
+_FG = (235, 235, 235)
+_DIM = (150, 150, 150)
+_FILE = (120, 145, 165)
+_ERR = (230, 90, 80)
+_OK = (110, 205, 120)
+
+# Below this the ETA is noise — a few small .ttl files land before the first model.
+_ETA_MIN_PROGRESS = 0.03
+
+
+def _mb(num_bytes: int) -> str:
+ return f"{num_bytes / 1_000_000:.0f}MB"
+
+
+def _ellipsize(text: str, font, max_w: int) -> str:
+ if get_text_size(text, font)[0] <= max_w:
+ return text
+ while text and get_text_size(text + "…", font)[0] > max_w:
+ text = text[:-1]
+ return text + "…"
+
+
+class ArchiveProgressPanel(Panel):
+ def __init__(
+ self,
+ *,
+ title: str,
+ noun: str,
+ subtitle: str,
+ job: ArchiveJob,
+ on_dismiss,
+ cancellable: bool,
+ done_label: str = "Close",
+ ) -> None:
+ super().__init__(
+ box=Box.xywh(0, 0, _W, _H),
+ auto_destroy=True,
+ no_dim=True,
+ opaque=True,
+ persist_on_board_change=True,
+ bkgnd_color=(0, 0, 0),
+ fgnd_color=_FG,
+ )
+ self._job = job
+ self._on_dismiss = on_dismiss
+ self._cancellable = cancellable
+ self._done_label = done_label
+ self._noun = noun
+ self._started = time.monotonic()
+ self._last_state = JobState.RUNNING
+
+ cfg = Config()
+ font = cfg.get_font("default")
+ title_font = cfg.get_font("default_title")
+ small = cfg.get_font("small")
+
+ self._title_lbl = Label(_MARGIN, 16, title_font, parent=self)
+ self._title_lbl.set_text(title, _FG)
+ self._sub_lbl = Label(_MARGIN, 44, small, parent=self)
+ self._sub_lbl.set_text(_ellipsize(subtitle, small, _W - 2 * _MARGIN), _DIM)
+
+ self._bar = ProgressBarWidget(
+ box=Box.xywh(0, _BAR_Y, _W, _BAR_H),
+ total_seconds=0.0,
+ font=font,
+ caption_font=small,
+ parent=self,
+ stops=STEEL_STOPS,
+ label_colors=STEEL_LABELS,
+ )
+
+ self._file_font = small
+ self._file_lbl = Label(_MARGIN, _FILE_Y, small, parent=self)
+ self._status_lbl = Label(_MARGIN, _STATUS_Y, small, parent=self)
+ self._status_lbl.set_text(f"0MB / {_mb(job.total_bytes)}", _DIM)
+
+ labels = ["Close", done_label] + (["Cancel"] if cancellable else [])
+ btn_w = max(_BTN_W, max(get_text_size(t, font)[0] for t in labels) + 2 * _BTN_PAD)
+ self._btn = Button(
+ box=Box.xywh((_W - btn_w) // 2, _BTN_Y, btn_w, _BTN_H),
+ text="Cancel" if cancellable else "",
+ font=font,
+ outline_radius=4,
+ parent=self,
+ action=lambda *_: self._on_button(),
+ )
+ self._btn.visible = cancellable
+ if cancellable:
+ self.add_sel_widget(self._btn)
+
+ def on_event(self, event: ControllerEvent) -> bool:
+ # Panel.handle resolves NAV before consulting on_event, so the axiom holds:
+ # everything reaching here is a footswitch/knob, and a bypass toggle or board
+ # change while unzip rewrites data/ would race the restore. Swallow until done.
+ return self._job.state is JobState.RUNNING
+
+ def _open_editor_for_selection(self) -> bool:
+ # NAV click with nothing selected (restore, mid-run): the click was aimed at
+ # this panel's empty selection, not at the board hidden behind it.
+ return self._job.state is JobState.RUNNING
+
+ @property
+ def job_state(self) -> JobState:
+ return self._job.state
+
+ def _on_button(self) -> None:
+ if self._job.state is JobState.RUNNING:
+ self._job.cancel()
+ self._title_lbl.set_text("Cancelling…", _DIM)
+ return
+ self._on_dismiss()
+
+ def _finish(self, state: JobState) -> None:
+ self._bar.set_done() if state is JobState.DONE else self._bar.freeze()
+ self._file_lbl.set_text("", _FILE)
+ if state is JobState.DONE:
+ self._title_lbl.set_text(f"{self._noun} complete", _OK)
+ self._status_lbl.set_text(f"{_mb(self._job.total_bytes)} processed", _DIM)
+ elif state is JobState.CANCELLED:
+ self._title_lbl.set_text(f"{self._noun} cancelled", _DIM)
+ self._status_lbl.set_text("Previous backup left intact", _DIM)
+ else:
+ self._title_lbl.set_text(f"{self._noun} failed", _ERR)
+ self._status_lbl.set_text(self._job.error.splitlines()[-1][:44] if self._job.error else "", _ERR)
+
+ self._btn.set_text(self._done_label if state is JobState.DONE else "Close")
+ if not self._btn.visible:
+ self._btn.visible = True
+ self.add_sel_widget(self._btn)
+ self.refresh()
+
+ def tick(self) -> None:
+ state = self._job.state
+ if state is not self._last_state:
+ self._last_state = state
+ if state is not JobState.RUNNING:
+ self._finish(state)
+ return
+ if state is not JobState.RUNNING:
+ return
+
+ p = self._job.progress()
+ elapsed = time.monotonic() - self._started
+ if p >= _ETA_MIN_PROGRESS:
+ self._bar.set_total(elapsed / p)
+ self._bar.set_progress(p)
+ self._status_lbl.set_text(f"{_mb(self._job.done_bytes)} / {_mb(self._job.total_bytes)}", _DIM)
+ entry = self._job.current_entry
+ if entry:
+ self._file_lbl.set_text(_ellipsize(os.path.basename(entry), self._file_font, _W - 2 * _MARGIN), _FILE)
diff --git a/uilib/progress_bar.py b/uilib/progress_bar.py
new file mode 100644
index 000000000..8be7ef2b7
--- /dev/null
+++ b/uilib/progress_bar.py
@@ -0,0 +1,177 @@
+# This file is part of pi-stomp.
+#
+# pi-stomp is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# pi-Stomp is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with pi-stomp. If not, see .
+
+from uilib.box import Box
+from uilib.misc import get_text_size
+from uilib.paint import PaintContext
+from uilib.widget import Widget
+
+ColorStops = list[tuple[float, tuple[int, int, int]]]
+
+# Green→red: reads as "time running out", for a capture with a fixed duration.
+HEAT_STOPS: ColorStops = [
+ (0.00, (0, 200, 75)),
+ (0.35, (120, 215, 0)),
+ (0.65, (230, 148, 0)),
+ (1.00, (215, 55, 10)),
+]
+
+# Dark→light steel blue, between the grid's wire blue and the Hz arc blue. For
+# work that merely takes as long as it takes — no urgency to signal.
+STEEL_STOPS: ColorStops = [
+ (0.00, (26, 58, 80)),
+ (0.50, (70, 150, 200)),
+ (1.00, (110, 200, 230)),
+]
+
+_BAR_DIM = 0.13 # brightness of unfilled segments
+
+# Elapsed / remaining label pairs, keyed to the bar they sit under.
+HEAT_LABELS = ((130, 118, 80), (205, 180, 110))
+STEEL_LABELS = ((92, 116, 136), (150, 190, 215))
+
+
+def fmt_time(seconds: float) -> str:
+ s = max(0, int(seconds))
+ return f"{s // 60}:{s % 60:02d}"
+
+
+class ProgressBarWidget(Widget):
+ """Segmented colour-gradient progress bar with elapsed/remaining time labels."""
+
+ _MARGIN = 12 # left/right inset
+ _BAR_Y = 30 # top of bar within widget
+ _BAR_H = 30 # bar height
+ _LABEL_GAP = 10 # gap between bar bottom and label top
+ # 42 segments of 5px with 2px gaps divide evenly (42*5 + 41*2 == 292), leaving
+ # 4px to split as an extra inset. Changing either without re-solving that
+ # identity brings back segments of uneven width.
+ _N_SEGS = 42
+ _SEG_GAP = 2
+
+ def __init__(
+ self,
+ box: Box,
+ total_seconds: float,
+ font,
+ caption_font,
+ parent: Widget,
+ stops: ColorStops | None = None,
+ label_colors: tuple[tuple[int, int, int], tuple[int, int, int]] = HEAT_LABELS,
+ ) -> None:
+ super().__init__(box=box, bkgnd_color=(0, 0, 0), parent=parent)
+ self._stops = HEAT_STOPS if stops is None else stops
+ self._label_colors = label_colors
+ self._total = total_seconds
+ self._progress = 0.0
+ self._frozen = False
+ self._elapsed = 0.0
+ self._remaining = total_seconds
+ self._font = font
+ self._caption_font = caption_font
+ self._segments = self._layout(int(box.width) - 2 * self._MARGIN)
+
+ @classmethod
+ def _layout(cls, inner_w: int) -> list[tuple[int, int]]:
+ # Uniform widths, with any leftover split between the two ends rather than
+ # stranded on the right. At 320 wide the leftover is zero by construction.
+ n = cls._N_SEGS
+ gap = cls._SEG_GAP
+ seg_w = max(1, (inner_w - (n - 1) * gap) // n)
+ x0 = (inner_w - (n * seg_w + (n - 1) * gap)) // 2
+ return [(x0 + i * (seg_w + gap), seg_w) for i in range(n)]
+
+ def set_total(self, total_seconds: float) -> None:
+ # Archiving has no known duration; callers revise this from throughput.
+ self._total = max(0.0, total_seconds)
+
+ def set_progress(self, progress: float) -> None:
+ if self._frozen:
+ return
+ p = max(0.0, min(1.0, progress))
+ old_filled = int(self._progress * self._N_SEGS)
+ self._progress = p
+ self._elapsed = p * self._total
+ self._remaining = self._total - self._elapsed
+ if int(p * self._N_SEGS) != old_filled:
+ self.refresh()
+
+ def freeze(self) -> None:
+ self._frozen = True
+
+ def set_done(self) -> None:
+ self._progress = 1.0
+ self._elapsed = self._total
+ self._remaining = 0.0
+ self._frozen = True
+
+ def reset(self) -> None:
+ self._progress = 0.0
+ self._elapsed = 0.0
+ self._remaining = self._total
+ self._frozen = False
+
+ def advance_rotation(self, dt: float) -> None:
+ pass
+
+ def _color_at(self, t: float) -> tuple[int, int, int]:
+ stops = self._stops
+ if t <= stops[0][0]:
+ return stops[0][1]
+ if t >= stops[-1][0]:
+ return stops[-1][1]
+ for i in range(len(stops) - 1):
+ t0, c0 = stops[i]
+ t1, c1 = stops[i + 1]
+ if t0 <= t <= t1:
+ f = (t - t0) / (t1 - t0)
+ return (
+ int(c0[0] + f * (c1[0] - c0[0])),
+ int(c0[1] + f * (c1[1] - c0[1])),
+ int(c0[2] + f * (c1[2] - c0[2])),
+ )
+ return stops[-1][1]
+
+ def _draw(self, ctx: PaintContext) -> None:
+ n = self._N_SEGS
+ filled = int(self._progress * n)
+ bx = self._MARGIN
+ by = self._BAR_Y
+ ctx.fill((0, 0, 0))
+
+ for i, (sx, sw) in enumerate(self._segments):
+ t = i / (n - 1) if n > 1 else 0.0
+ r, g, b = self._color_at(t)
+ if i < filled:
+ color: tuple[int, int, int] = (r, g, b)
+ else:
+ color = (int(r * _BAR_DIM), int(g * _BAR_DIM), int(b * _BAR_DIM))
+ ctx.draw_rectangle(Box.xywh(bx + sx, by, sw, self._BAR_H), fill=color)
+
+ # No duration yet (archiving, before throughput is measurable) — 0:00/−0:00
+ # would read as a stalled job, so draw no labels at all.
+ if self._total <= 0:
+ return
+
+ label_y = by + self._BAR_H + self._LABEL_GAP
+ right_x = ctx.width - self._MARGIN
+
+ elapsed_col, remaining_col = self._label_colors
+ elapsed_str = fmt_time(self._elapsed)
+ ctx.draw_text((bx, label_y), elapsed_str, fill=elapsed_col, font=self._font)
+
+ remaining_str = f"−{fmt_time(self._remaining)}"
+ rw, _ = get_text_size(remaining_str, self._font)
+ ctx.draw_text((right_x - rw, label_y), remaining_str, fill=remaining_col, font=self._font)
diff --git a/uilib/text.py b/uilib/text.py
index cfcac6a6d..2e2a442c2 100644
--- a/uilib/text.py
+++ b/uilib/text.py
@@ -443,6 +443,16 @@ def __init__(self, **kwargs):
self.sel_width = self._get_arg(kwargs, "sel_width", 2)
super(Button, self).__init__(**kwargs)
+ @override
+ def _get_margins(self):
+ # TextWidget top-aligns at v_margin, which leaves a fixed-height button's
+ # label riding high. Centre in the leftover space unless told otherwise.
+ h_margin, v_margin = super()._get_margins()
+ if self.v_margin is None and self.box is not None and self.box.height > 0:
+ _, th = self._get_text_size()
+ v_margin = max(v_margin, int((self.box.height - self.outline - th) // 2))
+ return (h_margin, v_margin)
+
class PluginTile(TextWidget):
"""TextWidget for plugin grid tiles.
diff --git a/util/data-backup.sh b/util/data-backup.sh
index 1491f424a..1fda5d824 100755
--- a/util/data-backup.sh
+++ b/util/data-backup.sh
@@ -21,14 +21,27 @@ fi
# Check if the destination parent directory exists
dest_dir=$(dirname "$1")
-if [ -d "$dest_dir" ]; then
- echo "Backup of: $src_dir"
- pushd $src_dir
- sudo zip -rq $1 . -x ".lv2/*"
- popd
-else
- echo "Parent directory does not exist: $parent_dir"
+if [ ! -d "$dest_dir" ]; then
+ echo "Parent directory does not exist: $dest_dir"
exit 2
fi
-exit 0
\ No newline at end of file
+# zip updates an existing archive in place, so deleted files would linger forever.
+tmp="$1.tmp"
+trap 'rm -f "$tmp"' EXIT INT TERM
+
+echo "Backup of: $src_dir"
+pushd "$src_dir" > /dev/null || exit 2
+# -1 over the default -6: 18s vs 26s, for 6MB on a 344MB archive.
+zip -r -1 "$tmp" . -x ".lv2/*"
+rc=$?
+popd > /dev/null
+
+if [ $rc -ne 0 ]; then
+ exit $rc
+fi
+
+mv -f "$tmp" "$1" || exit 2
+trap - EXIT
+
+exit 0
diff --git a/util/data-restore.sh b/util/data-restore.sh
index b02d311bf..2342ba52d 100755
--- a/util/data-restore.sh
+++ b/util/data-restore.sh
@@ -26,8 +26,10 @@ fi
# Restore
backup=$(realpath "$1")
-pushd $target_dir
-unzip -o -u $backup
-popd
+pushd "$target_dir" > /dev/null || exit 2
+unzip -o "$backup"
+rc=$?
+popd > /dev/null
+exit $rc
exit 0
\ No newline at end of file