| |
| """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] = { |
| |
| "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", |
| |
| "attn_output": "Q1_0", |
| |
| "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) |
|
|
| |
| 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("<Configure>", _on_frame_configure) |
| self._canvas.bind("<Configure>", _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("<MouseWheel>", _on_mousewheel) |
| self.root.bind_all("<Button-4>", _on_mousewheel) |
| self.root.bind_all("<Button-5>", _on_mousewheel) |
|
|
| |
| 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) |
|
|
| |
| 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) |
|
|
| |
| 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)) |
|
|
| |
| 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("<KeyRelease>", 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("<<TreeviewSelect>>", 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) |
|
|
| |
| 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) |
|
|
| |
| 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 |
| 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 |
| 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 |
| 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() |
|
|
