Files
b0hemia e0bc4785e9 diskOS installer: initial public beta
Flashes the diskOS custom UI onto the FiiO Snowsky Disc over Ingenic mask-ROM USB,
building the image from your own stock firmware. Runs from source via install.sh.
2026-08-26 15:26:14 +10:00

571 lines
25 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Tkinter/ttk graphical installer.
Architecture:
* All Tk work on the main thread; a single worker thread runs the engine.
* The worker drives a QueueReporter; the GUI drains the queue via root.after.
* A blocking confirm() gate: the worker asks, the main thread shows a modal
acknowledgement dialog, the worker resumes on the user's decision.
* Explicit operation states; no Cancel once the destructive flash starts;
window-close is intercepted while flashing.
"""
import os
import queue
import sys
import threading
import time
from diskos_installer import __version__, bundle, platform_probe, service, state
from diskos_installer.reporter import QueueReporter
ACCENT = "#FF375F" # diskOS accent
BG = "#1c1c1e"
FG = "#f2f2f7"
SUBFG = "#9a9aa0"
PANEL = "#2c2c2e"
WARN = "#ffcc00"
OKC = "#34c759"
ERRC = "#ff453a"
# operation states
IDLE, RUNNING, CONFIRM, SUCCESS, FAILED = "idle", "running", "confirm", "success", "failed"
class App:
def __init__(self, root):
self.root = root
self.q = queue.Queue()
self.reporter = QueueReporter(self.q)
self.worker = None
self.state = IDLE
self.flashing = False # True only during the destructive phase
self._confirm_event = threading.Event()
self._confirm_result = False
self.start_time = None
root.title("diskOS Installer")
root.configure(bg=BG)
root.minsize(560, 640)
root.protocol("WM_DELETE_WINDOW", self._on_close)
self._build_style()
self._build_widgets()
self._refresh_state_labels()
self.root.after(80, self._drain)
self.root.after(1000, self._tick_elapsed)
self.root.after(4000, self._tick_device)
# ---- styling -----------------------------------------------------------
def _build_style(self):
import tkinter.font as tkfont
from tkinter import ttk
self.ttk = ttk
st = ttk.Style()
try:
st.theme_use("clam") # consistent across Linux/macOS
except Exception:
pass
# Derive from the platform's NAMED fonts (always present) rather than
# hard-coding families that may be absent (Helvetica/Menlo on Linux).
def _font(base, size, weight="normal"):
f = tkfont.nametofont(base).copy()
f.configure(size=size, weight=weight)
return f
self.f_title = _font("TkDefaultFont", 20, "bold")
self.f_h = _font("TkDefaultFont", 12, "bold")
self.f_b = _font("TkDefaultFont", 11)
self.f_mono = _font("TkFixedFont", 9)
st.configure("TFrame", background=BG)
st.configure("Panel.TFrame", background=PANEL)
st.configure("TLabel", background=BG, foreground=FG, font=self.f_b)
st.configure("Sub.TLabel", background=BG, foreground=SUBFG)
st.configure("Panel.TLabel", background=PANEL, foreground=FG)
st.configure("H.TLabel", background=BG, foreground=FG, font=self.f_h)
st.configure("TRadiobutton", background=BG, foreground=FG, font=self.f_b)
st.map("TRadiobutton", background=[("active", BG)])
st.configure("TCheckbutton", background=BG, foreground=FG, font=self.f_b)
st.map("TCheckbutton", background=[("active", BG)])
st.configure("Accent.TButton", font=self.f_h, padding=10)
st.configure("TButton", padding=6)
st.configure("diskos.Horizontal.TProgressbar", background=ACCENT, troughcolor=PANEL)
# ---- layout ------------------------------------------------------------
def _build_widgets(self):
ttk = self.ttk
pad = {"padx": 16}
head = ttk.Frame(self.root)
head.pack(fill="x", pady=(16, 4), **pad)
ttk.Label(head, text="diskOS Installer", font=self.f_title,
background=BG, foreground=FG).pack(anchor="w")
self.host_lbl = ttk.Label(head, text="", style="Sub.TLabel")
self.host_lbl.pack(anchor="w")
# mode
self.mode_frame = ttk.Frame(self.root)
self.mode_frame.pack(fill="x", pady=(10, 4), **pad)
self.mode = __import__("tkinter").StringVar(value="install")
for val, txt in (("install", "Install diskOS"),
("restore", "Remove diskOS (restore stock)"),
("remove", "Uninstall this tool")):
ttk.Radiobutton(self.mode_frame, text=txt, value=val, variable=self.mode,
command=self._on_mode).pack(anchor="w", pady=2)
# config panel (per-mode)
self.cfg = ttk.Frame(self.root, style="Panel.TFrame")
self.cfg.pack(fill="x", pady=10, **pad)
self._build_config()
# primary action
self.action_btn = ttk.Button(self.root, text="Install diskOS…",
style="Accent.TButton", command=self._on_action)
self.action_btn.pack(fill="x", pady=(4, 8), **pad)
# progress
prog = ttk.Frame(self.root)
prog.pack(fill="both", expand=True, **pad)
self.phase_lbl = ttk.Label(prog, text="Ready.", style="H.TLabel")
self.phase_lbl.pack(anchor="w")
self.status_lbl = ttk.Label(prog, text="", style="Sub.TLabel")
self.status_lbl.pack(anchor="w", pady=(0, 6))
self.bar = ttk.Progressbar(prog, style="diskos.Horizontal.TProgressbar",
mode="determinate", maximum=100)
self.bar.pack(fill="x")
self.elapsed_lbl = ttk.Label(prog, text="", style="Sub.TLabel")
self.elapsed_lbl.pack(anchor="w", pady=(2, 6))
self.warn_banner = __import__("tkinter").Label(
prog, text="", bg=WARN, fg="#111", font=self.f_h, anchor="center")
# packed only while flashing
import tkinter as tk
logwrap = ttk.Frame(prog)
logwrap.pack(fill="both", expand=True, pady=(4, 12))
self.log = tk.Text(logwrap, height=10, bg="#111", fg="#cfcfd4",
font=self.f_mono, wrap="word", relief="flat",
insertbackground=FG)
sb = ttk.Scrollbar(logwrap, command=self.log.yview)
self.log.configure(yscrollcommand=sb.set, state="disabled")
self.log.pack(side="left", fill="both", expand=True)
sb.pack(side="right", fill="y")
def _build_config(self):
for w in self.cfg.winfo_children():
w.destroy()
ttk = self.ttk
import tkinter as tk
m = self.mode.get()
inner = ttk.Frame(self.cfg, style="Panel.TFrame")
inner.pack(fill="x", padx=12, pady=12)
if m == "install":
ttk.Label(inner, text="Your FiiO firmware (.zip):", style="Panel.TLabel").grid(
row=0, column=0, sticky="w")
self.fw_var = tk.StringVar()
ttk.Entry(inner, textvariable=self.fw_var, width=44).grid(row=1, column=0, sticky="we", pady=(2, 8))
ttk.Button(inner, text="Browse…", command=self._pick_firmware).grid(row=1, column=1, padx=(8, 0))
ttk.Label(inner, text="Variant:", style="Panel.TLabel").grid(row=2, column=0, sticky="w")
self.variant = tk.StringVar(value="public")
vr = ttk.Frame(inner, style="Panel.TFrame")
vr.grid(row=3, column=0, sticky="w")
ttk.Radiobutton(vr, text="Public (recommended)", value="public",
variable=self.variant, command=self._variant_warn).pack(side="left")
ttk.Radiobutton(vr, text="Dev (root shell)", value="dev",
variable=self.variant, command=self._variant_warn).pack(side="left", padx=(12, 0))
self.variant_warn_lbl = tk.Label(
inner, text="", bg=PANEL, fg=WARN, font=self.f_b, justify="left", wraplength=460)
self.variant_warn_lbl.grid(row=4, column=0, sticky="w", pady=(4, 0))
inner.columnconfigure(0, weight=1)
elif m == "restore":
have = state.stock_image_exists() # fast; full verify happens at restore time
msg = ("A saved bone-stock image is available - restore is one click."
if have else
"No saved stock image. Pick your FiiO firmware .zip so I can rebuild it.")
ttk.Label(inner, text=msg, style="Panel.TLabel", wraplength=460).grid(
row=0, column=0, columnspan=2, sticky="w")
if not have:
self.fw_var = tk.StringVar()
ttk.Entry(inner, textvariable=self.fw_var, width=44).grid(row=1, column=0, sticky="we", pady=(8, 0))
ttk.Button(inner, text="Browse…", command=self._pick_firmware).grid(row=1, column=1, padx=(8, 0), pady=(8, 0))
inner.columnconfigure(0, weight=1)
else: # remove
ttk.Label(inner, wraplength=460, style="Panel.TLabel",
text=("Deletes this installer's saved files from THIS computer. "
"Nothing was installed system-wide. If diskOS is still on the "
"device, use ‘Remove diskOS’ first.")).pack(anchor="w")
# ---- helpers -----------------------------------------------------------
def _pick_firmware(self):
from tkinter import filedialog
p = filedialog.askopenfilename(title="Select FiiO firmware .zip",
filetypes=[("Firmware zip", "*.zip"), ("All", "*")])
if p:
self.fw_var.set(p)
def _variant_warn(self):
lbl = getattr(self, "variant_warn_lbl", None)
if lbl is None:
return
if self.variant.get() == "dev":
lbl.config(text="⚠ Dev installs an ALWAYS-ON PASSWORDLESS ROOT SHELL over USB - "
"anyone with physical access gets root on every boot. It bypasses "
"device security. Development devices only, never an everyday one.")
else:
lbl.config(text="")
def _refresh_state_labels(self):
o, a = platform_probe.host()
n = platform_probe.maskrom_count()
dev = {0: "no device in flash mode", -1: "USB not enumerable"}.get(n, f"{n} device(s) in flash mode")
self.host_lbl.config(text=f"{o}-{a} · v{__version__} · {dev}")
def _on_mode(self):
self._build_config()
self.action_btn.config(text={"install": "Install diskOS…",
"restore": "Remove diskOS (restore stock)…",
"remove": "Uninstall this tool"}[self.mode.get()])
def _log(self, line, color=None):
self.log.config(state="normal")
self.log.insert("end", line + "\n")
# cap the log so it can't grow unbounded
if int(self.log.index("end-1c").split(".")[0]) > 500:
self.log.delete("1.0", "100.0")
self.log.see("end")
self.log.config(state="disabled")
# ---- action / worker ---------------------------------------------------
def _on_action(self):
if self.state == RUNNING:
return
m = self.mode.get()
params = {}
if m == "install":
fw = getattr(self, "fw_var", None) and self.fw_var.get().strip()
if not fw:
self._log("Choose your FiiO firmware .zip first.")
return
params = {"firmware": fw, "variant": self.variant.get()}
fn = service.do_install
elif m == "restore":
fw = getattr(self, "fw_var", None)
params = {"firmware": (fw.get().strip() if fw else None)}
fn = service.do_restore
else:
params = {"force": False}
fn = service.do_remove
self.state = RUNNING
self.start_time = time.monotonic()
self._set_controls(False)
self.bar.config(mode="determinate", value=0)
self.status_lbl.config(text="")
self.elapsed_lbl.config(text="")
self.phase_lbl.config(text="Starting…")
self._clear_finish_extra()
def run():
try:
r = fn(params, self.reporter, self._worker_confirm)
self.q.put(("done", r))
except Exception as e: # BuildError/FlashError/etc.
self.q.put(("failed", {"error": str(e)}))
self.worker = threading.Thread(target=run, daemon=True)
self.worker.start()
def _set_controls(self, enabled):
"""Enable/disable all pre-flight controls (mode + config + action) so nothing
can be changed or re-triggered while a worker is running."""
st = "normal" if enabled else "disabled"
def walk(w):
for c in w.winfo_children():
cls = c.winfo_class()
if cls in ("TRadiobutton", "TButton", "TEntry", "TCheckbutton",
"Radiobutton", "Button", "Entry", "Checkbutton"):
try:
c.config(state=st)
except Exception:
pass
walk(c)
walk(self.mode_frame)
walk(self.cfg)
self.action_btn.config(state=st)
def _clear_finish_extra(self):
w = getattr(self, "_finish_extra", None)
if w is not None:
w.destroy()
self._finish_extra = None
def _worker_confirm(self, summary):
"""Called ON THE WORKER THREAD. Ask the UI (main thread) and block."""
self._confirm_result = False
self._confirm_event.clear()
self.q.put(("confirm", summary))
self._confirm_event.wait()
return self._confirm_result
def _show_confirm_dialog(self, summary):
import tkinter as tk
from tkinter import ttk
W = 480
dlg = tk.Toplevel(self.root)
dlg.title("Confirm")
dlg.configure(bg=BG)
dlg.transient(self.root)
dlg.resizable(False, False)
destructive = summary.get("action") in ("install", "restore-stock")
body = ttk.Frame(dlg)
body.pack(fill="both", expand=True, padx=18, pady=16)
ttk.Label(body, text=summary.get("action", "confirm").upper(),
style="H.TLabel").pack(anchor="w", pady=(0, 6))
for k in ("variant", "duration", "consequence"):
if summary.get(k):
ttk.Label(body, text=f"{k}: {summary[k]}", style="Sub.TLabel",
wraplength=W - 36, justify="left").pack(anchor="w", pady=1)
def _resolve(val):
# Mark the destructive boundary SYNCHRONOUSLY (main thread) the instant the
# user accepts - before the worker unblocks and spawns usbboot - so the
# window can't be closed in the gap before the 'phase' event is drained.
if val and destructive:
self.flashing = True
self._confirm_result = val
dlg.destroy()
self._confirm_event.set()
# button row FIRST (both buttons are direct children of it)
row = ttk.Frame(body)
row.pack(side="bottom", fill="x", pady=(14, 0))
ttk.Button(row, text="Cancel", command=lambda: _resolve(False)).pack(side="right", padx=(8, 0))
begin = ttk.Button(row, text=("Begin 60-90 minute flash" if destructive else "Proceed"),
style="Accent.TButton", command=lambda: _resolve(True))
begin.pack(side="right")
ack = tk.BooleanVar(value=not destructive)
if destructive:
# classic tk.Checkbutton (ttk.Checkbutton has no wraplength)
tk.Checkbutton(body, variable=ack, wraplength=W - 40, justify="left",
text="I understand this rewrites my device and must not be interrupted.",
bg=BG, fg=FG, selectcolor=PANEL, activebackground=BG,
activeforeground=FG, highlightthickness=0, font=self.f_b,
command=lambda: begin.config(state=("normal" if ack.get() else "disabled"))
).pack(anchor="w", pady=(12, 8))
begin.config(state="disabled")
# size to content, then centre over the main window
dlg.update_idletasks()
h = dlg.winfo_reqheight()
self.root.update_idletasks()
px, py = self.root.winfo_rootx(), self.root.winfo_rooty()
pw = self.root.winfo_width()
dlg.geometry(f"{W}x{h}+{px + max(0, (pw - W) // 2)}+{py + 110}")
dlg.grab_set()
dlg.protocol("WM_DELETE_WINDOW", lambda: _resolve(False))
dlg.bind("<Escape>", lambda e: _resolve(False))
dlg.bind("<Return>", lambda e: _resolve(True) if str(begin["state"]) == "normal" else None)
(begin if not destructive else dlg).focus_set()
# ---- queue drain (main thread) ----------------------------------------
def _drain(self):
budget = 200 # bounded per tick so a flood of events can't starve Tk
try:
while budget > 0:
kind, payload = self.q.get_nowait()
budget -= 1
try:
self._handle(kind, payload)
except Exception as e: # one malformed event must not stop the drain
self._log(f"[ui error handling {kind}: {e}]")
except queue.Empty:
pass
self.root.after(80, self._drain)
def _handle(self, kind, p):
if kind == "phase":
self.phase_lbl.config(text=p["name"])
if p.get("destructive"):
self.flashing = True
self.bar.config(mode="indeterminate")
self.bar.start(12)
self.warn_banner.config(text="DO NOT DISCONNECT THE DEVICE OR LET THE HOST SLEEP")
self.warn_banner.pack(fill="x", pady=6, before=self.log.master)
self._log("▶ " + p["name"])
elif kind == "status":
self.status_lbl.config(text=p["message"])
self._log(p["message"])
elif kind == "log":
self._log(" " + p["line"])
elif kind == "progress":
if self.bar["mode"] == "indeterminate":
self.bar.stop(); self.bar.config(mode="determinate")
total = p["total"] or 1
self.bar.config(maximum=total, value=p["completed"])
elif kind == "indeterminate":
if p["active"]:
self.bar.config(mode="indeterminate"); self.bar.start(12)
self.status_lbl.config(text=p.get("note", ""))
else:
self.bar.stop(); self.bar.config(mode="determinate")
elif kind == "warning":
self._log("! " + p["message"])
elif kind == "ok":
self._log("✓ " + p["message"])
elif kind == "error":
self._log("✗ " + p["message"])
elif kind == "confirm":
self._show_confirm_dialog(p)
elif kind == "done":
self._finish(p)
elif kind == "failed":
self._fail(p.get("error", "unknown error"))
def _finish(self, r):
self.flashing = False
if self.bar["mode"] == "indeterminate":
self.bar.stop()
self.warn_banner.pack_forget()
self._set_controls(True)
if r.get("aborted"):
self.state = IDLE
self.phase_lbl.config(text="Cancelled - device unchanged.")
elif r.get("errors"): # e.g. uninstall couldn't remove everything
self.state = FAILED
self.phase_lbl.config(text="Finished with problems - see log.")
else:
self.state = SUCCESS
self.bar.config(mode="determinate", maximum=100, value=100)
if self.mode.get() == "install":
self.phase_lbl.config(text="Flashed ✓ One more step ↓")
self._install_followup()
elif self.mode.get() == "remove":
self.phase_lbl.config(text="Removed ✓ Delete the app to finish.")
else:
self.phase_lbl.config(text="Done ✓ Power-cycle the device.")
self._refresh_state_labels()
def _install_followup(self):
"""The UI is embedded in the flashed image, so first boot installs diskOS on its own -
no microSD step. Offer an OPTIONAL SD copy only as a recovery fallback."""
import tkinter as tk
from tkinter import ttk
self._clear_finish_extra()
box = tk.Frame(self.root, bg=OKC)
box.pack(fill="x", padx=16, pady=(0, 8))
self._finish_extra = box
tk.Label(box, bg=OKC, fg="#111", font=self.f_h, justify="left", wraplength=500,
text="Done - just power-cycle the device. diskOS is embedded in the flash and "
"installs itself on first boot; no microSD step needed.").pack(
anchor="w", padx=10, pady=(8, 4))
tk.Label(box, bg=OKC, fg="#111", font=self.f_b, justify="left", wraplength=500,
text="Optional: you can also stage a fallback copy on a microSD (used only if the "
"embedded copy is ever unreadable).").pack(anchor="w", padx=10, pady=(0, 2))
ttk.Button(box, text="Copy fallback UI to microSD…", command=self._export_ui).pack(
anchor="w", padx=10, pady=(0, 8))
def _export_ui(self):
from tkinter import filedialog, messagebox
src = bundle.data("mq_ui", required=False)
if not src:
messagebox.showerror("diskOS Installer", "bundled mq_ui not found.")
return
d = filedialog.askdirectory(title="Select your microSD card (root)")
if not d:
return
try:
dest_dir = os.path.join(d, "diskos")
os.makedirs(dest_dir, exist_ok=True)
dest = os.path.join(dest_dir, "mq_ui")
with open(src, "rb") as s, open(dest, "wb") as o:
o.write(s.read())
os.chmod(dest, 0o755)
self._log(f"✓ copied UI to {dest}")
messagebox.showinfo("diskOS Installer",
f"Copied a fallback diskOS UI to:\n{dest}\n\nThis is optional - the "
"device installs the embedded copy on its own. The card is only used "
"if that embedded copy is ever unreadable.")
except OSError as e:
messagebox.showerror("diskOS Installer", f"Could not copy to the card:\n{e}")
def _fail(self, msg):
self.flashing = False
try:
self.bar.stop()
except Exception:
pass
self.warn_banner.pack_forget()
self.state = FAILED
self._set_controls(True)
self.phase_lbl.config(text="Failed")
self._log("✗ " + msg)
from tkinter import messagebox
messagebox.showerror("diskOS Installer", msg)
def _tick_elapsed(self):
if self.state == RUNNING and self.start_time:
secs = int(time.monotonic() - self.start_time)
self.elapsed_lbl.config(text=f"elapsed {secs // 60}m{secs % 60:02d}s")
self.root.after(1000, self._tick_elapsed)
def _tick_device(self):
# keep the device-status line fresh, but never scan USB mid-operation
if self.state != RUNNING:
self._refresh_state_labels()
self.root.after(4000, self._tick_device)
def _on_close(self):
from tkinter import messagebox
if self.flashing:
messagebox.showwarning(
"Flash in progress",
"A flash is in progress. Closing now can leave the device needing a "
"mask-ROM recovery. Please wait until it finishes.")
return
if self.state == RUNNING:
if not messagebox.askokcancel(
"Operation in progress",
"An operation is still running. Close anyway?"):
return
self.root.destroy()
def main():
try:
import tkinter as tk
except Exception as e:
sys.stderr.write(f"diskOS Installer: Tkinter unavailable ({e}). Use the CLI: "
"diskos-installer --help\n")
return 2
try:
root = tk.Tk()
App(root)
root.mainloop()
return 0
except Exception as e:
# A packaged GUI hides early tracebacks - persist one and show a dialog.
import tempfile
import traceback
# Unique, 0600 crash log (a fixed name in a world-writable temp dir could be pre-placed as a
# symlink to truncate an arbitrary target). The exact path is shown in the message below.
try:
fd, logp = tempfile.mkstemp(prefix="diskos-installer-startup-", suffix=".log")
with os.fdopen(fd, "w") as f:
f.write(traceback.format_exc())
except OSError:
logp = "(could not write a log file)"
sys.stderr.write(f"diskOS Installer failed to start: {e}\n(log: {logp})\n"
"Try the CLI: diskos-installer doctor\n")
try:
import tkinter.messagebox as mb
mb.showerror("diskOS Installer",
f"Failed to start:\n{e}\n\nDetails: {logp}\n"
"You can still use the command line: diskos-installer doctor")
except Exception:
pass
return 1
if __name__ == "__main__":
sys.exit(main())