feat: add multimodal engagement estimation and adaptive tutoring

ai/engagement/__init__.py

@@ -0,0 +1,12 @@
+"""Engagement domain — multimodal engagement analysis (text / audio / video).
+
+Pure compute lives in ``text_core`` / ``audio_core`` / ``video_core`` /
+``fusion``; ``cache`` holds short-lived runtime signals; ``service`` and
+``repository`` provide the business-logic and persistence layers used by the
+``/engagement`` HTTP router.
+"""
+
+from .service import EngagementService
+from .repository import EngagementRepository
+
+__all__ = ["EngagementService", "EngagementRepository"]

ai/engagement/audio_core.py

@@ -0,0 +1,140 @@
+import base64
+import logging
+import io
+from typing import Optional
+
+logger = logging.getLogger("audio_engagement")
+
+# Optional heavy deps — import lazily and tolerate absence so the rest of
+# the app can run without audio libraries installed.
+NUMPY_AVAILABLE = False
+LIBROSA_AVAILABLE = False
+try:
+    import numpy as np
+
+    NUMPY_AVAILABLE = True
+except Exception:
+    np = None
+    NUMPY_AVAILABLE = False
+
+try:
+    import librosa
+    import soundfile as sf
+
+    LIBROSA_AVAILABLE = True
+    logger.debug("librosa and soundfile loaded successfully.")
+except Exception:
+    logger.debug("librosa or soundfile missing — audio scoring disabled.")
+
+# Browsers (Chromium) record WebM/Opus by default, which libsndfile/soundfile
+# cannot decode. PyAV (already pulled in by faster-whisper) decodes it via
+# ffmpeg, so we use it as a fallback when soundfile fails.
+AV_AVAILABLE = False
+try:
+    import av
+
+    AV_AVAILABLE = True
+except Exception:
+    av = None
+
+
+def _decode_audio(audio_bytes: bytes, sr: int):
+    """Decode arbitrary audio bytes to a mono float32 array at ``sr`` Hz.
+
+    Tries soundfile first (WAV/FLAC/OGG), then falls back to PyAV for formats
+    libsndfile cannot read (notably WebM/Opus from browser MediaRecorder).
+    """
+    # Fast path: soundfile (handles WAV/FLAC/OGG-Vorbis).
+    try:
+        y, native_sr = sf.read(io.BytesIO(audio_bytes), dtype="float32")
+        if y.ndim > 1:
+            y = librosa.to_mono(y.T)
+        if native_sr != sr:
+            y = librosa.resample(y, orig_sr=native_sr, target_sr=sr)
+        return y
+    except Exception:
+        pass  # likely WebM/Opus — try PyAV below
+
+    if not AV_AVAILABLE:
+        raise RuntimeError("soundfile failed and PyAV not available")
+
+    container = av.open(io.BytesIO(audio_bytes))
+    resampler = av.AudioResampler(format="flt", layout="mono", rate=sr)
+    chunks = []
+    for frame in container.decode(audio=0):
+        for rframe in resampler.resample(frame):
+            chunks.append(rframe.to_ndarray().reshape(-1))
+    container.close()
+    if not chunks:
+        return np.array([], dtype="float32")
+    return np.concatenate(chunks).astype("float32")
+
+
+def compute_audio_score(audio_base64: str, sr: int = 16000) -> Optional[float]:
+    # If required DSP libs are missing, gracefully skip audio scoring.
+    if not (LIBROSA_AVAILABLE and NUMPY_AVAILABLE):
+        logger.debug("[Audio] Skipping compute: librosa/numpy not available.")
+        return None
+
+    try:
+        try:
+            audio_bytes = base64.b64decode(audio_base64)
+            y = _decode_audio(audio_bytes, sr)
+        except Exception as e:
+            logger.warning(f"[Audio] Decoding error: {e}")
+            return None
+
+        rms = librosa.feature.rms(y=y)[0]
+        avg_rms = np.mean(rms)
+
+        if len(y) < sr * 0.2 or avg_rms < 0.001:
+            return None
+
+        pitches, magnitudes = librosa.piptrack(
+            y=y, sr=sr, n_fft=1024, hop_length=256, fmin=80.0, fmax=400.0
+        )
+
+        voiced_mask = magnitudes > np.median(magnitudes)
+        voiced_pitches = pitches[voiced_mask]
+        voiced_pitches = voiced_pitches[voiced_pitches > 0]
+
+        if len(voiced_pitches) > 2:
+            pitch_std = np.std(voiced_pitches)
+            pitch_score = min(1.0, pitch_std / 60.0)
+        else:
+            pitch_score = 0.3
+
+        energy_score = min(1.0, max(0.0, avg_rms / 0.04))
+
+        db_rms = librosa.amplitude_to_db(rms, ref=np.max)
+        silence_ratio = np.sum(db_rms < -25) / len(db_rms)
+        silence_score = 1.0 - silence_ratio
+
+        zcr = librosa.feature.zero_crossing_rate(y)[0]
+        speech_rate_score = min(1.0, max(0.0, np.mean(zcr) * 6))
+
+        audio_score = (
+            0.35 * pitch_score
+            + 0.20 * energy_score
+            + 0.30 * silence_score
+            + 0.15 * speech_rate_score
+        )
+
+        if silence_score < 0.3:
+            audio_score *= 0.5
+
+        final_score = max(0.0, min(1.0, round(audio_score, 3)))
+
+        logger.debug(
+            "audio score=%s (pitch=%.2f energy=%.2f cadence=%.2f zcr=%.2f)",
+            final_score,
+            pitch_score,
+            energy_score,
+            silence_score,
+            speech_rate_score,
+        )
+        return final_score
+
+    except Exception as e:
+        logger.error(f"[Audio] Unexpected error: {e}")
+        return None

