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,
    "attente_modificateurs_s": 2.0,  # on patiente avant de coller (cf inject_text)
    "delai_presse_papier_s": 0.12,   # 0.05 s etait juste : le presse-papier n'etait pas pret
    "max_duration_s": 120.0,     # sécurité : arrêt auto après 2 min (fenêtre jamais figée)
    "sample_rate": 16000,
    "fps": 30,
    # --- overlay « barre de dictee » (bas de l'ecran) ---
    "bar_width": 330.0,
    "bar_height": 40.0,
    "bar_radius": 12.0,
    "bar_bg_rgba": (0.09, 0.09, 0.10, 0.95),
    "stop_button_rgb": (0.89, 0.31, 0.23),   # rond rouge-orange
    "stop_glyph_rgb": (1.0, 1.0, 1.0),       # carre arrondi blanc au centre
    "waveform_rgb": (0.95, 0.95, 0.97),
    "waveform_bars": 28,          # nombre de barres de la forme d'onde
    "waveform_floor": 0.07,       # plancher : dessine en petit point (traine pointillee)
    "mic_rgb": (0.78, 0.78, 0.82),
    "bubble_text_rgb": (1.0, 1.0, 1.0),
    "bubble_font_size": 12.0,
    "bubble_max_width": 240.0,
    "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)
    "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) + "."


# Proportions internes de la barre, exprimees en fraction de sa HAUTEUR : reduire
# `bar_height` reduit alors le bouton, le micro et les marges dans le meme
# mouvement, au lieu de laisser une grosse pastille dans une barre retrecie.
_PROPORTIONS = {
    "stop_diametre": 0.538,
    "mic_largeur": 0.173,
    "marge_laterale": 0.423,
    "mic_corps_hauteur": 0.250,
    "mic_arceau_rayon": 0.135,
    "ecart_interne": 0.346,
}


def attendre(condition, delai_max_s, pas_s, dormir=None, horloge=None):
    """Attend que `condition()` soit vraie, au plus `delai_max_s`. Rend True/False.

    Bornee volontairement : mieux vaut coller avec un modificateur encore enfonce
    que perdre la transcription si une touche reste bloquee. `dormir` et `horloge`
    sont injectables pour tester sans attendre reellement.
    """
    import time as _t
    dormir = dormir or _t.sleep
    horloge = horloge or _t.monotonic
    debut = horloge()
    while True:
        if condition():
            return True
        if horloge() - debut >= delai_max_s:
            return False
        dormir(pas_s)


def bar_metrics(bar_height):
    """Mesures de l'overlay pour une hauteur de barre donnee, en points."""
    if bar_height <= 0:
        raise ValueError("bar_height doit etre > 0")
    return {cle: bar_height * f for cle, f in _PROPORTIONS.items()}


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()