waveform_gui.py

#!/usr/bin/env python3
"""
Simple Tkinter front-end for generate_trace_video.py.

Lets users pick an audio file, tweak a handful of options, and launch the
ffmpeg render without touching the command line.
"""
from __future__ import annotations

import queue
import shlex
import subprocess
import threading
import time
import webbrowser
from pathlib import Path
from types import SimpleNamespace
from typing import Optional

import tkinter as tk
from tkinter import filedialog, messagebox, scrolledtext, ttk

import generate_trace_video as gtv


class TraceVideoApp(ttk.Frame):
    def __init__(self, master: tk.Tk) -> None:
        super().__init__(master, padding=16)
        self.pack(fill="both", expand=True)

        self.style = ttk.Style()
        self.style.configure("Accent.TButton", font=("Helvetica", 12, "bold"), padding=6)
        self.style.configure("Hyperlink.TLabel", foreground="#2563eb")
        self.style.map("Hyperlink.TLabel", foreground=[("active", "#1d4ed8")], underline=[("active", 1), ("!active", 1)])

        self.output_queue: queue.Queue[str] = queue.Queue()
        self.render_thread: Optional[threading.Thread] = None
        self.current_process: Optional[subprocess.Popen[str]] = None
        self._cancel_requested = False

        self.audio_var = tk.StringVar()
        self.output_var = tk.StringVar()
        self.title_var = tk.StringVar()
        self.font_var = tk.StringVar(value="Helvetica")
        self.font_size_var = tk.StringVar(value="72")
        self.mode_var = tk.StringVar(value="line")
        self.wave_color_var = tk.StringVar(value=gtv.DEFAULT_WAVE_COLOR)
        self.bg_style_var = tk.StringVar(value="gradient")
        self.bg_color_var = tk.StringVar(value=gtv.DEFAULT_BACKGROUND_COLOR)
        self.gradient_var = tk.StringVar(value=gtv.DEFAULT_GRADIENT_COLORS)
        self.bg_image_var = tk.StringVar()
        self.preset_var = tk.StringVar(value="faster")
        self.dry_run_var = tk.BooleanVar(value=False)
        self.filter_threads_var = tk.StringVar(value="")
        self.fast_render_var = tk.BooleanVar(value=False)
        self.progress_var = tk.DoubleVar(value=0.0)
        self.status_var = tk.StringVar(value="Ready to render")
        self._last_progress = 0.0

        self._build_form()
        self.after(100, self._drain_output)

    def _build_form(self) -> None:
        self.columnconfigure(0, weight=1)
        self.rowconfigure(5, weight=1)

        files_frame = ttk.LabelFrame(self, text="Source & Output")
        files_frame.grid(row=0, column=0, sticky="ew", pady=(0, 12))
        files_frame.columnconfigure(1, weight=1)

        ttk.Label(files_frame, text="Audio file").grid(row=0, column=0, sticky="w", padx=(0, 8), pady=6)
        audio_entry = ttk.Entry(files_frame, textvariable=self.audio_var)
        audio_entry.grid(row=0, column=1, sticky="ew", pady=6)
        ttk.Button(files_frame, text="Browse…", command=self._choose_audio).grid(row=0, column=2, padx=(8, 0), pady=6)

        ttk.Label(files_frame, text="Output MP4").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=6)
        output_entry = ttk.Entry(files_frame, textvariable=self.output_var)
        output_entry.grid(row=1, column=1, sticky="ew", pady=6)
        ttk.Button(files_frame, text="Save as…", command=self._choose_output).grid(row=1, column=2, padx=(8, 0), pady=6)

        options_frame = ttk.Frame(self)
        options_frame.grid(row=1, column=0, sticky="ew", pady=(0, 12))
        options_frame.columnconfigure(0, weight=1)
        options_frame.columnconfigure(1, weight=1)

        display_frame = ttk.LabelFrame(options_frame, text="Waveform & Title")
        display_frame.grid(row=0, column=0, sticky="nsew", padx=(0, 8))
        display_frame.columnconfigure(1, weight=1)

        ttk.Label(display_frame, text="Title text").grid(row=0, column=0, sticky="w", padx=(0, 8), pady=6)
        ttk.Entry(display_frame, textvariable=self.title_var).grid(row=0, column=1, sticky="ew", pady=6)

        ttk.Label(display_frame, text="Font").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=6)
        ttk.Entry(display_frame, textvariable=self.font_var, width=18).grid(row=1, column=1, sticky="w", pady=6)
        ttk.Label(display_frame, text="Size").grid(row=1, column=2, sticky="w", padx=(12, 4))
        ttk.Entry(display_frame, textvariable=self.font_size_var, width=6).grid(row=1, column=3, sticky="w", pady=6)

        ttk.Label(display_frame, text="Waveform style").grid(row=2, column=0, sticky="w", padx=(0, 8), pady=6)
        ttk.Combobox(display_frame, textvariable=self.mode_var, values=("line", "point", "bar"), width=12, state="readonly").grid(
            row=2, column=1, sticky="w", pady=6
        )
        ttk.Label(display_frame, text="Trace colour").grid(row=2, column=2, sticky="w", padx=(12, 4))
        ttk.Entry(display_frame, textvariable=self.wave_color_var, width=12).grid(row=2, column=3, sticky="w", pady=6)

        background_frame = ttk.LabelFrame(options_frame, text="Background")
        background_frame.grid(row=0, column=1, sticky="nsew")
        background_frame.columnconfigure(1, weight=1)

        ttk.Label(background_frame, text="Style").grid(row=0, column=0, sticky="w", padx=(0, 8), pady=6)
        self.bg_style_combo = ttk.Combobox(
            background_frame,
            textvariable=self.bg_style_var,
            values=("gradient", "plasma", "solid"),
            width=12,
            state="readonly",
        )
        self.bg_style_combo.grid(row=0, column=1, sticky="w", pady=6)

        ttk.Label(background_frame, text="Solid colour").grid(row=1, column=0, sticky="w", padx=(0, 8), pady=6)
        self.bg_color_entry = ttk.Entry(background_frame, textvariable=self.bg_color_var, width=12)
        self.bg_color_entry.grid(row=1, column=1, sticky="w", pady=6)

        ttk.Label(background_frame, text="Gradient colours").grid(row=2, column=0, sticky="w", padx=(0, 8), pady=6)
        self.gradient_entry = ttk.Entry(background_frame, textvariable=self.gradient_var)
        self.gradient_entry.grid(row=2, column=1, sticky="ew", pady=6)

        ttk.Label(background_frame, text="Background image").grid(row=3, column=0, sticky="w", padx=(0, 8), pady=6)
        bg_img_entry = ttk.Entry(background_frame, textvariable=self.bg_image_var)
        bg_img_entry.grid(row=3, column=1, sticky="ew", pady=6)
        ttk.Button(background_frame, text="Browse…", command=self._choose_background).grid(
            row=3, column=2, padx=(8, 0), pady=6
        )

        perf_frame = ttk.LabelFrame(self, text="Performance & Advanced")
        perf_frame.grid(row=2, column=0, sticky="ew")
        perf_frame.columnconfigure(1, weight=1)

        ttk.Label(perf_frame, text="Encoder preset").grid(row=0, column=0, sticky="w", padx=(0, 8), pady=6)
        ttk.Combobox(
            perf_frame,
            textvariable=self.preset_var,
            values=("ultrafast", "superfast", "veryfast", "faster", "fast", "medium", "slow"),
            width=12,
        ).grid(row=0, column=1, sticky="w", pady=6)

        ttk.Label(perf_frame, text="Filter threads").grid(row=0, column=2, sticky="w", padx=(12, 4))
        ttk.Entry(perf_frame, textvariable=self.filter_threads_var, width=6).grid(row=0, column=3, sticky="w", pady=6)

        ttk.Checkbutton(
            perf_frame,
            text="Dry run (print ffmpeg command only)",
            variable=self.dry_run_var,
        ).grid(row=1, column=0, columnspan=4, sticky="w", pady=(0, 6))

        ttk.Checkbutton(
            perf_frame,
            text="Fast render (solid background + ultrafast encoder)",
            variable=self.fast_render_var,
            command=self._on_fast_toggle,
        ).grid(row=2, column=0, columnspan=4, sticky="w", pady=(0, 6))

        button_row = ttk.Frame(self)
        button_row.grid(row=3, column=0, sticky="ew", pady=(12, 8))
        button_row.columnconfigure(0, weight=1)
        button_row.columnconfigure(1, weight=0)
        button_row.columnconfigure(2, weight=0)

        self.render_btn = ttk.Button(
            button_row,
            text="Render Video",
            command=self._start_render,
            style="Accent.TButton",
        )
        self.render_btn.grid(row=0, column=0, sticky="w")

        self.cancel_btn = ttk.Button(
            button_row,
            text="Cancel",
            command=self._cancel_render,
            state="disabled",
        )
        self.cancel_btn.grid(row=0, column=1, padx=(12, 12))

        ttk.Button(button_row, text="Quit", command=self.master.destroy).grid(row=0, column=2, sticky="e")

        status_row = ttk.Frame(self)
        status_row.grid(row=4, column=0, sticky="ew", pady=(0, 8))
        status_row.columnconfigure(1, weight=1)
        ttk.Label(status_row, textvariable=self.status_var).grid(row=0, column=0, sticky="w")
        self.progress_bar = ttk.Progressbar(
            status_row,
            variable=self.progress_var,
            maximum=1.0,
            mode="determinate",
        )
        self.progress_bar.grid(row=0, column=1, sticky="ew", padx=(12, 0))

        self.log_widget = scrolledtext.ScrolledText(self, height=14, state="disabled")
        self.log_widget.grid(row=5, column=0, sticky="nsew")

        footer = ttk.Label(
            self,
            text="Open Source Software by Jeff Pittman (youtube.com/@JeffPittman)",
            style="Hyperlink.TLabel",
            cursor="hand2",
            anchor="center",
        )
        footer.grid(row=6, column=0, sticky="ew", pady=(8, 0))
        footer.configure(font=(None, 10, "underline"))
        footer.bind("<Button-1>", lambda _event: webbrowser.open("https://www.youtube.com/@JeffPittman"))

    def _choose_audio(self) -> None:
        path = filedialog.askopenfilename(
            title="Select audio file",
            filetypes=[("Audio files", "*.wav *.mp3 *.aiff *.aif"), ("All files", "*.*")],
        )
        if path:
            self.audio_var.set(path)
            if not self.output_var.get():
                self.output_var.set(str(Path(path).with_suffix(".mp4")))

    def _choose_output(self) -> None:
        path = filedialog.asksaveasfilename(
            title="Save video as",
            defaultextension=".mp4",
            filetypes=[("MP4 video", "*.mp4")],
        )
        if path:
            self.output_var.set(path)

    def _choose_background(self) -> None:
        path = filedialog.askopenfilename(
            title="Select background image",
            filetypes=[("Image files", "*.jpg *.jpeg *.png"), ("All files", "*.*")],
        )
        if path:
            self.bg_image_var.set(path)

    def _on_fast_toggle(self) -> None:
        fast = self.fast_render_var.get()
        if fast:
            self.bg_style_var.set("solid")
            self.bg_style_combo.configure(state="disabled")
            self.gradient_entry.configure(state="disabled")
        else:
            self.bg_style_combo.configure(state="readonly")
            self.gradient_entry.configure(state="normal")

    def _cancel_render(self) -> None:
        if self.cancel_btn.instate(("disabled",)):
            return
        self._cancel_requested = True
        self.status_var.set("Cancelling render…")
        self.cancel_btn.configure(state="disabled")
        self.output_queue.put(("LOG", "Cancel requested."))
        if self.current_process:
            try:
                self.current_process.terminate()
            except Exception:  # noqa: BLE001
                pass

    def _append_log(self, text: str) -> None:
        self.log_widget.configure(state="normal")
        self.log_widget.insert("end", text + "\n")
        self.log_widget.see("end")
        self.log_widget.configure(state="disabled")

    def _drain_output(self) -> None:
        while True:
            try:
                line = self.output_queue.get_nowait()
            except queue.Empty:
                break
            if isinstance(line, tuple):
                kind, payload = line
                if kind == "LOG":
                    self._append_log(payload)
                elif kind == "PROGRESS":
                    progress, eta_text = payload
                    self.progress_var.set(progress)
                    self._last_progress = progress
                    percent = max(0, min(int(progress * 100), 100))
                    if progress >= 1.0:
                        self.status_var.set("Finalizing video…")
                    elif eta_text:
                        self.status_var.set(f"Rendering… {percent}% (ETA {eta_text})")
                    else:
                        self.status_var.set(f"Rendering… {percent}%")
                elif kind == "STATUS":
                    self.status_var.set(payload)
                elif kind == "MODE":
                    self._set_progress_mode(payload)
            else:
                self._append_log(line)
        self.after(100, self._drain_output)

    def _start_render(self) -> None:
        if self.render_thread and self.render_thread.is_alive():
            messagebox.showinfo("Trace Video", "Render already in progress.")
            return

        audio_path = self.audio_var.get().strip()
        if not audio_path:
            messagebox.showerror("Trace Video", "Please choose an audio file.")
            return

        try:
            font_size = int(self.font_size_var.get())
        except ValueError:
            messagebox.showerror("Trace Video", "Font size must be a number.")
            return

        try:
            filter_threads = int(self.filter_threads_var.get()) if self.filter_threads_var.get() else 0
        except ValueError:
            messagebox.showerror("Trace Video", "Filter thread count must be an integer.")
            return

        audio_duration = self._probe_duration(Path(audio_path))
        if audio_duration:
            self._set_progress_mode("determinate")
            self.progress_var.set(0.0)
            self.status_var.set("Rendering… 0%")
            self._last_progress = 0.0
        else:
            self._set_progress_mode("indeterminate")
            self.status_var.set("Rendering… (estimating)")
            self._last_progress = 0.0
        frame_target = None
        if audio_duration:
            frame_target = max(int(round(audio_duration * 30)), 1)

        args = SimpleNamespace(
            input_audio=Path(audio_path),
            output=Path(self.output_var.get().strip()) if self.output_var.get().strip() else None,
            background_image=Path(self.bg_image_var.get().strip()) if self.bg_image_var.get().strip() else None,
            background_color=self.bg_color_var.get().strip() or gtv.DEFAULT_BACKGROUND_COLOR,
            background_style=self.bg_style_var.get(),
            gradient_colors=self.gradient_var.get().strip() or gtv.DEFAULT_GRADIENT_COLORS,
            mode=self.mode_var.get(),
            wave_color=self.wave_color_var.get().strip() or gtv.DEFAULT_WAVE_COLOR,
            width=1920,
            height=1080,
            wave_height=0.35,
            bottom_margin=160,
            title=self.title_var.get().strip() or None,
            title_color="#ffffff",
            font=self.font_var.get().strip() or "Helvetica",
            font_size=font_size,
            video_bitrate=gtv.DEFAULT_VIDEO_BITRATE,
            audio_bitrate=gtv.DEFAULT_AUDIO_BITRATE,
            encoder_preset=self.preset_var.get(),
            filter_complex_threads=filter_threads,
            ffmpeg_path="ffmpeg",
            dry_run=self.dry_run_var.get(),
            fast_render=self.fast_render_var.get(),
        )

        self.render_btn.configure(state="disabled")
        self.output_queue.put(("LOG", "Starting render…"))
        self._cancel_requested = False
        self.cancel_btn.configure(state="normal")
        self.current_process = None

        self.render_thread = threading.Thread(
            target=self._run_render,
            args=(args, frame_target),
            daemon=True,
        )
        self.render_thread.start()

    def _run_render(self, args: SimpleNamespace, frame_target: Optional[int]) -> None:
        try:
            cmd = gtv.build_ffmpeg_command(args)
        except Exception as exc:  # noqa: BLE001
            self.output_queue.put(("LOG", f"Error: {exc}"))
            self.output_queue.put(("STATUS", "Render failed to start."))
            self._finish_render(success=False, determinate=bool(frame_target), cancelled=False)
            return

        command_str = " ".join(shlex.quote(part) for part in cmd)
        self.output_queue.put(("LOG", command_str))

        if args.dry_run:
            self.output_queue.put(("STATUS", "Dry run complete."))
            self._finish_render(success=True, determinate=bool(frame_target), cancelled=False)
            return

        ffmpeg_cmd = self._with_progress(cmd) if frame_target else cmd
        start_time = time.monotonic()
        last_eta_emit = start_time

        process = subprocess.Popen(
            ffmpeg_cmd,
            stdout=subprocess.PIPE,
            stderr=subprocess.STDOUT,
            text=True,
        )
        self.current_process = process

        if self._cancel_requested:
            try:
                process.terminate()
            except Exception:  # noqa: BLE001
                pass

        assert process.stdout is not None
        for line in process.stdout:
            stripped = line.strip()
            if not stripped:
                continue
            if frame_target and "=" in stripped:
                key, value = stripped.split("=", 1)
                if key == "frame":
                    try:
                        frame_count = max(int(value), 0)
                    except ValueError:
                        continue
                    progress = min(frame_count / frame_target, 1.0)
                    eta_text: Optional[str] = None
                    now = time.monotonic()
                    if progress >= 1.0:
                        eta_text = self._format_eta(0.0)
                        self.output_queue.put(("STATUS", "Finalizing video…"))
                    elif now - last_eta_emit >= 15 and frame_count > 0:
                        elapsed = max(now - start_time, 1e-6)
                        frames_per_second = frame_count / elapsed
                        if frames_per_second > 0:
                            remaining_frames = max(frame_target - frame_count, 0)
                            remaining_seconds = remaining_frames / frames_per_second if remaining_frames else 0.0
                            eta_text = self._format_eta(remaining_seconds)
                        last_eta_emit = now
                    self.output_queue.put(("PROGRESS", (progress, eta_text)))
                    continue
                if key == "progress" and value == "end":
                    self.output_queue.put(("PROGRESS", (1.0, self._format_eta(0.0))))
                    continue
            self.output_queue.put(("LOG", stripped))

        cancelled = False
        try:
            return_code = process.wait()
            cancelled = self._cancel_requested
            if return_code == 0 and not cancelled:
                self.output_queue.put(("LOG", "Render finished successfully."))
                self.output_queue.put(("STATUS", "Render completed."))
                self._finish_render(success=True, determinate=bool(frame_target), cancelled=False)
            elif cancelled:
                self.output_queue.put(("LOG", "Render cancelled by user."))
                self.output_queue.put(("STATUS", "Render cancelled."))
                self._finish_render(success=False, determinate=bool(frame_target), cancelled=True)
            else:
                self.output_queue.put(("LOG", f"ffmpeg exited with code {return_code}."))
                self.output_queue.put(("STATUS", f"Render failed (code {return_code})."))
                self._finish_render(success=False, determinate=bool(frame_target), cancelled=False)
        finally:
            self.current_process = None
            self._cancel_requested = False

    def _finish_render(self, success: bool, determinate: bool, cancelled: bool) -> None:
        def update() -> None:
            self.render_btn.configure(state="normal")
            self.cancel_btn.configure(state="disabled")
            if determinate:
                self.progress_bar.stop()
                self.progress_bar.configure(mode="determinate")
                if success:
                    self.progress_var.set(1.0)
                else:
                    self.progress_var.set(self._last_progress)
            else:
                self.progress_bar.stop()
                self.progress_bar.configure(mode="determinate")
                if success:
                    self.progress_var.set(1.0)
            if cancelled:
                self.status_var.set("Render cancelled.")
            elif success:
                self.status_var.set("Render completed.")
            else:
                self.status_var.set("Render failed.")

        self.after(0, update)

    def _set_progress_mode(self, mode: str) -> None:
        if mode == "indeterminate":
            self.progress_bar.configure(mode="indeterminate")
            self.progress_bar.start(10)
        else:
            self.progress_bar.stop()
            self.progress_bar.configure(mode="determinate")

    def _with_progress(self, cmd: list[str]) -> list[str]:
        new_cmd = cmd.copy()
        output_index = len(new_cmd) - 1
        new_cmd[output_index:output_index] = ["-progress", "pipe:1", "-nostats"]
        return new_cmd

    def _probe_duration(self, path: Path) -> Optional[float]:
        try:
            result = subprocess.run(
                [
                    "ffprobe",
                    "-v",
                    "error",
                    "-show_entries",
                    "format=duration",
                    "-of",
                    "default=noprint_wrappers=1:nokey=1",
                    str(path),
                ],
                capture_output=True,
                text=True,
                check=True,
            )
            duration_str = result.stdout.strip()
            if duration_str:
                return max(float(duration_str), 0.0)
        except (subprocess.SubprocessError, ValueError):
            pass
        return None

    def _format_eta(self, seconds: float) -> Optional[str]:
        remaining = max(seconds, 0.0)
        if remaining == float("inf"):
            return None
        if remaining >= 3600:
            hours = int(remaining // 3600)
            minutes = int((remaining % 3600) // 60)
            return f"{hours}h {minutes}m"
        if remaining >= 60:
            minutes = int(remaining // 60)
            seconds = int(remaining % 60)
            return f"{minutes}m {seconds}s"
        return f"{int(remaining)}s"


def main() -> None:
    root = tk.Tk()
    root.title("Trace Wave Video Generator")
    root.geometry("720x600")
    TraceVideoApp(root)
    root.mainloop()


if __name__ == "__main__":
    main()