ai/engagement/cache.py

@@ -0,0 +1,153 @@
+"""In-memory cache for engagement scores.
+
+Lightweight runtime cache used to store recent per-user signals so we can
+compute temporal/textual metrics without hitting the DB for every event.
+"""
+
+import time
+
+# Entries idle for longer than this (seconds) are evicted to bound memory.
+DEFAULT_TTL = 3600
+# Minimum gap between opportunistic eviction passes.
+EVICT_INTERVAL = 300
+# Smoothing factor for live modality scores (higher = more responsive).
+DEFAULT_EMA_ALPHA = 0.4
+# How recently the webcam must have produced a score for it to be attached to a
+# chat/voice message (seconds). Prevents saving a long-stale video score.
+VIDEO_FRESH_WINDOW = 60
+
+# Live per-user modality scores — keyed by ``user_id``.
+latest_video_scores = {}
+latest_audio_scores = {}
+
+# Text tracking — keyed by scope key ``"<session_id>|<user_id>"``:
+# last message timestamp (epoch seconds), message count, recent history.
+latest_text_ts = {}
+latest_text_count = {}
+latest_text_history = {}
+
+# key -> last activity epoch; drives TTL eviction across every store.
+_last_seen = {}
+_last_evict = [0.0]
+
+_ALL_STORES = (
+    latest_video_scores,
+    latest_audio_scores,
+    latest_text_ts,
+    latest_text_count,
+    latest_text_history,
+)
+
+
+# Key helpers
+
+
+def scope_key(user_id: str, session_id=None) -> str:
+    """Build the per-(session, user) key used by the text caches."""
+    return f"{session_id or 'global'}|{user_id}"
+
+
+def get_recent_video_score(user_id: str, max_age: int = VIDEO_FRESH_WINDOW, now=None):
+    """Return the user's cached webcam score if it is recent enough to attach."""
+    score = latest_video_scores.get(user_id)
+    if score is None:
+        return None
+    seen = _last_seen.get(user_id)
+    if seen is not None:
+        now = now if now is not None else time.time()
+        if now - seen > max_age:
+            return None
+    return score
+
+
+def session_message_total(session_id=None) -> int:
+    """Total messages recorded across all users in a session.
+
+    Used so a user's participation rate is measured within their own session
+    rather than against every user on the server.
+    """
+    prefix = f"{session_id or 'global'}|"
+    return sum(c for k, c in latest_text_count.items() if k.startswith(prefix))
+
+
+def most_recent_text_session(user_id: str):
+    """Session id of the user's most recent text activity, or ``None``.
+
+    Text caches are keyed ``"<session>|<user>"``; this finds the session whose
+    entry for this user has the latest timestamp, so live consumers (e.g. the
+    adaptive prompt) can read the user's current text engagement without knowing
+    the session id up front.
+    """
+    suffix = f"|{user_id}"
+    best_session = None
+    best_ts = -1
+    for key, ts in latest_text_ts.items():
+        if key.endswith(suffix) and ts > best_ts:
+            best_ts = ts
+            best_session = key[: -len(suffix)]
+    return best_session
+
+
+# Eviction
+
+
+def touch(key: str, now=None) -> None:
+    """Mark a cache key as active so it survives the next eviction pass."""
+    _last_seen[key] = now if now is not None else time.time()
+
+
+def evict_stale(ttl: int = DEFAULT_TTL, now=None) -> int:
+    """Drop entries idle longer than ``ttl`` seconds from every store."""
+    now = now if now is not None else time.time()
+    stale = [k for k, seen in _last_seen.items() if now - seen > ttl]
+    for k in stale:
+        for store in _ALL_STORES:
+            store.pop(k, None)
+        _last_seen.pop(k, None)
+    return len(stale)
+
+
+def maybe_evict(
+    now=None, interval: int = EVICT_INTERVAL, ttl: int = DEFAULT_TTL
+) -> None:
+    """Run :func:`evict_stale`, but at most once per ``interval`` seconds."""
+    now = now if now is not None else time.time()
+    if now - _last_evict[0] >= interval:
+        _last_evict[0] = now
+        evict_stale(ttl=ttl, now=now)
+
+
+# Smoothing
+
+
+def decay_score(store: dict, key: str, factor: float = 0.55, floor: float = 0.05):
+    """Multiply the cached score toward ``floor`` and return it.
+
+    Used when the webcam sends a frame but no face is found: instead of
+    freezing the last (often high) value, the score decays so that sustained
+    absence/looking-away drives it down within a few frames.
+    """
+    prev = store.get(key)
+    if prev is None:
+        return None
+    new = round(max(floor, prev * factor), 3)
+    store[key] = new
+    touch(key)
+    return new
+
+
+def ema_update(store: dict, key: str, value, alpha: float = DEFAULT_EMA_ALPHA):
+    """Exponentially smooth ``value`` into ``store[key]`` and return the result.
+
+    The first observation seeds the average directly; subsequent ones blend
+    ``alpha`` of the new value with ``1 - alpha`` of the running estimate.
+    A ``None`` value leaves the stored estimate untouched.
+    """
+    if value is None:
+        return store.get(key)
+    prev = store.get(key)
+    smoothed = value if prev is None else round(alpha * value + (1 - alpha) * prev, 3)
+    store[key] = smoothed
+    touch(key)
+    maybe_evict()
+    return smoothed

