From 6d351fcdb3c452c362d9ccc5df52115841bc57f6 Mon Sep 17 00:00:00 2001
From: phg <mail@philip-henning.com>
Date: Fri, 6 Feb 2026 10:40:11 +0100
Subject: [PATCH] Refactor LUKS unlock script to enhance spinner functionality
 and improve message colorization

---
 _scripts/unlock-luks-after-install.py | 59 ++++++++++++++-------------
 1 file changed, 31 insertions(+), 28 deletions(-)

diff --git a/_scripts/unlock-luks-after-install.py b/_scripts/unlock-luks-after-install.py
index cc98e86..af6d47d 100755
--- a/_scripts/unlock-luks-after-install.py
+++ b/_scripts/unlock-luks-after-install.py
@@ -4,6 +4,7 @@
 # dependencies = [
 #     "python-hcl2==4.*",
 #     "requests==2.*",
+#     "yaspin==3.*",
 # ]
 # ///
 
@@ -23,6 +24,7 @@ from pathlib import Path
 
 import hcl2
 import requests
+from yaspin import yaspin
 
 
 def load_hcl(path: Path) -> dict:
@@ -125,16 +127,11 @@ def main() -> int:
         stream.write(f"{color}[luks-unlock-wrapper] {message}{reset}\n")
         stream.flush()
 
-    def write_status(message: str, stream: object = sys.stdout, *, newline: bool) -> None:
+    def colorize_message(message: str, stream: object = sys.stdout) -> str:
         color, reset = stream_colors(stream)
-        is_tty = getattr(stream, "isatty", None)
-        prefix = f"{color}[luks-unlock-wrapper] "
-        suffix = f"{reset}\n" if newline else reset
-        if callable(is_tty) and is_tty():
-            stream.write(f"\r\033[2K{prefix}{message}{suffix}")
-        else:
-            stream.write(f"{prefix}{message}{suffix}")
-        stream.flush()
+        if not color:
+            return message
+        return f"{color}{message}{reset}"
 
     def serve() -> None:
         httpd.serve_forever()
@@ -215,42 +212,48 @@ def main() -> int:
                 time.sleep(retry_delay)
                 retry_delay += 1
 
+        spinner = None
         try:
             countdown_seconds = max(0, args.luks_wait_seconds)
-            # Braille spinner: 8 dots, one missing dot rotates clockwise.
-            full_mask = 0xFF  # dots 1-8
-            dot_bits = {
-                1: 0x01,
-                2: 0x02,
-                3: 0x04,
-                4: 0x08,
-                5: 0x10,
-                6: 0x20,
-                7: 0x40,
-                8: 0x80,
-            }
-            rotation = [1, 4, 5, 6, 8, 7, 3, 2]  # clockwise around the cell
-            spinner = [chr(0x2800 + (full_mask - dot_bits[dot])) for dot in rotation]
+            spinner = yaspin(
+                text=colorize_message(
+                    f"[luks-unlock-wrapper] {countdown_seconds // 60:02d}:{countdown_seconds % 60:02d}",
+                    sys.stdout,
+                ),
+                color="cyan",
+                stream=sys.stdout,
+            )
+            spinner.start()
             for remaining in range(countdown_seconds, -1, -1):
                 minutes, seconds = divmod(remaining, 60)
                 countdown = f"{minutes:02d}:{seconds:02d}"
-                frame = spinner[(countdown_seconds - remaining) % len(spinner)]
-                write_status(f"{frame} {countdown}", newline=False)
+                spinner.text = colorize_message(
+                    f"[luks-unlock-wrapper] {countdown}", sys.stdout
+                )
                 if remaining:
                     time.sleep(1)
-            write_status(f"{spinner[0]} 00:00", newline=True)
             for char in "packer":
                 send_key(char)
                 time.sleep(0.1)
             send_key("ret")
-            log("Unlocking encrypted disk. Entering LUKS password.")
+            spinner.text = colorize_message(
+                "[luks-unlock-wrapper] ✅ Unlocking encrypted disk. Entering LUKS password.",
+                sys.stdout,
+            )
+            spinner.ok("")
         except Exception as exc:
+            if spinner:
+                spinner.text = colorize_message(
+                    "[luks-unlock-wrapper] 💥 Post-install actions failed.",
+                    sys.stdout,
+                )
+                spinner.fail("")
             log(f"Post-install actions failed after auth: {exc}", stream=sys.stderr)
 
     try:
         while True:
             if server_event.is_set() and not notified:
-                log("Installation finished.\nRestarting.")
+                log("Installation finished. -- Restarting.")
                 notified = True
             if server_event.is_set() and not action_started:
                 action_started = True