Re-write app to OOP with additional options

app.py

@@ -1,443 +1,873 @@
 import os
 import subprocess
 import signal
-os.environ["GRADIO_ANALYTICS_ENABLED"] = "False"
-import gradio as gr
 import tempfile
+from pathlib import Path
+from textwrap import dedent
+from typing import Optional, Tuple, List, Union
+from dataclasses import dataclass, field
 
+os.environ["GRADIO_ANALYTICS_ENABLED"] = "False"
+
+import gradio as gr
 from huggingface_hub import HfApi, ModelCard, whoami
 from gradio_huggingfacehub_search import HuggingfaceHubSearch
-from pathlib import Path
-from textwrap import dedent
 from apscheduler.schedulers.background import BackgroundScheduler
 
 
-# used for restarting the space
-HF_TOKEN = os.environ.get("HF_TOKEN")
-CONVERSION_SCRIPT = "./llama.cpp/convert_hf_to_gguf.py"
-
-# escape HTML for logging
-def escape(s: str) -> str:
-    s = s.replace("&", "&") # Must be done first!
-    s = s.replace("<", "&lt;")
-    s = s.replace(">", "&gt;")
-    s = s.replace('"', "&quot;")
-    s = s.replace("\n", "<br/>")
-    return s
-
-def generate_importance_matrix(model_path: str, train_data_path: str, output_path: str):
-    imatrix_command = [
-        "./llama.cpp/llama-imatrix",
-        "-m", model_path,
-        "-f", train_data_path,
-        "-ngl", "99",
-        "--output-frequency", "10",
-        "-o", output_path,
-    ]
-
-    if not os.path.isfile(model_path):
-        raise Exception(f"Model file not found: {model_path}")
-
-    print("Running imatrix command...")
-    process = subprocess.Popen(imatrix_command, shell=False)
-
-    try:
-        process.wait(timeout=60)  # added wait
-    except subprocess.TimeoutExpired:
-        print("Imatrix computation timed out. Sending SIGINT to allow graceful termination...")
-        process.send_signal(signal.SIGINT)
+@dataclass
+class QuantizationConfig:
+    """Configuration for model quantization."""
+    method: str
+    use_imatrix: bool = False
+    imatrix_method: str = "IQ4_NL"
+    quant_embedding: bool = False
+    embedding_tensor_method: str = "Q8_0"
+    leave_output: bool = False
+    quant_output: bool = False
+    output_tensor_method: str = "Q8_0"
+    # Generated values - These will be set during processing
+    fp16_model: str = field(default="", init=False)
+    quantized_gguf: str = field(default="", init=False)
+    imatrix_file: str = field(default="", init=False)
+
+
+@dataclass
+class SplitConfig:
+    """Configuration for model splitting."""
+    enabled: bool = False
+    max_tensors: int = 256
+    max_size: Optional[str] = None
+
+
+@dataclass
+class OutputConfig:
+    """Configuration for output settings."""
+    private_repo: bool = False
+    repo_name: str = ""
+    filename: str = ""
+
+
+@dataclass
+class ModelProcessingConfig:
+    """Configuration for the entire model processing pipeline."""
+    token: str
+    model_id: str
+    model_name: str
+    outdir: str
+    quant_config: QuantizationConfig
+    split_config: SplitConfig
+    output_config: OutputConfig
+    # Generated values - These will be set during processing
+    new_repo_url: str = field(default="", init=False)
+    new_repo_id: str = field(default="", init=False)
+
+
+class GGUFConverterError(Exception):
+    """Custom exception for GGUF conversion errors."""
+    pass
+
+
+class HuggingFaceModelProcessor:
+    """Handles the processing of Hugging Face models to GGUF format."""
+
+    ERROR_LOGIN = "You must be logged in to use GGUF-my-repo."
+    DOWNLOAD_FOLDER = "./downloads"
+    OUTPUT_FOLDER = "./outputs"
+    CALIBRATION_FILE = "calibration_data_v5_rc.txt"
+
+    QUANTIZE_TIMEOUT=86400
+    HF_TO_GGUF_TIMEOUT=3600
+    IMATRIX_TIMEOUT=86400
+    SPLIT_TIMEOUT=3600
+    KILL_TIMEOUT=5
+
+    def __init__(self):
+        self.SPACE_ID = os.environ.get("SPACE_ID", "")
+        self.SPACE_URL = f"https://{self.SPACE_ID.replace('/', '-')}.hf.space/" if self.SPACE_ID else "http://localhost:7860/"
+        self.HF_TOKEN = os.environ.get("HF_TOKEN")
+        self.RUN_LOCALLY = os.environ.get("RUN_LOCALLY")
+
+        # Create necessary folders
+        self._create_folder(self.DOWNLOAD_FOLDER)
+        self._create_folder(self.OUTPUT_FOLDER)
+
+    def _create_folder(self, folder_name: str) -> str:
+        """Create a folder if it doesn't exist."""
+        if not os.path.exists(folder_name):
+            print(f"Creating folder: {folder_name}")
+            os.makedirs(folder_name)
+        return folder_name
+
+    def _validate_token(self, oauth_token: Optional[gr.OAuthToken]) -> str:
+        """Validate the OAuth token and return the token string."""
+        if oauth_token is None or oauth_token.token is None:
+            raise GGUFConverterError(self.ERROR_LOGIN)
+
+        try:
+            whoami(oauth_token.token)
+            return oauth_token.token
+        except Exception as e:
+            raise GGUFConverterError(self.ERROR_LOGIN)
+
+    def _escape_html(self, s: str) -> str:
+        """Escape HTML characters for safe display."""
+        replacements = [
+            ("&", "&amp;"),
+            ("<", "&lt;"),
+            (">", "&gt;"),
+            ('"', "&quot;"),
+            ("\n", "<br/>")
+        ]
+        for old, new in replacements:
+            s = s.replace(old, new)
+        return s
+
+    def _get_model_creator(self, model_id: str) -> str:
+        """Extract model creator from model ID."""
+        return model_id.split('/')[0]
+
+    def _get_model_name(self, model_id: str) -> str:
+        """Extract model name from model ID."""
+        return model_id.split('/')[-1]
+
+    def _upload_file(self, processing_config: ModelProcessingConfig, path_or_fileobj: str, path_in_repo: str) -> None:
+        """Upload a file to Hugging Face repository."""
+        if self.RUN_LOCALLY == "1":
+            print("Skipping upload...")
+            return
+
+        api = HfApi(token=processing_config.token)
+        api.upload_file(
+            path_or_fileobj=path_or_fileobj,
+            path_in_repo=path_in_repo,
+            repo_id=processing_config.new_repo_id,
+        )
+
+    def _generate_importance_matrix(self, quant_config: QuantizationConfig) -> None:
+        """Generate importance matrix for quantization."""
+        if not os.path.isfile(quant_config.fp16_model):
+            raise GGUFConverterError(f"Model file not found: {quant_config.fp16_model}")
+
+        train_data_path = self.CALIBRATION_FILE
+        if not os.path.isfile(train_data_path):
+            raise GGUFConverterError(f"Training data file not found: {train_data_path}")
+
+        print(f"Training data file path: {train_data_path}")
+        print("Running imatrix command...")
+
+        imatrix_command = [
+            "llama-imatrix",
+            "-m", quant_config.fp16_model,
+            "-f", train_data_path,
+            "-ngl", "99",
+            "--output-frequency", "10",
+            "-o", quant_config.imatrix_file,
+        ]
+
+        process = subprocess.Popen(imatrix_command, shell=False, stderr=subprocess.STDOUT)
+        try:
+            process.wait(timeout=self.IMATRIX_TIMEOUT)
+        except subprocess.TimeoutExpired:
+            print("Imatrix computation timed out. Sending SIGINT to allow graceful termination...")
+            process.send_signal(signal.SIGINT)
+            try:
+                process.wait(timeout=self.KILL_TIMEOUT)
+            except subprocess.TimeoutExpired:
+                print("Imatrix proc still didn't term. Forcefully terminating process...")
+                process.kill()
+            raise GGUFConverterError("Error generating imatrix: Operation timed out.")
+
+        if process.returncode != 0:
+            raise GGUFConverterError(f"Error generating imatrix: code={process.returncode}.")
+
+        print(f"Importance matrix generation completed: {os.path.abspath(quant_config.imatrix_file)}")
+
+    def _split_and_upload_model(self, processing_config: ModelProcessingConfig) -> None:
+        """Split large model files and upload shards."""
+        quant_config = processing_config.quant_config
+        split_config = processing_config.split_config
+
+        print(f"Model path: {quant_config.quantized_gguf}")
+        print(f"Output dir: {processing_config.outdir}")
+
+        split_cmd = ["llama-gguf-split", "--split"]
+
+        if split_config.max_size:
+            split_cmd.extend(["--split-max-size", split_config.max_size])
+        else:
+            split_cmd.extend(["--split-max-tensors", str(split_config.max_tensors)])
+
+        model_path_prefix = '.'.join(quant_config.quantized_gguf.split('.')[:-1])
+        split_cmd.extend([quant_config.quantized_gguf, model_path_prefix])
+
+        print(f"Split command: {split_cmd}")
+        process = subprocess.Popen(split_cmd, shell=False, stderr=subprocess.STDOUT)
         try:
-            process.wait(timeout=5)  # grace period
+            process.wait(timeout=self.SPLIT_TIMEOUT)
         except subprocess.TimeoutExpired:
-            print("Imatrix proc still didn't term. Forecfully terming process...")
-            process.kill()
-
-    print("Importance matrix generation completed.")
-
-def split_upload_model(model_path: str, outdir: str, repo_id: str, oauth_token: gr.OAuthToken | None, split_max_tensors=256, split_max_size=None):
-    print(f"Model path: {model_path}")
-    print(f"Output dir: {outdir}")
-
-    if oauth_token is None or oauth_token.token is None:
-        raise ValueError("You have to be logged in.")
-    
-    split_cmd = [
-        "./llama.cpp/llama-gguf-split",
-        "--split",
-    ]
-    if split_max_size:
-        split_cmd.append("--split-max-size")
-        split_cmd.append(split_max_size)
-    else:
-        split_cmd.append("--split-max-tensors")
-        split_cmd.append(str(split_max_tensors))
-
-    # args for output
-    model_path_prefix = '.'.join(model_path.split('.')[:-1]) # remove the file extension
-    split_cmd.append(model_path)
-    split_cmd.append(model_path_prefix)
-
-    print(f"Split command: {split_cmd}") 
-    
-    result = subprocess.run(split_cmd, shell=False, capture_output=True, text=True)
-    print(f"Split command stdout: {result.stdout}") 
-    print(f"Split command stderr: {result.stderr}") 
-    
-    if result.returncode != 0:
-        stderr_str = result.stderr.decode("utf-8")
-        raise Exception(f"Error splitting the model: {stderr_str}")
-    print("Model split successfully!")
-
-    # remove the original model file if needed
-    if os.path.exists(model_path):
-        os.remove(model_path)
-
-    model_file_prefix = model_path_prefix.split('/')[-1]
-    print(f"Model file name prefix: {model_file_prefix}") 
-    sharded_model_files = [f for f in os.listdir(outdir) if f.startswith(model_file_prefix) and f.endswith(".gguf")]
-    if sharded_model_files:
+            print("Splitting timed out. Sending SIGINT to allow graceful termination...")
+            process.send_signal(signal.SIGINT)
+            try:
+                process.wait(timeout=self.KILL_TIMEOUT)
+            except subprocess.TimeoutExpired:
+                print("Splitting timed out. Killing process...")
+                process.kill()
+            raise GGUFConverterError("Error splitting the model: Operation timed out.")
+
+        if process.returncode != 0:
+            raise GGUFConverterError(f"Error splitting the model: code={process.returncode}")
+
+        print("Model split successfully!")
+
+        # Remove original model file
+        if os.path.exists(quant_config.quantized_gguf):
+            os.remove(quant_config.quantized_gguf)
+
+        model_file_prefix = model_path_prefix.split('/')[-1]
+        print(f"Model file name prefix: {model_file_prefix}")
+
+        sharded_model_files = [
+            f for f in os.listdir(processing_config.outdir)
+            if f.startswith(model_file_prefix) and f.endswith(".gguf")
+        ]
+
+        if not sharded_model_files:
+            raise GGUFConverterError("No sharded files found.")
+
         print(f"Sharded model files: {sharded_model_files}")
-        api = HfApi(token=oauth_token.token)
+
         for file in sharded_model_files:
