cf-voice/cf_voice/prefs.py

# cf_voice/prefs.py — user preference hooks for cf-core preferences module
#
# MIT licensed. Provides voice-specific preference keys and helpers.
#
# When circuitforge_core is installed, reads/writes from the shared preference
# store (LocalFileStore or cloud backend). When it is not installed (standalone
# cf-voice use), falls back to environment variables only.
#
# Preference paths use dot-separated notation (cf-core convention):
#   "voice.elcor_mode"           bool   — Elcor-style tone annotations
#   "voice.confidence_threshold" float  — minimum confidence to emit a frame
#   "voice.whisper_model"        str    — faster-whisper model size
#   "voice.elcor_prior_frames"   int    — rolling context window for Elcor LLM
from __future__ import annotations

import logging
import os
from typing import Any

logger = logging.getLogger(__name__)

# ── Preference key constants ──────────────────────────────────────────────────

PREF_ELCOR_MODE = "voice.elcor_mode"
PREF_CONFIDENCE_THRESHOLD = "voice.confidence_threshold"
PREF_WHISPER_MODEL = "voice.whisper_model"
PREF_ELCOR_PRIOR_FRAMES = "voice.elcor_prior_frames"

# Defaults used when neither preference store nor environment has a value
_DEFAULTS: dict[str, Any] = {
    PREF_ELCOR_MODE: False,
    PREF_CONFIDENCE_THRESHOLD: 0.55,
    PREF_WHISPER_MODEL: "small",
    PREF_ELCOR_PRIOR_FRAMES: 4,
}

# ── Environment variable fallbacks ────────────────────────────────────────────

_ENV_KEYS: dict[str, str] = {
    PREF_ELCOR_MODE: "CF_VOICE_ELCOR",
    PREF_CONFIDENCE_THRESHOLD: "CF_VOICE_CONFIDENCE_THRESHOLD",
    PREF_WHISPER_MODEL: "CF_VOICE_WHISPER_MODEL",
    PREF_ELCOR_PRIOR_FRAMES: "CF_VOICE_ELCOR_PRIOR_FRAMES",
}

_COERCE: dict[str, type] = {
    PREF_ELCOR_MODE: bool,
    PREF_CONFIDENCE_THRESHOLD: float,
    PREF_WHISPER_MODEL: str,
    PREF_ELCOR_PRIOR_FRAMES: int,
}


def _from_env(pref_path: str) -> Any:
    """Read a preference from its environment variable fallback."""
    env_key = _ENV_KEYS.get(pref_path)
    if env_key is None:
        return None
    raw = os.environ.get(env_key)
    if raw is None:
        return None
    coerce = _COERCE.get(pref_path, str)
    try:
        if coerce is bool:
            return raw.strip().lower() in ("1", "true", "yes")
        return coerce(raw)
    except (ValueError, TypeError):
        logger.warning("prefs: could not parse env %s=%r as %s", env_key, raw, coerce)
        return None


def _cf_core_store():
    """Return the cf-core default preference store, or None if not available."""
    try:
        from circuitforge_core.preferences import store as _store_mod
        return _store_mod._DEFAULT_STORE
    except ImportError:
        return None


# ── Public API ────────────────────────────────────────────────────────────────


def get_voice_pref(
    pref_path: str,
    user_id: str | None = None,
    store=None,
) -> Any:
    """
    Read a voice preference value.

    Resolution order:
      1. Explicit store (passed in by caller — used for testing or cloud backends)
      2. cf-core LocalFileStore (if circuitforge_core is installed)
      3. Environment variable fallback
      4. Built-in default

    pref_path  One of the PREF_* constants, e.g. PREF_ELCOR_MODE.
    user_id    Passed to the store for cloud backends; local store ignores it.
    """
    # 1. Explicit store
    if store is not None:
        val = store.get(user_id=user_id, path=pref_path, default=None)
        if val is not None:
            return val

    # 2. cf-core default store
    cf_store = _cf_core_store()
    if cf_store is not None:
        val = cf_store.get(user_id=user_id, path=pref_path, default=None)
        if val is not None:
            return val

    # 3. Environment variable
    env_val = _from_env(pref_path)
    if env_val is not None:
        return env_val

    # 4. Built-in default
    return _DEFAULTS.get(pref_path)


def set_voice_pref(
    pref_path: str,
    value: Any,
    user_id: str | None = None,
    store=None,
) -> None:
    """
    Write a voice preference value.

    Writes to the explicit store if provided, otherwise to the cf-core default
    store. Raises RuntimeError if neither is available (env-only mode has no
    writable persistence).
    """
    target = store or _cf_core_store()
    if target is None:
        raise RuntimeError(
            "No writable preference store available. "
            "Install circuitforge_core or pass a store explicitly."
        )
    target.set(user_id=user_id, path=pref_path, value=value)


def is_elcor_enabled(user_id: str | None = None, store=None) -> bool:
    """
    Convenience: return True if the user has Elcor annotation mode enabled.

    Elcor mode switches tone subtext from generic format ("Tone: Frustrated")
    to the Mass Effect Elcor prefix format ("With barely concealed frustration:").
    It is an accessibility feature for autistic and ND users who benefit from
    explicit tonal annotation. Opt-in, local-only — no data leaves the device.

    Defaults to False.
    """
    return bool(get_voice_pref(PREF_ELCOR_MODE, user_id=user_id, store=store))


def get_confidence_threshold(user_id: str | None = None, store=None) -> float:
    """Return the minimum confidence threshold for emitting VoiceFrames (0.0–1.0)."""
    return float(
        get_voice_pref(PREF_CONFIDENCE_THRESHOLD, user_id=user_id, store=store)
    )


def get_whisper_model(user_id: str | None = None, store=None) -> str:
    """Return the faster-whisper model name to use (e.g. "small", "medium")."""
    return str(get_voice_pref(PREF_WHISPER_MODEL, user_id=user_id, store=store))


def get_elcor_prior_frames(user_id: str | None = None, store=None) -> int:
    """
    Return the number of prior VoiceFrames to include as context for Elcor
    label generation. Larger windows produce more contextually aware annotations
    but increase LLM prompt length and latency.

    Default: 4 frames (~8–10 seconds of rolling context at 2s intervals).
    """
    return int(
        get_voice_pref(PREF_ELCOR_PRIOR_FRAMES, user_id=user_id, store=store)
    )