app/active_learning.py

"""
app/active_learning.py

Scores PENDING patches by edge complexity (variance of Laplacian).
High complexity → low confidence_index → shown first in the labeler UI.
Runs every 2 hours as a background task started from app/cron.py.
"""

from __future__ import annotations

import asyncio
import datetime
import logging
from pathlib import Path

import cv2
import numpy as np
from sqlalchemy import or_
from sqlalchemy.orm import Session

from app.database import SessionLocal
from app.models import TelemetryComponent

logger = logging.getLogger("active_learning")

IMAGES_DIR = Path(__file__).resolve().parent.parent / "data" / "active_learning_ds" / "images"


def _laplacian_variance(img_path: Path) -> float:
    """Returns variance of Laplacian for a grayscale image. Higher = more edges."""
    try:
        img = cv2.imread(str(img_path), cv2.IMREAD_GRAYSCALE)
        if img is None:
            return 0.0
        return float(cv2.Laplacian(img, cv2.CV_64F).var())
    except Exception as e:
        logger.warning("Could not score %s: %s", img_path.name, e)
        return 0.0


def score_unlabeled_pool(db: Session, batch_size: int = 200) -> int:
    """
    Scores up to batch_size unlocked PENDING components and updates confidence_index.
    Returns number of components scored.
    """
    now = datetime.datetime.now(datetime.timezone.utc)

    components = (
        db.query(TelemetryComponent)
        .filter(
            TelemetryComponent.validation_status == "PENDING",
            or_(
                TelemetryComponent.locked_until == None,
                TelemetryComponent.locked_until <= now,
            ),
        )
        .order_by(TelemetryComponent.confidence_index.asc())
        .limit(batch_size)
        .all()
    )

    if not components:
        logger.info("No pending components to score.")
        return 0

    logger.info("Scoring %d components...", len(components))
    scored = 0

    for item in components:
        img_path = IMAGES_DIR / item.file_path
        if not img_path.exists():
            logger.warning("Image not found on disk: %s", item.file_path)
            continue

        var = _laplacian_variance(img_path)
        # High variance → complex image → low confidence_index → labelled first
        item.confidence_index = float(1.0 - np.exp(-var / 400.0))
        scored += 1

    try:
        db.commit()
        logger.info("Scored %d components.", scored)
    except Exception as e:
        db.rollback()
        logger.error("Commit failed after scoring: %s", e)

    return scored


async def active_learning_loop() -> None:
    """Runs every 2 hours. Called from app/cron.py start_background_tasks()."""
    logger.info("Active learning loop started (interval: 2h).")
    await asyncio.sleep(15)  # short delay to let FastAPI finish startup

    while True:
        db = SessionLocal()
        try:
            score_unlabeled_pool(db)
        except Exception as e:
            logger.error("Scoring loop error: %s", e)
            db.rollback()
        finally:
            db.close()

        await asyncio.sleep(7200)