File size: 5,430 Bytes
9337185 624b770 e441a1a 0a56987 624b770 494af84 8fb4e5c 494af84 9337185 624b770 d744966 624b770 9337185 59519b7 9337185 624b770 d841847 624b770 d841847 624b770 d841847 624b770 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 |
# utils.py
import opencc
import os
from pathlib import Path
import sys
from typing import Optional
import multiprocessing
# Detect logical cores (vCPUs available to the container)
# On HF Spaces free tier, cpu_count() reports 16 but only 2 are actually available
detected_cpus = multiprocessing.cpu_count()
if os.environ.get('SPACE_ID'):
# HF Spaces free tier limitation
num_vcpus = min(detected_cpus, 2)
else:
num_vcpus = detected_cpus
model_names = {
"tiny English":"tiny",
"tiny Arabic":"tiny-ar",
"tiny Chinese":"tiny-zh",
"tiny Japanese":"tiny-ja",
"tiny Korean":"tiny-ko",
"tiny Ukrainian":"tiny-uk",
"tiny Vietnamese":"tiny-vi",
"base English":"base",
"base Spanish":"base-es"
}
# Using only the two specified sherpa-onnx models from Hugging Face
sensevoice_models = {
"SenseVoice Small (2024)": "csukuangfj/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17",
"SenseVoice Small (2025 int8)": "csukuangfj/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-int8-2025-09-09",
}
available_gguf_llms = {
"Gemma-3-1B": ("bartowski/google_gemma-3-1b-it-qat-GGUF", "google_gemma-3-1b-it-qat-Q4_0.gguf"),
"Gemma-3-270M": ("bartowski/google_gemma-3-270m-it-qat-GGUF", "google_gemma-3-270m-it-qat-Q8_0.gguf"),
"Gemma-3-3N-E2B": ("unsloth/gemma-3n-E2B-it-GGUF", "gemma-3n-E2B-it-Q4_0.gguf"),
"Gemma-3-3N-E4B": ("unsloth/gemma-3n-E4B-it-GGUF", "gemma-3n-E4B-it-Q4_0.gguf"),
}
s2tw_converter = opencc.OpenCC('s2twp')
def get_writable_model_dir():
"""Get appropriate model directory for HF Spaces"""
# Check for HF Spaces environment
if os.environ.get('SPACE_ID'):
# Use HF Spaces cache directory
cache_dir = Path('/tmp/models')
else:
# Use standard cache directory
cache_dir = Path.home() / ".cache" / "speech_assistant" / "models"
# Ensure directory exists
cache_dir.mkdir(parents=True, exist_ok=True)
return cache_dir
def download_sensevoice_model(model_name: str) -> Path:
"""Download SenseVoice model from Hugging Face using official tools"""
try:
from huggingface_hub import snapshot_download
from huggingface_hub.utils import HFValidationError
except ImportError:
raise ImportError("Please install huggingface_hub: pip install huggingface_hub")
# Use model_name directly as repo_id
repo_id = model_name
model_cache_dir = get_writable_model_dir()
local_dir = model_cache_dir / model_name.replace("/", "--")
# Check if model already exists
model_file = "model.int8.onnx" if "int8" in model_name else "model.onnx"
model_file_path = local_dir / model_file
tokens_file_path = local_dir / "tokens.txt"
if model_file_path.exists() and tokens_file_path.exists():
print(f"Model {model_name} already exists, skipping download")
return local_dir
# Remove existing incomplete model directory
if local_dir.exists():
import shutil
print(f"Removing incomplete model directory: {local_dir}")
shutil.rmtree(local_dir)
print(f"Downloading {model_name} from Hugging Face")
print("This may take several minutes depending on your connection...")
try:
# Use HF's snapshot_download for reliable download
snapshot_download(
repo_id=repo_id,
local_dir=str(local_dir),
resume_download=True, # Resume if interrupted
max_workers=4, # Parallel downloads
)
print(f"Model {model_name} downloaded successfully!")
return local_dir
except HFValidationError as e:
print(f"Hugging Face validation error: {e}")
raise
except Exception as e:
print(f"Download failed: {str(e)}")
# Clean up partial download
if local_dir.exists():
import shutil
shutil.rmtree(local_dir)
raise e
def load_sensevoice_model(model_name: str = "csukuangfj/sherpa-onnx-sense-voice-zh-en-ja-ko-yue-2024-07-17"):
"""Load SenseVoice ONNX model from Hugging Face"""
try:
# Try to import sherpa-onnx
import sherpa_onnx
print(f"Loading model: {model_name}")
# Download model if not exists
model_path = download_sensevoice_model(model_name)
# Determine which model file to use
model_file = "model.int8.onnx" if "int8" in model_name else "model.onnx"
model_file_path = model_path / model_file
print("Initializing recognizer...")
# Initialize recognizer with proper settings
recognizer = sherpa_onnx.OfflineRecognizer.from_sense_voice(
model=str(model_file_path),
tokens=str(model_path / "tokens.txt"),
use_itn=True, # Enable inverse text normalization
language="auto" # Auto-detect language
)
print("Model loaded successfully!")
return recognizer
except Exception as e:
print(f"Failed to load SenseVoice model: {e}")
# Try to force redownload on next attempt
model_cache_dir = get_writable_model_dir()
model_dir = model_cache_dir / model_name.replace("/", "--")
if model_dir.exists():
import shutil
print(f"Removing model directory for redownload: {model_dir}")
shutil.rmtree(model_dir)
raise e |