#!/usr/bin/env python3 """Graphical User Interface for HPC-Quantize V2 (Factored Q2_K Engine). Launch:: python3 gui.py Provides: * Source GGUF, iMatrix (.gguf / .dat), and Output (.gguf / .npz) file pickers * Interactive GGUF tensor table with live per-tensor status & RMSE updates * Real-time Telemetry Dashboard showing active tensor, current pipeline phase, dual progress bars (overall model + intra-tensor superblocks), throughput (blk/s), ETA, running RMSE, RMSE/sigma, Boltzmann temperature, and floor hits * Full HPCConfig controls (Geometry-Preserving Analogs, Coupled Bilinear Q2_K, Sieve + 36-state Viterbi, Spectral/Vesica/DC shaping, Candidate lattice size) * Timestamped verbose log stream with selectable verbosity level """ from __future__ import annotations import argparse import datetime as dt import queue import shlex import threading import time import tkinter as tk from pathlib import Path from tkinter import filedialog, messagebox, ttk import numpy as np _HERE = Path(__file__).resolve().parent if not (_HERE / "hpc_quantize_v2").is_dir() and (_HERE / "__init__.py").is_file(): import importlib.util import sys if "hpc_quantize_v2" not in sys.modules: _spec = importlib.util.spec_from_file_location( "hpc_quantize_v2", _HERE / "__init__.py", submodule_search_locations=[str(_HERE)], ) if _spec is not None and _spec.loader is not None: _mod = importlib.util.module_from_spec(_spec) sys.modules["hpc_quantize_v2"] = _mod _spec.loader.exec_module(_mod) from hpc_quantize_v2 import ( GGUFReader, HPCConfig, Q1_0_BLOCK_BYTES, Q2KQuantizer, Q3_K_BLOCK_BYTES, QK1_0, QK3_K, QK4_0, QK_K, cuda_device_name, cuda_is_available, dequantize_q1_0_bytes, dequantize_q3_k_bytes, dequantize_q4_0_bytes, expand_importance, factor_only_solve, lookup_imatrix, quantize_tensor_q1_0, quantize_tensor_q3_k, quantize_tensor_q4_0, read_imatrix, ) from hpc_quantize_v2.gguf import ( is_q1_0_eligible, is_q2k_eligible, is_q3_k_eligible, is_q4_0_eligible, target_quant_type, tensor_type_key, write_quantized_gguf, ) from hpc_quantize_v2.reference import reference_solve, synthetic_cases DEFAULT_CUSTOM_RECIPE: dict[str, str] = { # Q4_0: quality-critical embeddings, attention, and SSM projections "token_embd": "Q4_0", "output": "Q4_0", "attn_gate": "Q4_0", "ssm_out": "Q4_0", "attn_q": "Q4_0", "attn_k": "Q4_0", "attn_v": "Q4_0", # Q1_0: the one tensor type that tolerates it well "attn_output": "Q1_0", # Q2_K: the remaining compute-heavy tensors "attn_qkv": "Q2_K", "ffn_up": "Q2_K", "ffn_gate": "Q2_K", "ffn_down": "Q2_K", "ssm_alpha": "Q2_K", "ssm_beta": "Q2_K", "nextn.eh_proj": "Q2_K", } def _ts() -> str: return dt.datetime.now().strftime("%H:%M:%S.%f")[:-3] def _fmt_eta(seconds: float) -> str: if not np.isfinite(seconds) or seconds <= 0.0: return "0s" s = int(round(seconds)) if s < 60: return f"{seconds:.1f}s" m, sec = divmod(s, 60) if m < 60: return f"{m}m{sec:02d}s" h, m = divmod(m, 60) return f"{h}h{m:02d}m{sec:02d}s" class HPCQuantizeGUI: def __init__(self, root: tk.Tk) -> None: self.root = root self.root.title("HPC-Quantize V2 — Factored Q1_0 / Q2_K / Q3_K / Q4_0 & Geometry-Preserving Analog Engine") self.root.geometry("1260x900") self.root.minsize(1040, 740) self._log_queue: queue.Queue[tuple[str, object]] = queue.Queue() self._stop_event = threading.Event() self._worker: threading.Thread | None = None self._reader: GGUFReader | None = None self._imatrix: dict[str, np.ndarray] | None = None self._start_time: float | None = None self._last_logged_phase: str = "" self._init_vars() self._apply_theme() self._build_layout() if self.quant_format_var.get() == "custom": self._on_format_changed() self._auto_populate_defaults() self._update_cli_preview() self.root.after(80, self._poll_queue) def _init_vars(self) -> None: self.gguf_var = tk.StringVar() self.imatrix_var = tk.StringVar() self.output_var = tk.StringVar() self.mode_var = tk.StringVar(value="all") self.filter_var = tk.StringVar() self.log_level_var = tk.StringVar(value="ultra") self.optimizer_var = tk.StringVar(value="hpc") self.cand_d_var = tk.IntVar(value=6) self.cand_m_var = tk.IntVar(value=6) self.starts_var = tk.IntVar(value=2) self.iters_var = tk.IntVar(value=8) self.ulp_var = tk.IntVar(value=4) self.limit_var = tk.IntVar(value=0) self.workers_var = tk.IntVar(value=0) self.device_var = tk.StringVar(value="auto") self.use_analogs_var = tk.BooleanVar(value=True) self.coupled_q2k_var = tk.BooleanVar(value=True) self.joint_subscales_var = tk.BooleanVar(value=True) self.use_spectral_var = tk.BooleanVar(value=True) self.subblock_sieve_var = tk.BooleanVar(value=True) self.factor_floor_var = tk.BooleanVar(value=True) self.cand_floor_var = tk.BooleanVar(value=True) self.dc_carry_var = tk.BooleanVar(value=True) self.q4_fallback_var = tk.BooleanVar(value=True) self.embd_q4_var = tk.BooleanVar(value=True) self.promote_attn_q4_var = tk.BooleanVar(value=False) self.quant_format_var = tk.StringVar(value="custom") self.plan_filter_var = tk.StringVar(value="ALL") self.custom_type_vars: dict[str, tk.StringVar] = {} self.dc_lambda_var = tk.DoubleVar(value=1.0) self.vw_lambda_var = tk.DoubleVar(value=1.0) self.dc_decay_var = tk.DoubleVar(value=0.85) self.sse_budget_var = tk.DoubleVar(value=0.0005) self.thermal_var = tk.IntVar(value=0) self.thermal_tilt_var = tk.DoubleVar(value=1.0) self.pin_dc_var = tk.BooleanVar(value=False) self.cli_preview_var = tk.StringVar() self.status_var = tk.StringVar(value="Ready.") self.model_progress_var = tk.DoubleVar(value=0.0) self.tensor_progress_var = tk.DoubleVar(value=0.0) # Live telemetry variables self.tele_tensor_var = tk.StringVar(value="Idle") self.tele_phase_var = tk.StringVar(value="Waiting to start") self.tele_blocks_var = tk.StringVar(value="0 / 0") self.tele_speed_var = tk.StringVar(value="0.0 blk/s (ETA --)") self.tele_rmse_var = tk.StringVar(value="--") self.tele_temp_var = tk.StringVar(value="--") self.tele_floors_var = tk.StringVar(value="cand=0, factor=0") self.tele_elapsed_var = tk.StringVar(value="0.0s") for v in ( self.gguf_var, self.imatrix_var, self.output_var, self.mode_var, self.optimizer_var, self.cand_d_var, self.cand_m_var, self.limit_var, self.workers_var, self.device_var, self.use_spectral_var, self.q4_fallback_var, self.embd_q4_var, self.promote_attn_q4_var, self.quant_format_var, self.thermal_var, ): v.trace_add("write", lambda *_: self._update_cli_preview()) def _apply_theme(self) -> None: style = ttk.Style(self.root) if "clam" in style.theme_names(): style.theme_use("clam") bg = "#141820" panel = "#1c222d" fg = "#e6edf3" accent = "#3b82f6" self.root.configure(bg=bg) style.configure(".", background=bg, foreground=fg, fieldbackground=panel) style.configure("TFrame", background=bg) style.configure("TLabelframe", background=bg, foreground=fg) style.configure("TLabelframe.Label", background=bg, foreground="#93c5fd", font=("Sans", 10, "bold")) style.configure("TLabel", background=bg, foreground=fg) style.configure("TeleKey.TLabel", background=bg, foreground="#9ca3af", font=("Sans", 9)) style.configure("TeleVal.TLabel", background=bg, foreground="#60a5fa", font=("Monospace", 9, "bold")) style.configure("TCheckbutton", background=bg, foreground=fg) style.configure("TRadiobutton", background=bg, foreground=fg) style.configure("TButton", background=panel, foreground=fg, padding=(10, 5)) style.map("TButton", background=[("active", accent)]) style.configure("Accent.TButton", background=accent, foreground="#ffffff", font=("Sans", 10, "bold")) style.configure( "Treeview", background=panel, fieldbackground=panel, foreground=fg, rowheight=23, ) style.configure("Treeview.Heading", background="#262f3f", foreground=fg, font=("Sans", 9, "bold")) style.map("Treeview", background=[("selected", accent)]) def _build_layout(self) -> None: outer = ttk.Frame(self.root) outer.pack(fill=tk.BOTH, expand=True) outer.rowconfigure(0, weight=1) outer.columnconfigure(0, weight=1) self._canvas = tk.Canvas(outer, bg="#141820", highlightthickness=0) v_scroll = ttk.Scrollbar(outer, orient=tk.VERTICAL, command=self._canvas.yview) h_scroll = ttk.Scrollbar(outer, orient=tk.HORIZONTAL, command=self._canvas.xview) self._canvas.configure(yscrollcommand=v_scroll.set, xscrollcommand=h_scroll.set) self._canvas.grid(row=0, column=0, sticky=tk.NSEW) v_scroll.grid(row=0, column=1, sticky=tk.NS) h_scroll.grid(row=1, column=0, sticky=tk.EW) main = ttk.Frame(self._canvas, padding=10) self._canvas_win = self._canvas.create_window((0, 0), window=main, anchor=tk.NW) def _on_frame_configure(_event=None) -> None: self._canvas.configure(scrollregion=self._canvas.bbox("all")) def _on_canvas_configure(event) -> None: req_w = max(event.width, main.winfo_reqwidth()) req_h = max(event.height, main.winfo_reqheight()) self._canvas.itemconfigure(self._canvas_win, width=req_w, height=req_h) main.bind("", _on_frame_configure) self._canvas.bind("", _on_canvas_configure) def _on_mousewheel(event) -> None: w = event.widget if isinstance(w, (ttk.Treeview, tk.Text)): return if getattr(event, "num", None) == 4: self._canvas.yview_scroll(-1, "units") elif getattr(event, "num", None) == 5: self._canvas.yview_scroll(1, "units") elif getattr(event, "delta", 0): self._canvas.yview_scroll(int(-1 * (event.delta / 120)), "units") self.root.bind_all("", _on_mousewheel) self.root.bind_all("", _on_mousewheel) self.root.bind_all("", _on_mousewheel) # 1. File Selection Frame files_frame = ttk.LabelFrame(main, text="1. GGUF & iMatrix Paths", padding=8) files_frame.pack(fill=tk.X, pady=(0, 6)) files_frame.columnconfigure(1, weight=1) ttk.Label(files_frame, text="Source GGUF:").grid(row=0, column=0, sticky=tk.W, padx=(0, 8), pady=2) ttk.Entry(files_frame, textvariable=self.gguf_var).grid(row=0, column=1, sticky=tk.EW, pady=2) ttk.Button(files_frame, text="Browse...", command=self._browse_gguf).grid(row=0, column=2, padx=4, pady=2) ttk.Button(files_frame, text="Inspect GGUF", command=self.inspect_files).grid(row=0, column=3, pady=2) ttk.Label(files_frame, text="iMatrix (.gguf/.dat):").grid(row=1, column=0, sticky=tk.W, padx=(0, 8), pady=2) ttk.Entry(files_frame, textvariable=self.imatrix_var).grid(row=1, column=1, sticky=tk.EW, pady=2) ttk.Button(files_frame, text="Browse...", command=self._browse_imatrix).grid(row=1, column=2, padx=4, pady=2) ttk.Button(files_frame, text="Clear", command=lambda: self.imatrix_var.set("")).grid(row=1, column=3, pady=2) ttk.Label(files_frame, text="Output (.gguf/.npz):").grid(row=2, column=0, sticky=tk.W, padx=(0, 8), pady=2) ttk.Entry(files_frame, textvariable=self.output_var).grid(row=2, column=1, sticky=tk.EW, pady=2) ttk.Button(files_frame, text="Save As...", command=self._browse_output).grid(row=2, column=2, padx=4, pady=2) # 2. Middle Split: Left = Settings, Right = Tensor Browser mid = ttk.PanedWindow(main, orient=tk.HORIZONTAL) mid.pack(fill=tk.BOTH, expand=True, pady=(0, 6)) left_pane = ttk.Frame(mid) right_pane = ttk.Frame(mid) mid.add(left_pane, weight=2) mid.add(right_pane, weight=3) # Left: Engine & Factorization Options opt_frame = ttk.LabelFrame(left_pane, text="2. Factored Q1_0 / Q2_K / Q3_K / Q4_0 & Analog Controls", padding=8) opt_frame.pack(fill=tk.BOTH, expand=True, padx=(0, 6)) mode_row = ttk.Frame(opt_frame) mode_row.pack(fill=tk.X, pady=(0, 4)) ttk.Label(mode_row, text="Target Scope:").pack(side=tk.LEFT, padx=(0, 8)) ttk.Radiobutton( mode_row, text="All Eligible Tensors", variable=self.mode_var, value="all", command=self._on_scope_changed, ).pack(side=tk.LEFT, padx=4) ttk.Radiobutton( mode_row, text="Selected Tensor(s)", variable=self.mode_var, value="selected", command=self._on_scope_changed, ).pack(side=tk.LEFT, padx=4) fmt_row = ttk.Frame(opt_frame) fmt_row.pack(fill=tk.X, pady=(0, 4)) ttk.Label(fmt_row, text="Quant Format:").pack(side=tk.LEFT, padx=(0, 8)) for lbl, val in ( ("Mixed Q2_K + Q4_0", "mixed"), ("Mixed Q1_0 + Q4_0", "mixed_q1"), ("Pure Q1_0 (1.125 bpw)", "q1_0"), ("Pure Q2_K (2.625 bpw)", "q2_k"), ("Pure Q3_K (3.438 bpw)", "q3_k"), ("Pure Q4_0 (4.500 bpw)", "q4_0"), ): ttk.Radiobutton( fmt_row, text=lbl, variable=self.quant_format_var, value=val, command=self._on_format_changed, ).pack(side=tk.LEFT, padx=3) custom_row = ttk.Frame(opt_frame) custom_row.pack(fill=tk.X, pady=(0, 4)) ttk.Radiobutton( custom_row, text="Custom (per tensor type)", variable=self.quant_format_var, value="custom", command=self._on_format_changed, ).pack(side=tk.LEFT, padx=4) ttk.Label( custom_row, text="Override the quant type per tensor category (Auto = mixed plan).", ).pack(side=tk.LEFT, padx=6) self.custom_frame = ttk.LabelFrame(opt_frame, text="Custom Tensor-Type Targets", padding=6) self.custom_body = ttk.Frame(self.custom_frame) self.custom_body.pack(fill=tk.X) self.custom_hint_var = tk.StringVar(value="Load a source GGUF to list tensor types.") ttk.Label( self.custom_frame, textvariable=self.custom_hint_var, style="TeleKey.TLabel" ).pack(anchor=tk.W, pady=(4, 0)) opt_row = ttk.Frame(opt_frame) opt_row.pack(fill=tk.X, pady=(0, 4)) self._custom_anchor = opt_row ttk.Label(opt_row, text="Global Optimizer:").pack(side=tk.LEFT, padx=(0, 8)) ttk.Radiobutton( opt_row, text="HPC (Sieve + 36-State Viterbi)", variable=self.optimizer_var, value="hpc" ).pack(side=tk.LEFT, padx=4) ttk.Radiobutton( opt_row, text="MSE (Local Greedy)", variable=self.optimizer_var, value="mse" ).pack(side=tk.LEFT, padx=4) checks = ttk.Frame(opt_frame) checks.pack(fill=tk.X, pady=2) ttk.Checkbutton( checks, text="Geometry-Preserving Weight Analogs (Iso-Moment + Vesica Gram + Subscale Lattice)", variable=self.use_analogs_var, ).pack(anchor=tk.W, pady=1) ttk.Checkbutton( checks, text="Coupled Orthogonal-Basis Bilinear Q2_K Factorization", variable=self.coupled_q2k_var, ).pack(anchor=tk.W, pady=1) ttk.Checkbutton( checks, text="Exact Joint (Ls, Lm) 16x16 Sub-block Scan", variable=self.joint_subscales_var, ).pack(anchor=tk.W, pady=1) ttk.Checkbutton( checks, text="Spectral Fold / Vesica / DC Objective Penalty", variable=self.use_spectral_var, ).pack(anchor=tk.W, pady=1) ttk.Checkbutton( checks, text="Sub-block 6-State Sieve & Rolling DC Residual Carry", variable=self.subblock_sieve_var, ).pack(anchor=tk.W, pady=1) ttk.Checkbutton( checks, text="Local-Safety Candidate & Factored Floors (Monotone Bound)", variable=self.factor_floor_var, ).pack(anchor=tk.W, pady=1) ttk.Checkbutton( checks, text="Q4_0 Fallback for 32-Aligned Non-Q2_K Tensors & Recurrent Gates (ssm_alpha/beta, ffn_down_exps, hc_*)", variable=self.q4_fallback_var, command=self._populate_tensor_tree, ).pack(anchor=tk.W, pady=1) ttk.Checkbutton( checks, text="Quantize token_embd.weight & output.weight to Q4_0 (Saves 1.3–3.7 GB)", variable=self.embd_q4_var, command=self._populate_tensor_tree, ).pack(anchor=tk.W, pady=1) ttk.Checkbutton( checks, text="Promote Attention Q/K/V/O Projections to Q4_0 (Mixed Q2_K + Q4_0)", variable=self.promote_attn_q4_var, command=self._populate_tensor_tree, ).pack(anchor=tk.W, pady=1) grid_f = ttk.Frame(opt_frame) grid_f.pack(fill=tk.X, pady=4) for col in range(4): grid_f.columnconfigure(col, weight=1) self._spin_pair(grid_f, 0, 0, "Cand d (1-24):", self.cand_d_var, 1, 24) self._spin_pair(grid_f, 0, 2, "Cand dmin (1-24):", self.cand_m_var, 1, 24) self._spin_pair(grid_f, 1, 0, "Solver Starts:", self.starts_var, 1, 8) self._spin_pair(grid_f, 1, 2, "Solver Iters:", self.iters_var, 1, 32) self._spin_pair(grid_f, 2, 0, "FP16 ULP Radius:", self.ulp_var, 0, 16) self._spin_pair(grid_f, 2, 2, "Tensor Limit (0=all):", self.limit_var, 0, 9999) self._spin_pair(grid_f, 3, 0, "Workers (0=Auto CPU):", self.workers_var, 0, 256) ttk.Label(grid_f, text="Compute Device:").grid(row=3, column=2, sticky=tk.W, padx=4, pady=2) dev_cb = ttk.Combobox( grid_f, textvariable=self.device_var, values=("auto", "cuda", "cpu"), state="readonly", width=8 ) dev_cb.grid(row=3, column=3, sticky=tk.EW, padx=4, pady=2) spec_f = ttk.LabelFrame(opt_frame, text="Spectral & DC Parameters", padding=6) spec_f.pack(fill=tk.X, pady=2) for col in range(4): spec_f.columnconfigure(col, weight=1) ttk.Label(spec_f, text="DC Lambda:").grid(row=0, column=0, sticky=tk.W, padx=4, pady=2) ttk.Entry(spec_f, textvariable=self.dc_lambda_var, width=8).grid(row=0, column=1, sticky=tk.EW, padx=4, pady=2) ttk.Label(spec_f, text="Vesica Lambda:").grid(row=0, column=2, sticky=tk.W, padx=4, pady=2) ttk.Entry(spec_f, textvariable=self.vw_lambda_var, width=8).grid(row=0, column=3, sticky=tk.EW, padx=4, pady=2) ttk.Label(spec_f, text="DC Decay:").grid(row=1, column=0, sticky=tk.W, padx=4, pady=2) ttk.Entry(spec_f, textvariable=self.dc_decay_var, width=8).grid(row=1, column=1, sticky=tk.EW, padx=4, pady=2) ttk.Label(spec_f, text="SSE Budget:").grid(row=1, column=2, sticky=tk.W, padx=4, pady=2) ttk.Entry(spec_f, textvariable=self.sse_budget_var, width=8).grid(row=1, column=3, sticky=tk.EW, padx=4, pady=2) ttk.Label(spec_f, text="Thermal Passes (0=off):").grid(row=2, column=0, sticky=tk.W, padx=4, pady=2) ttk.Spinbox( spec_f, from_=0, to=16, increment=1, textvariable=self.thermal_var, width=8 ).grid(row=2, column=1, sticky=tk.EW, padx=4, pady=2) ttk.Label(spec_f, text="Zero-Residual Tilt (1=off):").grid(row=2, column=2, sticky=tk.W, padx=4, pady=2) ttk.Entry(spec_f, textvariable=self.thermal_tilt_var, width=8).grid(row=2, column=3, sticky=tk.EW, padx=4, pady=2) ttk.Checkbutton( spec_f, text="Pin DC residual to 0 (Q2_K free lunch)", variable=self.pin_dc_var, ).grid(row=3, column=0, columnspan=2, sticky=tk.W, padx=4, pady=1) ttk.Label( spec_f, text="(λ multiplier ≥1 during thermal anneal; ~1e2–1e3 ≈ zero-coherent-residual '0-error' anneal)", foreground="#93c5fd", ).grid(row=4, column=0, columnspan=4, sticky=tk.W, padx=4, pady=(0, 2)) # Right: GGUF Tensor Browser browser_frame = ttk.LabelFrame(right_pane, text="3. GGUF Tensor Inspector & Live Status", padding=8) browser_frame.pack(fill=tk.BOTH, expand=True) search_row = ttk.Frame(browser_frame) search_row.pack(fill=tk.X, pady=(0, 6)) ttk.Label(search_row, text="Filter:").pack(side=tk.LEFT, padx=(0, 4)) filter_entry = ttk.Entry(search_row, textvariable=self.filter_var) filter_entry.pack(side=tk.LEFT, fill=tk.X, expand=True) filter_entry.bind("", lambda _e: self._populate_tensor_tree()) for label, val in (("All", "ALL"), ("Q1_0", "Q1_0"), ("Q2_K", "Q2_K"), ("Q3_K", "Q3_K"), ("Q4_0", "Q4_0"), ("Copy", "COPY")): ttk.Radiobutton( search_row, text=label, variable=self.plan_filter_var, value=val, command=self._populate_tensor_tree, ).pack(side=tk.LEFT, padx=3) ttk.Button( search_row, text="Select Quantized", command=self._select_all_eligible ).pack(side=tk.LEFT, padx=4) cols = ("name", "type", "dims", "row", "q2k", "imat", "state") self.tree = ttk.Treeview(browser_frame, columns=cols, show="headings", selectmode="extended") self.tree.heading("name", text="Tensor Name") self.tree.heading("type", text="Source") self.tree.heading("dims", text="Dims") self.tree.heading("row", text="Row Width") self.tree.heading("q2k", text="Target") self.tree.heading("imat", text="iMatrix") self.tree.heading("state", text="Live Status / RMSE") self.tree.tag_configure("Q1_0", foreground="#fbbf24") self.tree.tag_configure("Q2_K", foreground="#93c5fd") self.tree.tag_configure("Q3_K", foreground="#c4b5fd") self.tree.tag_configure("Q4_0", foreground="#34d399") self.tree.tag_configure("Copy", foreground="#9ca3af") self.tree.column("name", width=220, anchor=tk.W) self.tree.column("type", width=52, anchor=tk.CENTER) self.tree.column("dims", width=105, anchor=tk.CENTER) self.tree.column("row", width=70, anchor=tk.E) self.tree.column("q2k", width=48, anchor=tk.CENTER) self.tree.column("imat", width=56, anchor=tk.CENTER) self.tree.column("state", width=165, anchor=tk.W) self.tree.bind("<>", self._on_tree_selection) tree_scroll = ttk.Scrollbar(browser_frame, orient=tk.VERTICAL, command=self.tree.yview) self.tree.configure(yscrollcommand=tree_scroll.set) self.tree.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) tree_scroll.pack(side=tk.RIGHT, fill=tk.Y) # 4. Equivalent CLI Command Preview cli_frame = ttk.LabelFrame(main, text="4. Equivalent Terminal Command", padding=6) cli_frame.pack(fill=tk.X, pady=(0, 6)) ttk.Entry(cli_frame, textvariable=self.cli_preview_var, state="readonly").pack( side=tk.LEFT, fill=tk.X, expand=True, padx=(0, 6) ) ttk.Button(cli_frame, text="Copy Command", command=self._copy_cli).pack(side=tk.RIGHT) # 5. Live Telemetry Dashboard + Execution + Verbose Console bottom = ttk.LabelFrame(main, text="5. Live Telemetry Dashboard & Verbose Execution Stream", padding=8) bottom.pack(fill=tk.BOTH, expand=True) tele = ttk.Frame(bottom) tele.pack(fill=tk.X, pady=(0, 6)) for col in (1, 3, 5, 7): tele.columnconfigure(col, weight=1) ttk.Label(tele, text="Active Tensor:", style="TeleKey.TLabel").grid(row=0, column=0, sticky=tk.W, padx=4, pady=1) ttk.Label(tele, textvariable=self.tele_tensor_var, style="TeleVal.TLabel").grid( row=0, column=1, columnspan=3, sticky=tk.W, padx=4, pady=1 ) ttk.Label(tele, text="Current Stage:", style="TeleKey.TLabel").grid(row=0, column=4, sticky=tk.W, padx=4, pady=1) ttk.Label(tele, textvariable=self.tele_phase_var, style="TeleVal.TLabel").grid( row=0, column=5, columnspan=3, sticky=tk.W, padx=4, pady=1 ) ttk.Label(tele, text="Superblocks:", style="TeleKey.TLabel").grid(row=1, column=0, sticky=tk.W, padx=4, pady=1) ttk.Label(tele, textvariable=self.tele_blocks_var, style="TeleVal.TLabel").grid(row=1, column=1, sticky=tk.W, padx=4, pady=1) ttk.Label(tele, text="Throughput:", style="TeleKey.TLabel").grid(row=1, column=2, sticky=tk.W, padx=4, pady=1) ttk.Label(tele, textvariable=self.tele_speed_var, style="TeleVal.TLabel").grid(row=1, column=3, sticky=tk.W, padx=4, pady=1) ttk.Label(tele, text="Running RMSE:", style="TeleKey.TLabel").grid(row=1, column=4, sticky=tk.W, padx=4, pady=1) ttk.Label(tele, textvariable=self.tele_rmse_var, style="TeleVal.TLabel").grid(row=1, column=5, sticky=tk.W, padx=4, pady=1) ttk.Label(tele, text="Sieve Temp / Floors:", style="TeleKey.TLabel").grid(row=1, column=6, sticky=tk.W, padx=4, pady=1) ttk.Label(tele, textvariable=self.tele_floors_var, style="TeleVal.TLabel").grid(row=1, column=7, sticky=tk.W, padx=4, pady=1) bars = ttk.Frame(bottom) bars.pack(fill=tk.X, pady=(0, 6)) bars.columnconfigure(1, weight=1) ttk.Label(bars, text="Model Progress:", style="TeleKey.TLabel").grid(row=0, column=0, sticky=tk.W, padx=(0, 6), pady=2) self.model_progress = ttk.Progressbar(bars, variable=self.model_progress_var, maximum=100.0) self.model_progress.grid(row=0, column=1, sticky=tk.EW, pady=2) ttk.Label(bars, text="Tensor Blocks:", style="TeleKey.TLabel").grid(row=1, column=0, sticky=tk.W, padx=(0, 6), pady=2) self.tensor_progress = ttk.Progressbar(bars, variable=self.tensor_progress_var, maximum=100.0) self.tensor_progress.grid(row=1, column=1, sticky=tk.EW, pady=2) btn_row = ttk.Frame(bottom) btn_row.pack(fill=tk.X, pady=(0, 6)) self.run_btn = ttk.Button( btn_row, text="Run Quantization", style="Accent.TButton", command=self.start_quantization ) self.run_btn.pack(side=tk.LEFT, padx=(0, 6)) self.stop_btn = ttk.Button( btn_row, text="Cancel", state=tk.DISABLED, command=self.stop_worker ) self.stop_btn.pack(side=tk.LEFT, padx=(0, 6)) self.bench_btn = ttk.Button( btn_row, text="Run Solver Benchmark", command=self.start_benchmark ) self.bench_btn.pack(side=tk.LEFT, padx=(0, 6)) ttk.Button( btn_row, text="Clear Log", command=lambda: self.log_text.delete("1.0", tk.END) ).pack(side=tk.LEFT, padx=(0, 12)) ttk.Label(btn_row, text="Log Verbosity:").pack(side=tk.LEFT, padx=(0, 4)) ttk.Radiobutton( btn_row, text="Ultra-Verbose (Every Block Step)", variable=self.log_level_var, value="ultra" ).pack(side=tk.LEFT, padx=2) ttk.Radiobutton( btn_row, text="Stage Transitions", variable=self.log_level_var, value="stages" ).pack(side=tk.LEFT, padx=2) ttk.Label(btn_row, textvariable=self.status_var).pack(side=tk.RIGHT, padx=4) log_frame = ttk.Frame(bottom) log_frame.pack(fill=tk.BOTH, expand=True) self.log_text = tk.Text( log_frame, height=11, bg="#0f131a", fg="#d1d5db", insertbackground="#ffffff", font=("Monospace", 9), wrap=tk.NONE, ) log_scroll = ttk.Scrollbar(log_frame, orient=tk.VERTICAL, command=self.log_text.yview) self.log_text.configure(yscrollcommand=log_scroll.set) self.log_text.pack(side=tk.LEFT, fill=tk.BOTH, expand=True) log_scroll.pack(side=tk.RIGHT, fill=tk.Y) def _spin_pair(self, parent, row, col, label, var, low, high) -> None: ttk.Label(parent, text=label).grid(row=row, column=col, sticky=tk.W, padx=4, pady=2) sp = ttk.Spinbox(parent, from_=low, to=high, textvariable=var, width=6) sp.grid(row=row, column=col + 1, sticky=tk.EW, padx=4, pady=2) def _auto_populate_defaults(self) -> None: if cuda_is_available(): self._log(f"CUDA GPU detected: {cuda_device_name()} (device=auto will use GPU)") else: self._log("CUDA GPU not detected (using multi-core CPU backend)") default_gguf = Path("Raon.gguf") default_imat = Path("imatrix.gguf") if default_imat.exists(): self.imatrix_var.set(str(default_imat.resolve())) if default_gguf.exists(): self.gguf_var.set(str(default_gguf.resolve())) self.output_var.set(str(default_gguf.with_name("Raon-Q2K.gguf").resolve())) self.inspect_files() def _default_out_suffix(self) -> str: fmt = self.quant_format_var.get() if fmt == "q1_0": return "-Q1_0.gguf" if fmt == "mixed_q1": return "-Q1_0-Q4_0.gguf" if fmt == "custom": return "-Custom.gguf" if fmt == "q3_k": return "-Q3_K.gguf" if fmt == "q4_0": return "-Q4_0.gguf" return "-Q2K.gguf" def _on_format_changed(self) -> None: cur_out = self.output_var.get().strip() gguf_p = self.gguf_var.get().strip() if gguf_p and ( not cur_out or any( cur_out.endswith(sfx) for sfx in ("-Q2K.gguf", "-Q3_K.gguf", "-Q1_0.gguf", "-Q1_0-Q4_0.gguf", "-Custom.gguf", "-Q4_0.gguf") ) ): p = Path(gguf_p) self.output_var.set(str(p.with_name(f"{p.stem}{self._default_out_suffix()}"))) if self.quant_format_var.get() == "custom": self.custom_frame.pack(fill=tk.X, pady=(0, 4), before=self._custom_anchor) self._populate_custom_tensor_rows() else: self.custom_frame.pack_forget() self._populate_tensor_tree() def _custom_quant_map(self) -> dict[str, str]: return { key: var.get() for key, var in self.custom_type_vars.items() if var.get() not in ("", "Auto") } def _custom_map_for_plan(self) -> dict[str, str] | None: if self.quant_format_var.get() != "custom": return None return self._custom_quant_map() or None def _populate_custom_tensor_rows(self) -> None: if not hasattr(self, "custom_body"): return for child in self.custom_body.winfo_children(): child.destroy() if self._reader is None: self.custom_hint_var.set("Load a source GGUF to list tensor types.") self.custom_type_vars = {} return groups: dict[str, list] = {} for ti in self._reader.tensors: groups.setdefault(tensor_type_key(ti.name), []).append(ti) previous = {key: var.get() for key, var in self.custom_type_vars.items()} self.custom_type_vars = {} rows = 0 for key in sorted(groups): tensors = groups[key] options = ["Auto", "Copy"] if any(is_q1_0_eligible(t.name, t.dims) for t in tensors): options.append("Q1_0") if any(is_q2k_eligible(t.name, t.dims) for t in tensors): options.append("Q2_K") if any(is_q3_k_eligible(t.name, t.dims) for t in tensors): options.append("Q3_K") if any( is_q4_0_eligible(t.name, t.dims, include_embd=True, allow_q2k_tensors=True) for t in tensors ): options.append("Q4_0") if len(options) == 2: continue var = tk.StringVar( value=previous.get(key, DEFAULT_CUSTOM_RECIPE.get(key, "Auto")) ) if var.get() not in options: var.set("Auto") var.trace_add( "write", lambda *_a: self._on_custom_type_changed(), ) self.custom_type_vars[key] = var r, c = divmod(rows, 2) ttk.Label(self.custom_body, text=key, width=18, anchor=tk.W).grid( row=r, column=c * 2, sticky=tk.W, padx=(0, 4), pady=1 ) ttk.Combobox( self.custom_body, textvariable=var, values=options, state="readonly", width=8, ).grid(row=r, column=c * 2 + 1, sticky=tk.W, padx=(0, 12), pady=1) rows += 1 self.custom_hint_var.set( "Quantizable tensor types only; norms/1D tensors and other non-eligible " "weights are copied unchanged." ) self._update_cli_preview() def _on_custom_type_changed(self) -> None: self._populate_tensor_tree() self._update_cli_preview() def _browse_gguf(self) -> None: path = filedialog.askopenfilename( title="Select Source GGUF", filetypes=[("GGUF Model", "*.gguf"), ("All Files", "*.*")], ) if path: self.gguf_var.set(path) p = Path(path) if not self.output_var.get(): self.output_var.set(str(p.with_name(f"{p.stem}{self._default_out_suffix()}"))) self.inspect_files() def _browse_imatrix(self) -> None: path = filedialog.askopenfilename( title="Select iMatrix File", filetypes=[("iMatrix Files", "*.gguf *.dat *.imatrix"), ("All Files", "*.*")], ) if path: self.imatrix_var.set(path) self.inspect_files() def _browse_output(self) -> None: path = filedialog.asksaveasfilename( title="Select Output File", defaultextension=".gguf", filetypes=[("GGUF Model", "*.gguf"), ("NumPy Archive", "*.npz"), ("All Files", "*.*")], ) if path: self.output_var.set(path) def _log(self, msg: str) -> None: self.log_text.insert(tk.END, f"[{_ts()}] {msg}\n") self.log_text.see(tk.END) def build_config(self) -> HPCConfig: cfg = HPCConfig( n_cand_d=int(self.cand_d_var.get()), n_cand_m=int(self.cand_m_var.get()), factor_starts=int(self.starts_var.get()), factor_standalone_iters=int(self.iters_var.get()), ulp_radius=int(self.ulp_var.get()), optimizer=self.optimizer_var.get(), use_weight_analogs=bool(self.use_analogs_var.get()), factor_coupled_q2k=bool(self.coupled_q2k_var.get()), factor_joint_subscales=bool(self.joint_subscales_var.get()), use_spectral=bool(self.use_spectral_var.get()), subblock_sieve=bool(self.subblock_sieve_var.get()), use_factor_floor=bool(self.factor_floor_var.get()), use_candidate_floor=bool(self.cand_floor_var.get()), apply_dc_carry=bool(self.dc_carry_var.get()), dc_lambda=float(self.dc_lambda_var.get()), vw_lambda=float(self.vw_lambda_var.get()), dc_decay=float(self.dc_decay_var.get()), sse_budget=float(self.sse_budget_var.get()), therm_passes=max(0, int(self.thermal_var.get())), therm_lambda_tilt=max(1.0, float(self.thermal_tilt_var.get())), pin_dc=bool(self.pin_dc_var.get()), quant_format=self.quant_format_var.get(), custom_quant_map=self._custom_map_for_plan() or {}, enable_q4_fallback=bool(self.q4_fallback_var.get()) and self.quant_format_var.get() not in ("q2_k", "q3_k"), quantize_embd_q4=bool(self.embd_q4_var.get()), promote_attn_q4=bool(self.promote_attn_q4_var.get()), device=self.device_var.get(), workers=int(self.workers_var.get()), verbose=False, ) return cfg.validate() def build_cli_command(self) -> str: cmd = ["python3", "-m", "hpc_quantize_v2"] gguf = self.gguf_var.get().strip() if gguf: cmd += ["--gguf", gguf] selected = self._selected_tensor_names() if self.mode_var.get() == "selected" and len(selected) == 1: cmd += ["--tensor", next(iter(selected))] else: cmd.append("--all-q2k") imat = self.imatrix_var.get().strip() if imat: cmd += ["--imatrix-gguf", imat] out = self.output_var.get().strip() if out: cmd += ["--output", out] if self.limit_var.get() > 0: cmd += ["--limit", str(self.limit_var.get())] cmd += ["--format", self.quant_format_var.get()] if self.quant_format_var.get() == "custom": custom = self._custom_quant_map() for key in sorted(custom): cmd += ["--tensor-type", f"{key}={custom[key]}"] cmd += ["--device", self.device_var.get()] cmd += ["--workers", str(self.workers_var.get())] cmd += ["--cand-d", str(self.cand_d_var.get()), "--cand-m", str(self.cand_m_var.get())] if self.optimizer_var.get() == "mse": cmd.append("--mse") if not self.use_spectral_var.get(): cmd.append("--no-spectral") if not self.q4_fallback_var.get(): cmd.append("--no-q4-fallback") if not self.embd_q4_var.get(): cmd.append("--keep-embd") if self.promote_attn_q4_var.get(): cmd.append("--promote-attn-q4") if self.thermal_var.get() > 0: cmd += ["--thermal", str(int(self.thermal_var.get()))] if self.thermal_tilt_var.get() > 1.0: cmd += ["--thermal-tilt", str(float(self.thermal_tilt_var.get()))] if self.pin_dc_var.get(): cmd += ["--pin-dc"] cmd.append("--verbose") return " ".join(shlex.quote(part) for part in cmd) def _update_cli_preview(self) -> None: try: self.cli_preview_var.set(self.build_cli_command()) except Exception: pass def _copy_cli(self) -> None: cmd = self.cli_preview_var.get() self.root.clipboard_clear() self.root.clipboard_append(cmd) self.status_var.set("Copied CLI command to clipboard.") def inspect_files(self) -> None: imat_path = self.imatrix_var.get().strip() if imat_path and Path(imat_path).exists(): try: self._imatrix = read_imatrix(imat_path) self._log(f"Loaded iMatrix: {len(self._imatrix)} entries from {imat_path}") except Exception as exc: self._imatrix = None self._log(f"iMatrix load error: {exc}") else: self._imatrix = None gguf_path = self.gguf_var.get().strip() if not gguf_path or not Path(gguf_path).exists(): return try: self._reader = GGUFReader(gguf_path) if self.quant_format_var.get() == "custom": self._populate_custom_tensor_rows() self._populate_tensor_tree() except Exception as exc: self._reader = None self._log(f"GGUF inspect error: {exc}") def _plan_target(self, ti, chosen: set[str], scope_selected: bool): tgt = target_quant_type( ti.name, ti.dims, enable_q4_fallback=bool(self.q4_fallback_var.get()) and self.quant_format_var.get() not in ("q2_k", "q3_k"), quantize_embd_q4=bool(self.embd_q4_var.get()), promote_attn_q4=bool(self.promote_attn_q4_var.get()), quant_format=self.quant_format_var.get(), source_type=ti.type, custom_quant_map=self._custom_map_for_plan(), ) if scope_selected and ti.name not in chosen: return None return tgt def _on_scope_changed(self) -> None: self._refresh_plan_view() self._update_cli_preview() def _on_tree_selection(self, _event=None) -> None: if self.mode_var.get() == "selected": self._refresh_plan_view() self._update_cli_preview() def _refresh_plan_view(self) -> None: if self._reader is None: return chosen = self._selected_tensor_names() scope_selected = self.mode_var.get() == "selected" n_q2, n_q3, n_q4, n_copy = 0, 0, 0, 0 n_q1 = 0 planned_bytes = 0 for ti in self._reader.tensors: tgt = self._plan_target(ti, chosen, scope_selected) if tgt == "Q2_K": n_q2 += 1 elif tgt == "Q3_K": n_q3 += 1 elif tgt == "Q4_0": n_q4 += 1 elif tgt == "Q1_0": n_q1 += 1 else: n_copy += 1 if tgt == "Q2_K": planned_bytes += (ti.n_elements // QK_K) * 84 elif tgt == "Q3_K": planned_bytes += (ti.n_elements // QK3_K) * Q3_K_BLOCK_BYTES elif tgt == "Q4_0": planned_bytes += (ti.n_elements // QK4_0) * 18 elif tgt == "Q1_0": planned_bytes += (ti.n_elements // QK1_0) * Q1_0_BLOCK_BYTES else: planned_bytes += ti.n_bytes if not self.tree.exists(ti.name): continue tgt_label = tgt if tgt else "Copy" if scope_selected and ti.name not in chosen: state_str = "Not selected (copy)" else: state_str = ( f"Pending ({tgt_label})" if tgt else (f"Keep {ti.type_name} (<= target)" if ti.type_name in ("Q1_0", "Q2_K", "Q3_K", "Q4_0") else "Passthrough") ) vals = list(self.tree.item(ti.name, "values")) vals[4] = tgt_label vals[6] = state_str self.tree.item(ti.name, tags=(tgt_label,), values=vals) if planned_bytes >= 1024 ** 3: planned_text = f"{planned_bytes / (1024 ** 3):.2f} GiB" elif planned_bytes >= 1024 ** 2: planned_text = f"{planned_bytes / (1024 ** 2):.1f} MiB" else: planned_text = f"{planned_bytes / 1024:.1f} KiB" scope_text = "selected" if scope_selected else "all" self.status_var.set( f"Plan ({self.quant_format_var.get().upper()}, {scope_text}): {n_q1} Q1_0 | {n_q2} Q2_K | {n_q3} Q3_K | {n_q4} Q4_0 | {n_copy} Copy " f"(Total: {len(self._reader.tensors)} tensors, est. {planned_text})" ) def _populate_tensor_tree(self) -> None: chosen = self._selected_tensor_names() scope_selected = self.mode_var.get() == "selected" self.tree.delete(*self.tree.get_children()) if self._reader is None: return filt = self.filter_var.get().strip().lower() plan_filt = self.plan_filter_var.get() n_q2, n_q3, n_q4, n_copy = 0, 0, 0, 0 n_q1 = 0 planned_bytes = 0 for ti in self._reader.tensors: tgt = self._plan_target(ti, chosen, scope_selected) if tgt == "Q2_K": n_q2 += 1 elif tgt == "Q3_K": n_q3 += 1 elif tgt == "Q4_0": n_q4 += 1 elif tgt == "Q1_0": n_q1 += 1 else: n_copy += 1 if tgt == "Q2_K": planned_bytes += (ti.n_elements // QK_K) * 84 elif tgt == "Q3_K": planned_bytes += (ti.n_elements // QK3_K) * Q3_K_BLOCK_BYTES elif tgt == "Q4_0": planned_bytes += (ti.n_elements // QK4_0) * 18 elif tgt == "Q1_0": planned_bytes += (ti.n_elements // QK1_0) * Q1_0_BLOCK_BYTES else: planned_bytes += ti.n_bytes tgt_label = tgt if tgt else "Copy" if plan_filt != "ALL" and tgt_label.upper() != plan_filt: continue if ( filt and filt not in ti.name.lower() and filt not in ti.type_name.lower() and filt not in tgt_label.lower() ): continue has_imat = ( self._imatrix is not None and lookup_imatrix(self._imatrix, ti.name) is not None ) if scope_selected and ti.name not in chosen: state_str = "Not selected (copy)" else: state_str = ( f"Pending ({tgt_label})" if tgt else (f"Keep {ti.type_name} (<= target)" if ti.type_name in ("Q1_0", "Q2_K", "Q3_K", "Q4_0") else "Passthrough") ) self.tree.insert( "", tk.END, iid=ti.name, tags=(tgt_label,), values=( ti.name, ti.type_name, str(ti.dims), ti.row_width, tgt_label, "Yes" if has_imat else "-", state_str, ), ) keep_selected = [n for n in chosen if self.tree.exists(n)] if keep_selected: self.tree.selection_set(keep_selected) if planned_bytes >= 1024 ** 3: planned_text = f"{planned_bytes / (1024 ** 3):.2f} GiB" elif planned_bytes >= 1024 ** 2: planned_text = f"{planned_bytes / (1024 ** 2):.1f} MiB" else: planned_text = f"{planned_bytes / 1024:.1f} KiB" scope_text = "selected" if scope_selected else "all" self.status_var.set( f"Plan ({self.quant_format_var.get().upper()}, {scope_text}): {n_q1} Q1_0 | {n_q2} Q2_K | {n_q3} Q3_K | {n_q4} Q4_0 | {n_copy} Copy " f"(Total: {len(self._reader.tensors)} tensors, est. {planned_text})" ) def _select_all_eligible(self) -> None: if self._reader is None: return eligible_ids = [ ti.name for ti in self._reader.tensors if target_quant_type( ti.name, ti.dims, enable_q4_fallback=bool(self.q4_fallback_var.get()) and self.quant_format_var.get() not in ("q2_k", "q3_k"), quantize_embd_q4=bool(self.embd_q4_var.get()), promote_attn_q4=bool(self.promote_attn_q4_var.get()), quant_format=self.quant_format_var.get(), source_type=ti.type, custom_quant_map=self._custom_map_for_plan(), ) is not None and self.tree.exists(ti.name) ] self.tree.selection_set(eligible_ids) def _selected_tensor_names(self) -> set[str]: return set(self.tree.selection()) def stop_worker(self) -> None: if self._worker and self._worker.is_alive(): self._stop_event.set() self.status_var.set("Stopping immediately...") self._log("Cancel requested — halting active block loop...") def start_quantization(self) -> None: if self._worker and self._worker.is_alive(): return gguf_path = self.gguf_var.get().strip() if not gguf_path or not Path(gguf_path).exists(): messagebox.showerror("Missing Source GGUF", "Please select a valid source GGUF file.") return out_path = self.output_var.get().strip() if not out_path: p = Path(gguf_path) out_path = str(p.with_name(f"{p.stem}-Q2K.gguf")) self.output_var.set(out_path) try: cfg = self.build_config() except Exception as exc: messagebox.showerror("Invalid Configuration", str(exc)) return selected = self._selected_tensor_names() if self.mode_var.get() == "selected" else None if self.mode_var.get() == "selected" and not selected: messagebox.showwarning("No Tensor Selected", "Select one or more tensors in the table first.") return imat_path = self.imatrix_var.get().strip() limit = int(self.limit_var.get()) self._stop_event.clear() self._start_time = time.time() self._last_logged_phase = "" self.model_progress_var.set(0.0) self.tensor_progress_var.set(0.0) self.run_btn.configure(state=tk.DISABLED) self.bench_btn.configure(state=tk.DISABLED) self.stop_btn.configure(state=tk.NORMAL) self.status_var.set("Starting quantization...") scope_text = "all eligible tensors" if selected is None else f"{len(selected)} selected tensor(s)" if limit > 0: scope_text += f", LIMIT={limit} quantized tensor(s) (rest will be copied through!)" self._log( f"Starting quantization: source={Path(gguf_path).name} -> output={Path(out_path).name} " f"(scope={scope_text}, cand={cfg.n_cand_d}x{cfg.n_cand_m}, analogs={cfg.use_weight_analogs}, " f"coupled={cfg.factor_coupled_q2k}, optimizer={cfg.optimizer.upper()})" ) self._worker = threading.Thread( target=self._quantize_worker, args=(gguf_path, imat_path, out_path, cfg, selected, limit), daemon=True, ) self._worker.start() def _quantize_worker( self, gguf_path: str, imat_path: str, out_path: str, cfg: HPCConfig, selected: set[str] | None, limit: int, ) -> None: t0 = time.time() try: imat = None if imat_path and Path(imat_path).exists(): self._log_queue.put(("log", f"Loading iMatrix from {imat_path} ...")) imat = read_imatrix(imat_path) self._log_queue.put(("log", f"Loaded iMatrix: {len(imat)} tensor entries ready.")) quantizer = Q2KQuantizer(cfg) if out_path.lower().endswith(".npz") and selected and len(selected) == 1: reader = GGUFReader(gguf_path) tname = next(iter(selected)) ti = reader.info(tname) x, row = reader.read_tensor(tname) imp = None if imat is not None: raw = lookup_imatrix(imat, tname) if raw is not None: imp = expand_importance(raw, x.size, row) tgt = target_quant_type( tname, ti.dims, enable_q4_fallback=bool(getattr(cfg, "enable_q4_fallback", True)), quantize_embd_q4=bool(getattr(cfg, "quantize_embd_q4", True)), promote_attn_q4=bool(getattr(cfg, "promote_attn_q4", False)), quant_format=str(getattr(cfg, "quant_format", "mixed")), source_type=ti.type, custom_quant_map=getattr(cfg, "custom_quant_map", None) or None, ) if tgt == "Q1_0": raw_q1, nb_q1, rmse_q1 = quantize_tensor_q1_0(x, importance=imp, cfg=cfg) np.savez_compressed( out_path, q1_0_bytes=np.frombuffer(raw_q1, dtype=np.uint8), dequantized=dequantize_q1_0_bytes(raw_q1, nb_q1), rmse=np.float32(rmse_q1), ) self._log_queue.put( ("done", f"Saved Q1_0 {out_path} (RMSE={rmse_q1:.6e}, {time.time() - t0:.1f}s)") ) return if tgt == "Q3_K": raw_q3, nb_q3, rmse_q3 = quantize_tensor_q3_k(x, importance=imp, cfg=cfg) np.savez_compressed( out_path, q3_k_bytes=np.frombuffer(raw_q3, dtype=np.uint8), dequantized=dequantize_q3_k_bytes(raw_q3, nb_q3), rmse=np.float32(rmse_q3), ) self._log_queue.put( ("done", f"Saved Q3_K {out_path} (RMSE={rmse_q3:.6e}, {time.time() - t0:.1f}s)") ) return if tgt == "Q4_0": raw_q4, nb_q4, rmse_q4 = quantize_tensor_q4_0(x, importance=imp, cfg=cfg) np.savez_compressed( out_path, q4_0_bytes=np.frombuffer(raw_q4, dtype=np.uint8), dequantized=dequantize_q4_0_bytes(raw_q4, nb_q4), rmse=np.float32(rmse_q4), ) self._log_queue.put( ("done", f"Saved Q4_0 {out_path} (RMSE={rmse_q4:.6e}, {time.time() - t0:.1f}s)") ) return if tgt != "Q2_K": raise ValueError(f"{tname}: target {tgt!r} is not quantizable to .npz") res = quantizer.quantize( x, importance=imp, row_width=row, status_callback=lambda ev: self._log_queue.put(("stage", (1, 1, ti, ev))), stop_event=self._stop_event, ) np.savez_compressed( out_path, scales=res.scales, qs=res.qs, dm=res.dm, dequantized=res.dequantize(), rmse=np.float32(res.stats.rmse), ) self._log_queue.put( ("done", f"Saved {out_path} (RMSE={res.stats.rmse:.6e}, {time.time() - t0:.1f}s)") ) return def on_stage(idx: int, total: int, ti, ev: dict): self._log_queue.put(("stage", (idx, total, ti, ev))) def on_progress(done: int, total: int, name: str, action: str, rmse: float | None, has_imat: bool): self._log_queue.put(("tensor_done", (done, total, name, action, rmse, has_imat))) stats = write_quantized_gguf( gguf_path, out_path, quantizer, imat=imat, quantize_all=(selected is None), tensor_names=selected, limit=limit, verbose=False, progress_callback=on_progress, stage_callback=on_stage, stop_event=self._stop_event, ) dt_sec = time.time() - t0 self._log_queue.put( ( "done", f"Wrote {out_path}: {int(stats.get('q1_0', 0))} Q1_0 + {int(stats.get('q2_k', 0))} Q2_K + " f"{int(stats.get('q3_k', 0))} Q3_K + {int(stats.get('q4_0', 0))} Q4_0 " f"({int(stats['quantized'])}/{int(stats['tensors'])} tensors), " f"overall RMSE={stats['rmse']:.6e} ({dt_sec:.1f}s)", ) ) except InterruptedError as exc: self._log_queue.put(("cancelled", str(exc))) except Exception as exc: self._log_queue.put(("error", str(exc))) def start_benchmark(self) -> None: if self._worker and self._worker.is_alive(): return try: cfg = self.build_config() except Exception as exc: messagebox.showerror("Invalid Configuration", str(exc)) return self._stop_event.clear() self._start_time = time.time() self.model_progress_var.set(0.0) self.tensor_progress_var.set(0.0) self.run_btn.configure(state=tk.DISABLED) self.bench_btn.configure(state=tk.DISABLED) self.stop_btn.configure(state=tk.NORMAL) self.status_var.set("Running factored solver benchmark...") self._worker = threading.Thread(target=self._bench_worker, args=(cfg,), daemon=True) self._worker.start() def _bench_worker(self, cfg: HPCConfig) -> None: from collections import defaultdict try: cases = synthetic_cases(123, 8) gains = defaultdict(list) total = len(cases) for idx, (kind, x) in enumerate(cases, 1): if self._stop_event.is_set(): raise InterruptedError("Benchmark cancelled.") w = np.ones(x.size, dtype=np.float32) v2 = factor_only_solve(x, w, cfg) ref_obj, _ = reference_solve(x, w, cfg, cfg.use_spectral) gain_pct = (ref_obj - v2.objective) / max(abs(ref_obj), 1.0e-30) * 100.0 gains[kind].append(gain_pct / 100.0) self._log_queue.put( ( "bench_step", ( idx, total, kind, f"Case {idx}/{total} [{kind}]: ref={ref_obj:.5e} -> V2={v2.objective:.5e} " f"(+{gain_pct:.2f}%, d={float(v2.d):.5f}, dmin={float(v2.dmin):.5f}, starts={v2.starts})", ), ) ) self._log_queue.put( ("log", f"=== Summary (spectral={cfg.use_spectral}, analogs={cfg.use_weight_analogs}) ===") ) for kind, vals in gains.items(): arr = np.asarray(vals) self._log_queue.put( ( "log", f" {kind:14s} mean={arr.mean()*100:6.2f}% median={np.median(arr)*100:6.2f}% " f"min={arr.min()*100:6.2f}% worse={int(np.sum(arr < -1e-9))}/{arr.size}", ) ) self._log_queue.put(("done", "Benchmark complete.")) except InterruptedError as exc: self._log_queue.put(("cancelled", str(exc))) except Exception as exc: self._log_queue.put(("error", str(exc))) def _update_tree_row_state(self, name: str, state_text: str, focus: bool = False) -> None: if not self.tree.exists(name): return vals = list(self.tree.item(name, "values")) if len(vals) >= 7: vals[6] = state_text self.tree.item(name, values=vals) if focus: self.tree.see(name) def _poll_queue(self) -> None: while True: try: kind, payload = self._log_queue.get_nowait() except queue.Empty: break if kind == "log": self._log(str(payload)) elif kind == "stage": idx, total, ti, ev = payload # type: ignore[misc] phase = ev.get("phase", "") done = int(ev.get("done", 0)) tot_blk = max(int(ev.get("total", 1)), 1) detail = ev.get("detail", "") blk_pct = 100.0 * done / tot_blk overall_pct = float( ev.get("model_pct", 100.0 * ((idx - 1) + (done / tot_blk)) / max(total, 1)) ) self.model_progress_var.set(overall_pct) self.tensor_progress_var.set(blk_pct) self.tele_tensor_var.set(f"[{idx}/{total}] {ti.name} ({ti.type_name}, dims={ti.dims})") self.tele_phase_var.set(phase) if "model_total_blocks" in ev and ev["model_total_blocks"] > 0: self.tele_blocks_var.set( f"Tensor: {done:,}/{tot_blk:,} ({blk_pct:.1f}%) | " f"Model: {ev['model_done_blocks']:,}/{ev['model_total_blocks']:,} ({overall_pct:.1f}%)" ) else: self.tele_blocks_var.set(f"{done:,} / {tot_blk:,} ({blk_pct:.1f}%)") if "rate" in ev and "eta" in ev: t_eta = _fmt_eta(float(ev["eta"])) m_eta = _fmt_eta(float(ev.get("model_eta", ev["eta"]))) self.tele_speed_var.set( f"{ev['rate']:.1f} blk/s | Tensor ETA: {t_eta} | Model ETA: {m_eta}" ) if "rmse" in ev and "sigma" in ev: r, s = float(ev["rmse"]), float(ev["sigma"]) self.tele_rmse_var.set(f"{r:.6e} (r/σ={r/max(s, 1e-15):.4f})") if "temperature" in ev or "cand_floors" in ev: t_val = ev.get("temperature", 0.0) cf = ev.get("cand_floors", 0) ff = ev.get("fac_floors", 0) self.tele_floors_var.set(f"T={t_val:.2e} | cand={cf}, fac={ff}") self.status_var.set(f"[{idx}/{total}] {ti.name} — {phase} ({blk_pct:.0f}%)") self._update_tree_row_state(ti.name, f"{phase.split(':')[0]} ({blk_pct:.0f}%)", focus=True) is_Copy = phase.startswith("Passthrough Copy") phase_key = f"{ti.name}::{phase}" if not is_Copy: if self.log_level_var.get() == "ultra" or phase_key != self._last_logged_phase: self._log(f"[{idx}/{total}] {ti.name} | {phase} | {detail}") self._last_logged_phase = phase_key elif kind == "tensor_done": done, total, name, action, rmse, has_imat = payload # type: ignore[misc] self.model_progress_var.set(100.0 * done / max(total, 1)) self.tensor_progress_var.set(100.0) if action in ("Q1_0", "Q2_K", "Q3_K", "Q4_0") and rmse is not None: self._update_tree_row_state(name, f"{action} (RMSE={rmse:.4e})") self._log( f"[{done}/{total}] FINISHED {name:42s} -> {action} RMSE={rmse:.6e} (iMatrix={'yes' if has_imat else 'no'})" ) else: self._update_tree_row_state(name, "Copied verbatim") if self.log_level_var.get() == "ultra" or done % 25 == 0 or done == total: self._log(f"[{done}/{total}] COPIED {name}") elif kind == "bench_step": idx, total, bkind, msg = payload # type: ignore[misc] pct = 100.0 * idx / max(total, 1) self.model_progress_var.set(pct) self.tensor_progress_var.set(pct) self.tele_tensor_var.set(f"Synthetic Benchmark ({bkind})") self.tele_phase_var.set(f"Block {idx}/{total}") self.tele_blocks_var.set(f"{idx} / {total}") self._log(str(msg)) elif kind == "done": self.model_progress_var.set(100.0) self.tensor_progress_var.set(100.0) self.tele_phase_var.set("Complete") self.status_var.set(str(payload)) self._log(str(payload)) self._reset_buttons() elif kind in ("cancelled", "error"): self.tele_phase_var.set(kind.upper()) self.status_var.set(f"{kind.upper()}: {payload}") self._log(f"{kind.upper()}: {payload}") self._reset_buttons() self.root.after(80, self._poll_queue) def _reset_buttons(self) -> None: self.run_btn.configure(state=tk.NORMAL) self.bench_btn.configure(state=tk.NORMAL) self.stop_btn.configure(state=tk.DISABLED) def main() -> None: parser = argparse.ArgumentParser(description="HPC-Quantize V2 GUI") parser.add_argument("--gguf", help="pre-populate source GGUF path") parser.add_argument("--imatrix", help="pre-populate iMatrix path") parser.add_argument("--output", help="pre-populate output GGUF path") parser.add_argument("--check", action="store_true", help="validate GUI bindings without opening a window") args = parser.parse_args() if args.check: cfg = HPCConfig().validate() print("GUI check OK:", cfg) return root = tk.Tk() app = HPCQuantizeGUI(root) if args.gguf: app.gguf_var.set(args.gguf) if args.imatrix: app.imatrix_var.set(args.imatrix) if args.output: app.output_var.set(args.output) if args.gguf or args.imatrix: app.inspect_files() root.mainloop() if __name__ == "__main__": main()