zonza/core.py

"""Unités pures de Zonza — aucune dépendance matérielle.

Tout ce qui est ici est testable sans micro, sans clavier, sans modèle.
"""

import fcntl
import os
import re

CONFIG = {
    "model": "mlx-community/whisper-medium",
    "language": "fr",
    "hotkey": "<cmd>+<ctrl>+d",   # Cmd+Ctrl+D (libre : ni macOS ni VS Code ne l'utilisent)
    "sounds": True,
    "min_duration_s": 0.3,
    "max_duration_s": 120.0,     # sécurité : arrêt auto après 2 min (fenêtre jamais figée)
    "sample_rate": 16000,
    "fps": 30,
    "circle_base_radius": 20.0,
    "circle_amplitude": 30.0,
    "circle_color_rgb": (0.40, 0.50, 0.95),
    "level_gain": 8.0,        # amplification du niveau brut pour l'animation
    "level_attack": 0.6,      # lissage à la montée (0–1, proche de 1 = très réactif)
    "level_release": 0.15,    # lissage à la descente (plus petit = retombée douce)
    "wave_rings": 3,          # nombre d'anneaux d'onde concentriques
    "wave_speed": 0.04,       # vitesse de propagation des anneaux par frame
    "initial_prompt": None,   # rempli plus bas depuis VOCABULAIRE
}


# --- vocabulaire metier -------------------------------------------------
# Whisper accepte une « amorce » (initial_prompt) qui conditionne le decodage.
# Sans elle, le modele rendait « Playwright » en « PlayWreck », « iCloud » en
# « Auriclaw », « Supabase » en « Supabass », « Vercel » en « Versell »
# (mesure du 2026-08-21 : 4 termes sur 5 faux ; 5 sur 5 corrects avec l'amorce).
# Passer a large-v3-turbo a ete teste et fait PIRE (« commit sur Gitea » ->
# « commis sur JTA ») : c'est bien l'amorce qui paie, pas la taille du modele.
#
# Pour ajouter tes propres termes : complete cette liste, puis ./build_app.sh.
VOCABULAIRE = [
    # agents et terminal
    "Claude Code", "Codex", "OpenCode", "Warp", "Playwright", "MCP",
    # web et hebergement
    "Next.js", "TypeScript", "Vercel", "Convex", "Supabase", "Docker", "Caddy",
    "Gitea", "GitHub", "Hetzner", "Hostinger", "iCloud", "Clerk",
    # paiement, mail, automatisation
    "Stripe", "PayPal", "Shopify", "Brevo", "n8n", "Airtable", "Yabetoo",
    # projets
    "FacturPro", "Forescale", "Reflow", "CiteGain", "Faramaya", "Soko",
    "Moubote", "Yelessa", "Zonza", "MR TECH LAB",
    # jargon courant
    "repo", "commit", "webhook", "API", "npm", "build", "deploy", "conteneur",
    "migration", "endpoint", "prompt",
]

# Whisper ne lit que ~224 tokens d'amorce et tronque le reste EN SILENCE.
# On borne donc explicitement, pour qu'un ajout de termes ne soit jamais
# silencieusement sans effet.
PROMPT_MAX_CARACTERES = 800


def build_initial_prompt(termes=None):
    """Construit l'amorce Whisper a partir d'une liste de termes.

    Doublons retires (ordre conserve), une seule ligne, longueur bornee a
    PROMPT_MAX_CARACTERES : au-dela, les derniers termes sont ecartes plutot
    que de laisser Whisper tronquer au milieu d'un mot.
    """
    if termes is None:
        termes = VOCABULAIRE
    vus, uniques = set(), []
    for t in termes:
        if t not in vus:
            vus.add(t)
            uniques.append(t)
    prefixe = "Vocabulaire : "
    retenus = []
    for t in uniques:
        candidat = prefixe + ", ".join(retenus + [t]) + "."
        if len(candidat) > PROMPT_MAX_CARACTERES:
            break
        retenus.append(t)
    return prefixe + ", ".join(retenus) + "."


def validate_config(config):
    """Valide la config ; lève ValueError si une valeur est aberrante."""
    if config["min_duration_s"] < 0:
        raise ValueError("min_duration_s doit être >= 0")
    if config["sample_rate"] <= 0:
        raise ValueError("sample_rate doit être > 0")
    if not config["hotkey"]:
        raise ValueError("hotkey ne peut pas être vide")


def is_long_enough(num_samples, sample_rate, min_duration_s):
    """Renvoie True si l'enregistrement dépasse le seuil de durée minimal."""
    duration_s = num_samples / sample_rate
    return duration_s >= min_duration_s


def clean_transcript(text):
    """Nettoie le texte de Whisper : trim + espaces/retours multiples écrasés."""
    return re.sub(r"\s+", " ", text).strip()


LOCK_PATH = os.path.expanduser("~/Library/Application Support/Zonza/instance.lock")


def acquire_single_instance_lock(path=LOCK_PATH):
    """Prend un verrou exclusif non bloquant ; renvoie le fichier, ou None si une
    autre instance le detient deja.

    Pourquoi : deux bundles Zonza de meme identifiant (dont un fantome reste dans
    la Corbeille mais toujours enregistre aupres de LaunchServices) pouvaient
    tourner en meme temps. Chacun chargeait son propre modele MLX-Whisper et
    captait la MEME hotkey globale : un appui, deux transcriptions, machine
    saturee. Purger le fantome corrige l'incident ; ce verrou corrige la classe.

    Le handle doit rester vivant tant que l'app tourne : le verrou tombe a la
    fermeture du fichier, et le noyau le libere si le processus meurt brutalement
    (donc pas de verrou orphelin apres un plantage).
    """
    parent = os.path.dirname(path)
    if parent:
        os.makedirs(parent, exist_ok=True)
    handle = open(path, "a")
    try:
        fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB)
    except OSError:
        handle.close()
        return None
    return handle


CONFIG["initial_prompt"] = build_initial_prompt()