Update LUKS unlock wait behavior

_scripts/unlock-luks-after-install.py

@@ -15,6 +15,7 @@ from __future__ import annotations
 import argparse
 import random
 import subprocess
+import sys
 import threading
 import time
 from http.server import BaseHTTPRequestHandler, HTTPServer
@@ -46,11 +47,17 @@ def main() -> int:
         required=True,
         help="Path to directory containing variables.pkr.hcl (also passed to mise build).",
     )
+    parser.add_argument(
+        "--luks-wait-seconds",
+        type=int,
+        default=45,
+        help="Seconds to wait before sending the LUKS password (default: 45).",
+    )
     args = parser.parse_args()
 
     script_root = Path(__file__).resolve().parents[1]
     variables_common_path = script_root / "variables-common.pkr.hcl"
-    credentials_path = script_root / "debian/13-trixie/credentials.auto.pkrvars.hcl"
+    credentials_path = script_root / "credentials.auto.pkrvars.hcl"
     vars_dir = Path(args.template)
     if not vars_dir.is_absolute():
         vars_dir = script_root / vars_dir
@@ -101,16 +108,66 @@ def main() -> int:
 
     port, httpd = find_random_port()
 
+    def stream_colors(stream: object) -> tuple[str, str]:
+        color = ""
+        reset = ""
+        is_tty = getattr(stream, "isatty", None)
+        if callable(is_tty) and is_tty():
+            if stream is sys.stderr:
+                color = "\033[31m"
+            else:
+                color = "\033[36m"
+            reset = "\033[0m"
+        return color, reset
+
+    def log(message: str, stream: object = sys.stdout) -> None:
+        color, reset = stream_colors(stream)
+        stream.write(f"{color}[luks-unlock-wrapper] {message}{reset}\n")
+        stream.flush()
+
+    def write_status(message: str, stream: object = sys.stdout, *, newline: bool) -> None:
+        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()
+
     def serve() -> None:
         httpd.serve_forever()
 
     server_thread = threading.Thread(target=serve, daemon=True)
     server_thread.start()
 
-    print(f"Listening for POST on /install_finished at port {port}")
+    log(f"Listening for POST on /install_finished at port {port}")
 
     build_cmd = ["mise", "build", args.template, "-i", str(port)]
-    build_proc = subprocess.Popen(build_cmd)
+    build_proc = subprocess.Popen(
+        build_cmd,
+        stdout=subprocess.PIPE,
+        stderr=subprocess.PIPE,
+        text=True,
+        bufsize=1,
+    )
+
+    def relay_stream(stream: object, prefix: str, target: object) -> None:
+        if not stream:
+            return
+        for line in stream:
+            target.write(f"{prefix} {line}")
+            target.flush()
+
+    stdout_thread = threading.Thread(
+        target=relay_stream, args=(build_proc.stdout, "[packer]", sys.stdout), daemon=True
+    )
+    stderr_thread = threading.Thread(
+        target=relay_stream, args=(build_proc.stderr, "[packer]", sys.stderr), daemon=True
+    )
+    stdout_thread.start()
+    stderr_thread.start()
 
     notified = False
     action_started = False
@@ -148,28 +205,52 @@ def main() -> int:
             try:
                 response = proxmox_request("GET", "/version")
                 response.raise_for_status()
-                print("Authenticated to Proxmox VE API.")
+                log("Authenticated to Proxmox VE API.")
                 break
             except Exception as exc:
-                print(f"Proxmox auth failed: {exc}. Retrying in {retry_delay}s.")
+                log(
+                    f"Proxmox auth failed: {exc}. Retrying in {retry_delay}s.",
+                    stream=sys.stderr,
+                )
                 time.sleep(retry_delay)
                 retry_delay += 1
 
         try:
-            print("Waiting 45 seconds before sending LUKS password.")
-            time.sleep(45)
+            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]
+            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)
+                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")
-            print("Sent LUKS password and Enter.")
+            log("Unlocking encrypted disk. Entering LUKS password.")
         except Exception as exc:
-            print(f"Post-install actions failed after auth: {exc}")
+            log(f"Post-install actions failed after auth: {exc}", stream=sys.stderr)
 
     try:
         while True:
             if server_event.is_set() and not notified:
-                print("Installation finished.\nRestarting.")
+                log("Installation finished.\nRestarting.")
                 notified = True
             if server_event.is_set() and not action_started:
                 action_started = True