ai/engagement/database.py

@@ -0,0 +1,84 @@
+"""Dedicated database for engagement metrics.
+
+Engagement data lives in its own SQLite file (``var/engagement.db``),
+fully isolated from the main application database.
+Override the location with the ``ENGAGEMENT_DATABASE_URL`` env var.
+"""
+
+import os
+from contextlib import contextmanager
+from typing import Generator
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import declarative_base, sessionmaker, Session
+from sqlalchemy.pool import StaticPool
+
+# Independent declarative base — NOT shared with data.database.Base, so the
+# engagement table is never created in the main application DB.
+EngagementBase = declarative_base()
+
+ENGAGEMENT_DATABASE_URL = os.environ.get(
+    "ENGAGEMENT_DATABASE_URL", "sqlite:///./var/engagement.db"
+)
+
+
+def _make_engine(url: str):
+    if url.startswith("sqlite"):
+        # Ensure the parent directory exists for a file-based SQLite DB.
+        if url.startswith("sqlite:///") and ":memory:" not in url:
+            path = url.replace("sqlite:///", "")
+            directory = os.path.dirname(path)
+            if directory and not os.path.exists(directory):
+                os.makedirs(directory, exist_ok=True)
+        return create_engine(
+            url,
+            connect_args={"check_same_thread": False},
+            poolclass=StaticPool,
+        )
+    return create_engine(url)
+
+
+engagement_engine = _make_engine(ENGAGEMENT_DATABASE_URL)
+
+EngagementSessionLocal = sessionmaker(
+    autocommit=False, autoflush=False, bind=engagement_engine
+)
+
+
+def init_engagement_db() -> None:
+    """Create the engagement table in its dedicated database."""
+    from sqlalchemy import inspect, text
+    from . import models  # noqa: F401 — register models on EngagementBase
+
+    inspector = inspect(engagement_engine)
+    if inspector.has_table("engagement_metrics"):
+        columns = {c["name"] for c in inspector.get_columns("engagement_metrics")}
+        if not {"user_id", "text_score", "created_at"}.issubset(columns):
+            print(
+                "[Engagement DB] Dropping incompatible legacy engagement_metrics "
+                "table and recreating with the current schema.",
+                flush=True,
+            )
+            with engagement_engine.begin() as conn:
+                conn.execute(text("DROP TABLE engagement_metrics"))
+
+    EngagementBase.metadata.create_all(bind=engagement_engine)
+
+
+def get_engagement_db() -> Generator[Session, None, None]:
+    """FastAPI dependency yielding a session bound to the engagement DB."""
+    db = EngagementSessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()
+
+
+@contextmanager
+def engagement_session() -> Generator[Session, None, None]:
+    """Context manager for engagement sessions outside the request lifecycle."""
+    db = EngagementSessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()