-            file_path = os.path.join(outdir, file)
-            print(f"Uploading file: {file_path}")
+            file_path = os.path.join(processing_config.outdir, file)
             try:
-                api.upload_file(
-                    path_or_fileobj=file_path,
-                    path_in_repo=file,
-                    repo_id=repo_id,
-                )
+                print(f"Uploading file: {file_path}")
+                self._upload_file(processing_config, file_path, file)
             except Exception as e:
-                raise Exception(f"Error uploading file {file_path}: {e}")
-    else:
-        raise Exception("No sharded files found.")
-    
-    print("Sharded model has been uploaded successfully!")
-
-def process_model(model_id, q_method, use_imatrix, imatrix_q_method, private_repo, train_data_file, split_model, split_max_tensors, split_max_size, oauth_token: gr.OAuthToken | None):
-    if oauth_token is None or oauth_token.token is None:
-        raise gr.Error("You must be logged in to use GGUF-my-repo")
-
-    # validate the oauth token
-    try:
-        whoami(oauth_token.token)
-    except Exception as e:
-        raise gr.Error("You must be logged in to use GGUF-my-repo")
-
-    model_name = model_id.split('/')[-1]
-
-    try:
-        api = HfApi(token=oauth_token.token)
-
-        dl_pattern = ["*.md", "*.json", "*.model"]
-
-        pattern = (
-            "*.safetensors"
-            if any(
-                file.path.endswith(".safetensors")
-                for file in api.list_repo_tree(
-                    repo_id=model_id,
-                    recursive=True,
+                raise GGUFConverterError(f"Error uploading file {file_path}: {e}")
+
+        print("Sharded model has been uploaded successfully!")
+
+    def _download_base_model(self, processing_config: ModelProcessingConfig) -> str:
+        """Download and convert Hugging Face model to GGUF FP16 format."""
+        print(f"Downloading model {processing_config.model_name}")
+
+        if os.path.exists(processing_config.quant_config.fp16_model):
+            print("Skipping fp16 conversion...")
+            print(f"Converted model path: {os.path.abspath(processing_config.quant_config.fp16_model)}")
+            return processing_config.quant_config.fp16_model
+
+        with tempfile.TemporaryDirectory(dir=self.DOWNLOAD_FOLDER) as tmpdir:
+            local_dir = f"{Path(tmpdir)}/{processing_config.model_name}"
+            print(f"Local directory: {os.path.abspath(local_dir)}")
+
+            # Download model
+            api = HfApi(token=processing_config.token)
+            pattern = (
+                "*.safetensors"
+                if any(
+                    file.path.endswith(".safetensors")
+                    for file in api.list_repo_tree(
+                        repo_id=processing_config.model_id,
+                        recursive=True,
+                    )
                 )
+                else "*.bin"
             )
-            else "*.bin"
+            dl_pattern = ["*.md", "*.json", "*.model"]
+            dl_pattern += [pattern]
+            api.snapshot_download(repo_id=processing_config.model_id, local_dir=local_dir, allow_patterns=dl_pattern)
+            print("Model downloaded successfully!")
+            print(f"Model directory contents: {os.listdir(local_dir)}")
+
+            config_dir = os.path.join(local_dir, "config.json")
+            adapter_config_dir = os.path.join(local_dir, "adapter_config.json")
+            if os.path.exists(adapter_config_dir) and not os.path.exists(config_dir):
+                raise GGUFConverterError(
+                    'adapter_config.json is present.<br/><br/>If you are converting a LoRA adapter to GGUF, '
+                    'please use <a href="https://huggingface.co/spaces/ggml-org/gguf-my-lora" target="_blank" '
+                    'style="text-decoration:underline">GGUF-my-lora</a>.'
+                )
+
+            # Convert HF to GGUF
+            print(f"Converting to GGUF FP16: {os.path.abspath(processing_config.quant_config.fp16_model)}")
+            convert_command = [
+                "python3", "/app/convert_hf_to_gguf.py", local_dir,
+                "--outtype", "f16", "--outfile", processing_config.quant_config.fp16_model
+            ]
+            process = subprocess.Popen(convert_command, shell=False, stderr=subprocess.STDOUT)
+            try:
+                process.wait(timeout=self.HF_TO_GGUF_TIMEOUT)
+            except subprocess.TimeoutExpired:
+                print("Conversion timed out. Sending SIGINT to allow graceful termination...")
+                process.send_signal(signal.SIGINT)
+                try:
+                    process.wait(timeout=self.KILL_TIMEOUT)
+                except subprocess.TimeoutExpired:
+                    print("Conversion timed out. Killing process...")
+                    process.kill()
+                raise GGUFConverterError("Error converting to fp16: Operation timed out.")
+
+            if process.returncode != 0:
+                raise GGUFConverterError(f"Error converting to fp16: code={process.returncode}")
+
+            print("Model converted to fp16 successfully!")
+            print(f"Converted model path: {os.path.abspath(processing_config.quant_config.fp16_model)}")
+            return processing_config.quant_config.fp16_model
+
+    def _quantize_model(self, quant_config: QuantizationConfig) -> str:
+        """Quantize the GGUF model."""
+        quantize_cmd = ["llama-quantize"]
+
+        if quant_config.quant_embedding:
+            quantize_cmd.extend(["--token-embedding-type", quant_config.embedding_tensor_method])
+
+        if quant_config.leave_output:
+            quantize_cmd.append("--leave-output-tensor")
+        else:
+            if quant_config.quant_output:
+                quantize_cmd.extend(["--output-tensor-type", quant_config.output_tensor_method])
+
+        # Set imatrix file path if needed
+        if quant_config.use_imatrix:
+            self._generate_importance_matrix(quant_config)
+            quantize_cmd.extend(["--imatrix", quant_config.imatrix_file])
+        else:
+            print("Not using imatrix quantization.")
+
+        quantize_cmd.append(quant_config.fp16_model)
+        quantize_cmd.append(quant_config.quantized_gguf)
+
+        if quant_config.use_imatrix:
+            quantize_cmd.append(quant_config.imatrix_method)
+        else:
+            quantize_cmd.append(quant_config.method)
+
+        print(f"Quantizing model with {quantize_cmd}")
+
+        # Use Popen for quantization
+        process = subprocess.Popen(quantize_cmd, shell=False, stderr=subprocess.STDOUT)
+        try:
+            process.wait(timeout=self.QUANTIZE_TIMEOUT)
+        except subprocess.TimeoutExpired:
+            print("Quantization timed out. Sending SIGINT to allow graceful termination...")
+            process.send_signal(signal.SIGINT)
+            try:
+                process.wait(timeout=self.KILL_TIMEOUT)
+            except subprocess.TimeoutExpired:
+                print("Quantization timed out. Killing process...")
+                process.kill()
+            raise GGUFConverterError("Error quantizing: Operation timed out.")
+
+        if process.returncode != 0:
+            raise GGUFConverterError(f"Error quantizing: code={process.returncode}")
+
+        print(f"Quantized successfully with {quant_config.imatrix_method if quant_config.use_imatrix else quant_config.method} option!")
+        print(f"Quantized model path: {os.path.abspath(quant_config.quantized_gguf)}")
+        return quant_config.quantized_gguf
+
+    def _create_empty_repo(self, processing_config: ModelProcessingConfig):
+        api = HfApi(token=processing_config.token)
+        new_repo_url = api.create_repo(
+            repo_id=processing_config.output_config.repo_name,
+            exist_ok=True,
+            private=processing_config.output_config.private_repo
         )
+        processing_config.new_repo_url = new_repo_url.url
+        processing_config.new_repo_id = new_repo_url.repo_id
+        print("Repo created successfully!", processing_config.new_repo_url)
 
-        dl_pattern += [pattern]
-
-        if not os.path.exists("downloads"):
-            os.makedirs("downloads")
-
-        if not os.path.exists("outputs"):
-            os.makedirs("outputs")
-
-        with tempfile.TemporaryDirectory(dir="outputs") as outdir:
-            fp16 = str(Path(outdir)/f"{model_name}.fp16.gguf")
-
-            with tempfile.TemporaryDirectory(dir="downloads") as tmpdir:
-                # Keep the model name as the dirname so the model name metadata is populated correctly
-                local_dir = Path(tmpdir)/model_name
-                print(local_dir)
-                api.snapshot_download(repo_id=model_id, local_dir=local_dir, local_dir_use_symlinks=False, allow_patterns=dl_pattern)
-                print("Model downloaded successfully!")
-                print(f"Current working directory: {os.getcwd()}")
-                print(f"Model directory contents: {os.listdir(local_dir)}")
-
-                config_dir = local_dir/"config.json"
-                adapter_config_dir = local_dir/"adapter_config.json"
-                if os.path.exists(adapter_config_dir) and not os.path.exists(config_dir):
-                    raise Exception('adapter_config.json is present.<br/><br/>If you are converting a LoRA adapter to GGUF, please use <a href="https://huggingface.co/spaces/ggml-org/gguf-my-lora" target="_blank" style="text-decoration:underline">GGUF-my-lora</a>.')
-
-                result = subprocess.run([
-                    "python", CONVERSION_SCRIPT, local_dir, "--outtype", "f16", "--outfile", fp16
-                ], shell=False, capture_output=True)
-                print(result)
-                if result.returncode != 0:
-                    stderr_str = result.stderr.decode("utf-8")
-                    raise Exception(f"Error converting to fp16: {stderr_str}")
-                print("Model converted to fp16 successfully!")
-                print(f"Converted model path: {fp16}")
-
-            imatrix_path = Path(outdir)/"imatrix.dat"
-
-            if use_imatrix:
-                if train_data_file:
-                    train_data_path = train_data_file.name
-                else:
-                    train_data_path = "llama.cpp/groups_merged.txt" #fallback calibration dataset
-
-                print(f"Training data file path: {train_data_path}")
-
-                if not os.path.isfile(train_data_path):
-                    raise Exception(f"Training data file not found: {train_data_path}")
-
-                generate_importance_matrix(fp16, train_data_path, imatrix_path)
-            else:
-                print("Not using imatrix quantization.")
-            
-            # Quantize the model
-            quantized_gguf_name = f"{model_name.lower()}-{imatrix_q_method.lower()}-imat.gguf" if use_imatrix else f"{model_name.lower()}-{q_method.lower()}.gguf"
-            quantized_gguf_path = str(Path(outdir)/quantized_gguf_name)
-            if use_imatrix:
-                quantise_ggml = [
-                    "./llama.cpp/llama-quantize",
-                    "--imatrix", imatrix_path, fp16, quantized_gguf_path, imatrix_q_method
-                ]
-            else:
-                quantise_ggml = [
-                    "./llama.cpp/llama-quantize",
-                    fp16, quantized_gguf_path, q_method
-                ]
-            result = subprocess.run(quantise_ggml, shell=False, capture_output=True)
-            if result.returncode != 0:
-                stderr_str = result.stderr.decode("utf-8")
-                raise Exception(f"Error quantizing: {stderr_str}")
-            print(f"Quantized successfully with {imatrix_q_method if use_imatrix else q_method} option!")
-            print(f"Quantized model path: {quantized_gguf_path}")
-
-            # Create empty repo
-            username = whoami(oauth_token.token)["name"]
-            new_repo_url = api.create_repo(repo_id=f"{username}/{model_name}-{imatrix_q_method if use_imatrix else q_method}-GGUF", exist_ok=True, private=private_repo)
-            new_repo_id = new_repo_url.repo_id
-            print("Repo created successfully!", new_repo_url)
+        return new_repo_url
+
+    def _generate_readme(self, processing_config: ModelProcessingConfig) -> str:
+        """Generate README.md for the quantized model."""
+        creator = self._get_model_creator(processing_config.model_id)
+        username = whoami(processing_config.token)["name"]
 
+        try:
+            card = ModelCard.load(processing_config.model_id, token=processing_config.token)
+        except:
+            card = ModelCard("")
+
+        if card.data.tags is None:
+            card.data.tags = []
+        card.data.tags.extend(["llama-cpp", "gguf-my-repo"])
+        card.data.base_model = processing_config.model_id
+
+        card.text = dedent(
+            f"""
+# {processing_config.model_name}
+**Model creator:** [{creator}](https://huggingface.co/{creator})<br/>
+**Original model**: [{processing_config.model_id}](https://huggingface.co/{processing_config.model_id})<br/>
+**GGUF quantization:** provided by [{username}](https:/huggingface.co/{username}) using `llama.cpp`<br/>
+## Special thanks
+🙏 Special thanks to [Georgi Gerganov](https://github.com/ggerganov) and the whole team working on [llama.cpp](https://github.com/ggerganov/llama.cpp/) for making all of this possible.
+## Use with Ollama
+```bash
+ollama run "hf.co/{processing_config.new_repo_id}:<quantization>"
+```
+## Use with LM Studio
+```bash
+lms load "{processing_config.new_repo_id}"
+```
+## Use with llama.cpp CLI
+```bash
+llama-cli --hf-repo "{processing_config.new_repo_id}" --hf-file "{processing_config.output_config.filename}" -p "The meaning to life and the universe is"
+```
+## Use with llama.cpp Server:
+```bash
+llama-server --hf-repo "{processing_config.new_repo_id}" --hf-file "{processing_config.output_config.filename}" -c 4096
+```
+            """
+        )
+
+        readme_path = f"{processing_config.outdir}/README.md"
+        card.save(readme_path)
+        return readme_path
+
+    def process_model(self, processing_config: ModelProcessingConfig) -> Tuple[str, str]:
+        """Main method to process a model through the entire pipeline."""
+        quant_config = processing_config.quant_config
+        split_config = processing_config.split_config
+        output_config = processing_config.output_config
+
+        print(f"Current working directory: {os.path.abspath(os.getcwd())}")
+
+        # Download and convert base model
+        self._download_base_model(processing_config)
+
+        # Quantize the model
+        self._quantize_model(quant_config)
+
+        # Create empty repo
+        self._create_empty_repo(processing_config)
+
+        # Upload model
+        if split_config.enabled:
+            print(f"Splitting quantized model: {os.path.abspath(quant_config.quantized_gguf)}")
+            self._split_and_upload_model(processing_config)
+        else:
+            try:
+                print(f"Uploading quantized model: {os.path.abspath(quant_config.quantized_gguf)}")
+                self._upload_file(processing_config, quant_config.quantized_gguf, output_config.filename)
+            except Exception as e:
+                raise GGUFConverterError(f"Error uploading quantized model: {e}")
+
+        # Upload imatrix if it exists
+        if quant_config.use_imatrix and os.path.isfile(quant_config.imatrix_file):
             try:
-                card = ModelCard.load(model_id, token=oauth_token.token)
-            except:
-                card = ModelCard("")
-            if card.data.tags is None:
-                card.data.tags = []
-            card.data.tags.append("llama-cpp")
-            card.data.tags.append("gguf-my-repo")
-            card.data.base_model = model_id
-            card.text = dedent(
-                f"""
-                # {new_repo_id}
-                This model was converted to GGUF format from [`{model_id}`](https://huggingface.co/{model_id}) using llama.cpp via the ggml.ai's [GGUF-my-repo](https://huggingface.co/spaces/ggml-org/gguf-my-repo) space.
-                Refer to the [original model card](https://huggingface.co/{model_id}) for more details on the model.
-                
-                ## Use with llama.cpp
-                Install llama.cpp through brew (works on Mac and Linux)
-                
-                ```bash
-                brew install llama.cpp
-                
-                ```
-                Invoke the llama.cpp server or the CLI.
-                
-                ### CLI:
-                ```bash
-                llama-cli --hf-repo {new_repo_id} --hf-file {quantized_gguf_name} -p "The meaning to life and the universe is"
-                ```
-                
-                ### Server:
-                ```bash
-                llama-server --hf-repo {new_repo_id} --hf-file {quantized_gguf_name} -c 2048
-                ```
-                
-                Note: You can also use this checkpoint directly through the [usage steps](https://github.com/ggerganov/llama.cpp?tab=readme-ov-file#usage) listed in the Llama.cpp repo as well.
-
-                Step 1: Clone llama.cpp from GitHub.
-                ```
-                git clone https://github.com/ggerganov/llama.cpp
-                ```
-
-                Step 2: Move into the llama.cpp folder and build it with `LLAMA_CURL=1` flag along with other hardware-specific flags (for ex: LLAMA_CUDA=1 for Nvidia GPUs on Linux).
-                ```
-                cd llama.cpp && LLAMA_CURL=1 make
-                ```
-
-                Step 3: Run inference through the main binary.
-                ```
-                ./llama-cli --hf-repo {new_repo_id} --hf-file {quantized_gguf_name} -p "The meaning to life and the universe is"
-                ```
-                or 
-                ```
-                ./llama-server --hf-repo {new_repo_id} --hf-file {quantized_gguf_name} -c 2048
-                ```
-                """
+                print(f"Uploading imatrix.dat: {os.path.abspath(quant_config.imatrix_file)}")
+                self._upload_file(processing_config, quant_config.imatrix_file, f"{processing_config.model_name}-imatrix.gguf")
+            except Exception as e:
+                raise GGUFConverterError(f"Error uploading imatrix.dat: {e}")
+
+        # Upload README.md
+        readme_path = self._generate_readme(processing_config)
+        self._upload_file(processing_config, readme_path, "README.md")
+
+        print(f"Uploaded successfully with {quant_config.imatrix_method if quant_config.use_imatrix else quant_config.method} option!")
+
+
+class GGUFConverterUI:
+    """Gradio UI for the GGUF Converter."""
+
+    def __init__(self):
+        self.processor = HuggingFaceModelProcessor()
+        self.css = """/* Custom CSS to allow scrolling */
+        .gradio-container {overflow-y: auto;}
+        """
+
+        # Initialize components
+        self._initialize_components()
+        self._setup_interface()
+
+    def _initialize_components(self):
+        """Initialize all UI components."""
+        #####
+        # Base model section
+        #####
+        self.model_id = HuggingfaceHubSearch(
+            label="Hub Model ID",
+            placeholder="Search for model id on Huggingface",
+            search_type="model",
+        )
+
+        #####
+        # Quantization section
+        #####
+        self.use_imatrix = gr.Checkbox(
+            value=False,
+            label="Use Imatrix Quantization",
+            info="Use importance matrix for quantization."
+        )
+        self.q_method = gr.Dropdown(
+            choices=["Q2_K", "Q3_K_S", "Q3_K_M", "Q3_K_L", "Q4_0", "Q4_K_S", "Q4_K_M", "Q5_0", "Q5_K_S", "Q5_K_M", "Q6_K", "Q8_0", "F16", "BF16"],
+            label="Quantization Method",
+            info="GGML quantization type",
+            value="Q4_K_M",
+            filterable=False,
+            visible=True
+        )
+        self.imatrix_q_method = gr.Dropdown(
+            choices=["IQ3_M", "IQ3_XXS", "Q4_K_M", "Q4_K_S", "IQ4_NL", "IQ4_XS", "Q5_K_M", "Q5_K_S"],
+            label="Imatrix Quantization Method",
+            info="GGML imatrix quants type",
+            value="IQ4_NL",
+            filterable=False,
+            visible=False
+        )
+        self.train_data_file = gr.File(
+            label="Training Data File",
+            file_types=[".txt"],
+            visible=False
+        )
+
+        #####
+        # Advanced Options section
+        #####
+        self.split_model = gr.Checkbox(
+            value=False,
+            label="Split Model",
+            info="Shard the model using gguf-split."
+        )
+        self.split_max_tensors = gr.Number(
+            value=256,
+            label="Max Tensors per File",
+            info="Maximum number of tensors per file when splitting model.",
+            visible=False
+        )
+        self.split_max_size = gr.Textbox(
+            label="Max File Size",
+            info="Maximum file size when splitting model (--split-max-size). May leave empty to use the default. Accepted suffixes: M, G. Example: 256M, 5G",
+            visible=False
+        )
+        self.leave_output = gr.Checkbox(
+            value=False,
+            label="Leave output tensor",
+            info="Leaves output.weight un(re)quantized"
+        )
+        self.quant_embedding = gr.Checkbox(
+            value=False,
+            label="Quant embeddings tensor",
+            info="Quantize embeddings tensor separately"
+        )
+        self.embedding_tensor_method = gr.Dropdown(
+            choices=["Q2_K", "Q3_K", "Q4_K", "Q5_K", "Q6_K", "Q8_0"],
+            label="Embeddings Quantization Method",
+            info="use a specific quant type for the token embeddings tensor",
+            value="Q8_0",
+            filterable=False,
+            visible=False
+        )
+        self.quant_output = gr.Checkbox(
+            value=False,
+            label="Quant output tensor",
+            info="Quantize output tensor separately"
+        )
+        self.output_tensor_method = gr.Dropdown(
+            choices=["Q2_K", "Q3_K", "Q4_K", "Q5_K", "Q6_K", "Q8_0"],
+            label="Output Quantization Method",
+            info="use a specific quant type for the output.weight tensor",
+            value="Q8_0",
+            filterable=False,
+            visible=False
+        )
+
+        #####
+        # Output Settings section
+        #####
+        self.private_repo = gr.Checkbox(
+            value=False,
+            label="Private Repo",
+            info="Create a private repo under your username."
+        )
+        self.repo_name = gr.Textbox(
+            label="Output Repository Name",
+            info="Set your repository name",
+            max_lines=1
+        )
+        self.gguf_name = gr.Textbox(
+            label="Output File Name",
+            info="Set output file name",
+            max_lines=1
+        )
+
+        #####
+        # Buttons section
+        #####
+        self.clear_btn = gr.ClearButton(
+            value="Clear",
+            variant="secondary",
+            components=[
+                self.model_id,
+                self.q_method,
+                self.use_imatrix,
+                self.imatrix_q_method,
+                self.private_repo,
+                self.train_data_file,
+                self.leave_output,
+                self.quant_embedding,
+                self.embedding_tensor_method,
+                self.quant_output,
+                self.output_tensor_method,
+                self.split_model,
+                self.split_max_tensors,
+                self.split_max_size,
+                self.repo_name,
+                self.gguf_name,
+            ]
+        )
+        self.submit_btn = gr.Button(
+            value="Submit",
+            variant="primary"
+        )
+
+        #####
+        # Outputs section
+        #####
+        self.output_label = gr.Markdown(label="output")
+        self.output_image = gr.Image(
+            show_label=False,
+            show_download_button=False,
+            interactive=False
+        )
+
+    @staticmethod
+    def _update_output_repo(model_id: str, oauth_token: Optional[gr.OAuthToken]) -> str:
+        """Update output repository name based on model and user."""
+        if oauth_token is None or not oauth_token.token:
+            return ""
+        if not model_id:
+            return ""
+        try:
+            username = whoami(oauth_token.token)["name"]
+            model_name = model_id.split('/')[-1]
+            return f"{username}/{model_name}-GGUF"
+        except:
+            return ""
+
+    @staticmethod
+    def _update_output_filename(model_id: str, use_imatrix: bool, q_method: str, imatrix_q_method: str) -> str:
+        """Update output filename based on model and quantization settings."""
+        if not model_id:
+            return ""
+        model_name = model_id.split('/')[-1]
+        if use_imatrix:
+            return f"{model_name}-{imatrix_q_method.upper()}-imat.gguf"
+        return f"{model_name}-{q_method.upper()}.gguf"
+
+    def _setup_interface(self):
+        """Set up the Gradio interface."""
+        with gr.Blocks(css=self.css) as self.demo:
+            #####
+            # Layout
+            #####
+            gr.Markdown(HuggingFaceModelProcessor.ERROR_LOGIN)
+            gr.LoginButton(min_width=250)
+            gr.HTML("<h1 style=\"text-aling:center;\">Create your own GGUF Quants!</h1>")
+            gr.Markdown(f"The space takes an HF repo as an input, quantizes it and creates a Public repo containing the selected quant under your HF user namespace.<br/>Use via {self.processor.SPACE_URL}")
+
+            with gr.Row():
+                with gr.Column() as inputs:
+                    gr.Markdown("### Model Configuration")
+                    self.model_id.render()
+                    with gr.Column():
+                        self.use_imatrix.render()
+                        self.q_method.render()
+                        self.imatrix_q_method.render()
+                        self.train_data_file.render()
+                    gr.Markdown("### Advanced Options")
+                    self.quant_embedding.render()
+                    self.embedding_tensor_method.render()
+                    self.leave_output.render()
+                    self.quant_output.render()
+                    self.output_tensor_method.render()
+                    self.split_model.render()
+                    with gr.Row() as split_options:
+                        self.split_max_tensors.render()
+                        self.split_max_size.render()
+                    gr.Markdown("### Output Settings")
+                    gr.Markdown("You can customize settings for your GGUF repo.")
+                    self.private_repo.render()
+                    with gr.Row():
+                        self.repo_name.render()
+                        self.gguf_name.render()
+                    # Buttons
+                    with gr.Row() as buttons:
+                        self.clear_btn.render()
+                        self.submit_btn.render()
+                with gr.Column() as outputs:
+                    self.output_label.render()
+                    self.output_image.render()
+
+            #####
+            # Event handlers
+            #####
+            self.submit_btn.click(
+                fn=self._process_model_wrapper,
+                inputs=[
+                    self.model_id,
+                    self.q_method,
+                    self.use_imatrix,
+                    self.imatrix_q_method,
+                    self.private_repo,
+                    self.train_data_file,
+                    self.repo_name,
+                    self.gguf_name,
+                    self.quant_embedding,
+                    self.embedding_tensor_method,
+                    self.leave_output,
+                    self.quant_output,
+                    self.output_tensor_method,
+                    self.split_model,
+                    self.split_max_tensors,
+                    self.split_max_size
+                ],
+                outputs=[
+                    self.output_label,
+                    self.output_image,
+                ],
             )
-            readme_path = Path(outdir)/"README.md"
-            card.save(readme_path)
 
-            if split_model:
-                split_upload_model(str(quantized_gguf_path), outdir, new_repo_id, oauth_token, split_max_tensors, split_max_size)
-            else:
-                try:
-                    print(f"Uploading quantized model: {quantized_gguf_path}")
-                    api.upload_file(
-                        path_or_fileobj=quantized_gguf_path,
-                        path_in_repo=quantized_gguf_name,
-                        repo_id=new_repo_id,
-                    )
-                except Exception as e:
-                    raise Exception(f"Error uploading quantized model: {e}")
-            
-            if os.path.isfile(imatrix_path):
-                try:
-                    print(f"Uploading imatrix.dat: {imatrix_path}")
-                    api.upload_file(
-                        path_or_fileobj=imatrix_path,
-                        path_in_repo="imatrix.dat",
-                        repo_id=new_repo_id,
-                    )
-                except Exception as e:
-                    raise Exception(f"Error uploading imatrix.dat: {e}")
+            #####
+            # OnChange handlers
+            #####
+            self.use_imatrix.change(
+                fn=lambda use_imatrix: [gr.update(visible=not use_imatrix), gr.update(visible=use_imatrix), gr.update(visible=use_imatrix)],
+                inputs=self.use_imatrix,
+                outputs=[self.q_method, self.imatrix_q_method, self.train_data_file]
+            )
+            self.split_model.change(
+                fn=lambda split_model: [gr.update(visible=split_model), gr.update(visible=split_model)],
+                inputs=self.split_model,
+                outputs=[self.split_max_tensors, self.split_max_size]
+            )
+            self.quant_embedding.change(
+                fn=lambda quant_embedding: gr.update(visible=quant_embedding),
+                inputs=self.quant_embedding,
+                outputs=[self.embedding_tensor_method]
+            )
+            self.leave_output.change(
+                fn=lambda leave_output, quant_output: [gr.update(visible=not leave_output), gr.update(visible=not leave_output and quant_output)],
+                inputs=[self.leave_output, self.leave_output],
+                outputs=[self.quant_output, self.output_tensor_method]
+            )
+            self.quant_output.change(
+                fn=lambda quant_output: [gr.update(visible=not quant_output), gr.update(visible=quant_output)],
+                inputs=self.quant_output,
+                outputs=[self.leave_output, self.output_tensor_method]
+            )
+            self.model_id.change(
+                fn=self._update_output_repo,
+                inputs=[self.model_id],
+                outputs=[self.repo_name]
+            )
+            self.model_id.change(
+                fn=self._update_output_filename,
+                inputs=[self.model_id, self.use_imatrix, self.q_method, self.imatrix_q_method],
+                outputs=[self.gguf_name]
+            )
+            self.use_imatrix.change(
+                fn=self._update_output_filename,
+                inputs=[self.model_id, self.use_imatrix, self.q_method, self.imatrix_q_method],
+                outputs=[self.gguf_name]
+            )
+            self.q_method.change(
+                fn=self._update_output_filename,
+                inputs=[self.model_id, self.use_imatrix, self.q_method, self.imatrix_q_method],
+                outputs=[self.gguf_name]
+            )
+            self.imatrix_q_method.change(
+                fn=self._update_output_filename,
+                inputs=[self.model_id, self.use_imatrix, self.q_method, self.imatrix_q_method],
+                outputs=[self.gguf_name]
+            )
 
-            api.upload_file(
-                path_or_fileobj=readme_path,
-                path_in_repo="README.md",
-                repo_id=new_repo_id,
+    def _process_model_wrapper(self, model_id: str, q_method: str, use_imatrix: bool,
+                              imatrix_q_method: str, private_repo: bool, train_data_file,
+                              repo_name: str, gguf_name: str, quant_embedding: bool,
+                              embedding_tensor_method: str, leave_output: bool,
+                              quant_output: bool, output_tensor_method: str,
+                              split_model: bool, split_max_tensors, split_max_size: str, oauth_token: Optional[gr.OAuthToken]) -> Tuple[str, str]:
+        """Wrapper for the process_model method to handle the conversion using ModelProcessingConfig."""
+        try:
+            # Validate token and get token string
+            token = self.processor._validate_token(oauth_token)
+
+            # Create configuration objects
+            quant_config = QuantizationConfig(
+                method=q_method,
+                use_imatrix=use_imatrix,
+                imatrix_method=imatrix_q_method,
+                quant_embedding=quant_embedding,
+                embedding_tensor_method=embedding_tensor_method,
+                leave_output=leave_output,
+                quant_output=quant_output,
+                output_tensor_method=output_tensor_method
             )
-            print(f"Uploaded successfully with {imatrix_q_method if use_imatrix else q_method} option!")
 
-        # end of the TemporaryDirectory(dir="outputs") block; temporary outputs are deleted here
+            split_config = SplitConfig(
+                enabled=split_model,
+                max_tensors=split_max_tensors if isinstance(split_max_tensors, int) else 256,
+                max_size=split_max_size
+            )
 
-        return (
-            f'<h1>✅ DONE</h1><br/>Find your repo here: <a href="{new_repo_url}" target="_blank" style="text-decoration:underline">{new_repo_id}</a>',
-            "llama.png",
-        )
-    except Exception as e:
-        return (f'<h1>❌ ERROR</h1><br/><pre style="white-space:pre-wrap;">{escape(str(e))}</pre>', "error.png")
-
-
-css="""/* Custom CSS to allow scrolling */
-.gradio-container {overflow-y: auto;}
-"""
-model_id = HuggingfaceHubSearch(
-    label="Hub Model ID",
-    placeholder="Search for model id on Huggingface",
-    search_type="model",
-)
-
-q_method = gr.Dropdown(
-    ["Q2_K", "Q3_K_S", "Q3_K_M", "Q3_K_L", "Q4_0", "Q4_K_S", "Q4_K_M", "Q5_0", "Q5_K_S", "Q5_K_M", "Q6_K", "Q8_0"],
-    label="Quantization Method",
-    info="GGML quantization type",
-    value="Q4_K_M",
-    filterable=False,
-    visible=True
-)
-
-imatrix_q_method = gr.Dropdown(
-    ["IQ3_M", "IQ3_XXS", "Q4_K_M", "Q4_K_S", "IQ4_NL", "IQ4_XS", "Q5_K_M", "Q5_K_S"],
-    label="Imatrix Quantization Method",
-    info="GGML imatrix quants type",
-    value="IQ4_NL", 
-    filterable=False,
-    visible=False
-)
-
-use_imatrix = gr.Checkbox(
-    value=False,
-    label="Use Imatrix Quantization",
-    info="Use importance matrix for quantization."
-)
-
-private_repo = gr.Checkbox(
-    value=False,
-    label="Private Repo",
-    info="Create a private repo under your username."
-)
-
-train_data_file = gr.File(
-    label="Training Data File",
-    file_types=["txt"],
-    visible=False
-)
-
-split_model = gr.Checkbox(
-    value=False,
-    label="Split Model",
-    info="Shard the model using gguf-split."
-)
-
-split_max_tensors = gr.Number(
-    value=256,
-    label="Max Tensors per File",
-    info="Maximum number of tensors per file when splitting model.",
-    visible=False
-)
-
-split_max_size = gr.Textbox(
-    label="Max File Size",
-    info="Maximum file size when splitting model (--split-max-size). May leave empty to use the default. Accepted suffixes: M, G. Example: 256M, 5G",
-    visible=False
-)
-
-iface = gr.Interface(
-        fn=process_model,
-        inputs=[
-            model_id,
-            q_method,
-            use_imatrix,
-            imatrix_q_method,
-            private_repo,
-            train_data_file,
-            split_model,
-            split_max_tensors,
-            split_max_size,
-        ],
-        outputs=[
-            gr.Markdown(label="output"),
-            gr.Image(show_label=False),
-        ],
-        title="Create your own GGUF Quants, blazingly fast ⚡!",
-        description="The space takes an HF repo as an input, quantizes it and creates a Public repo containing the selected quant under your HF user namespace.",
-        api_name=False
-    )
-
-# Create Gradio interface
-with gr.Blocks(css=css) as demo: 
-    gr.Markdown("You must be logged in to use GGUF-my-repo.")
-    gr.LoginButton(min_width=250)
-
-    iface.render()
-
-    def update_split_visibility(split_model):
-        return gr.update(visible=split_model), gr.update(visible=split_model)
-
-    split_model.change(
-        fn=update_split_visibility,
-        inputs=split_model,
-        outputs=[split_max_tensors, split_max_size]
-    )
-
-    def update_visibility(use_imatrix):
-        return gr.update(visible=not use_imatrix), gr.update(visible=use_imatrix), gr.update(visible=use_imatrix)
-    
-    use_imatrix.change(
-        fn=update_visibility,
-        inputs=use_imatrix,
-        outputs=[q_method, imatrix_q_method, train_data_file]
-    )
-
-def restart_space():
-    HfApi().restart_space(repo_id="ggml-org/gguf-my-repo", token=HF_TOKEN, factory_reboot=True)
-
-scheduler = BackgroundScheduler()
-scheduler.add_job(restart_space, "interval", seconds=21600)
-scheduler.start()
-
-# Launch the interface
-demo.queue(default_concurrency_limit=1, max_size=5).launch(debug=True, show_api=False)
+            output_config = OutputConfig(
+                private_repo=private_repo,
+                repo_name=repo_name,
+                filename=gguf_name
+            )
+
+            model_name = self.processor._get_model_name(model_id)
+
+            with tempfile.TemporaryDirectory(dir=self.processor.OUTPUT_FOLDER) as outDirObj:
+                outdir = (
+                    self.processor._create_folder(os.path.join(self.processor.OUTPUT_FOLDER, model_name))
+                    if self.processor.RUN_LOCALLY == "1"
+                    else Path(outDirObj)
+                )
+
+                quant_config.fp16_model = f"{outdir}/{model_name}-fp16.gguf"
+                quant_config.imatrix_file = f"{outdir}/{model_name}-imatrix.gguf"
+                quant_config.quantized_gguf = f"{outdir}/{gguf_name}"
+
+                processing_config = ModelProcessingConfig(
+                    token=token,
+                    model_id=model_id,
+                    model_name=model_name,
+                    outdir=outdir,
+                    quant_config=quant_config,
+                    split_config=split_config,
+                    output_config=output_config
+                )
+
+                # Call the processor's main method with the config object
+                self.processor.process_model(processing_config)
+
+                return (
+                    f'<h1>✅ DONE</h1><br/>Find your repo here: <a href="{processing_config.new_repo_url}" target="_blank" style="text-decoration:underline">{processing_config.new_repo_id}</a>',
+                    "llama.png",
+                )
+
+        except Exception as e:
+            print(f"Error processing model: {e}")
+            return (f'<h1>❌ ERROR</h1><br/><pre style="white-space:pre-wrap;">{self.processor._escape_html(str(e))}</pre>', "error.png")
+
+
+    def launch(self):
+        """Launch the Gradio interface."""
+        # Set up space restart scheduler
+        def restart_space():
+            HfApi().restart_space(repo_id=self.processor.SPACE_ID, token=self.processor.HF_TOKEN, factory_reboot=True)
+
+        scheduler = BackgroundScheduler()
+        scheduler.add_job(restart_space, "interval", seconds=21600)
+        scheduler.start()
+
+        # Launch the interface
+        self.demo.queue(default_concurrency_limit=1, max_size=5).launch(debug=True, show_api=False)
+
+
+# Main execution
+if __name__ == "__main__":
+    ui = GGUFConverterUI()
+    ui.launch()