stats.py

"""
stats.py — Centralized statistics tracker for Iris Recognition App
Tracks live events in memory + persists to a JSON file on disk.
Call update_*() functions from your routes in app.py.
Call get_dashboard_stats() from a new /stats endpoint.
"""

import os
import json
import threading
from datetime import datetime, date
from collections import defaultdict

# ─────────────────────────────────────────
# Config
# ─────────────────────────────────────────
STATS_FILE = os.path.join(os.path.dirname(os.path.abspath(__file__)), "stats_data.json")
_lock = threading.Lock()  # Thread-safe writes


# ─────────────────────────────────────────
# Default stats structure
# ─────────────────────────────────────────
def _default_stats():
    return {
        # ── Login stats
        "total_logins":          0,
        "successful_logins":     0,
        "failed_logins":         0,        # identity not recognised
        "login_scores":          [],        # list of float scores (last 100)

        # ── Registration stats
        "total_registrations":   0,
        "registrations_by_date": {},        # {"2025-07-01": 3, ...}

        # ── Phase 1 (Iris detection)
        "phase1_total":          0,
        "phase1_iris":           0,
        "phase1_non_iris":       0,

        # ── Phase 2 (PAD — Presentation Attack Detection)
        "phase2_total":          0,
        "phase2_real":           0,
        "phase2_fake":           0,         # attacks blocked
        "phase2_confidences":    [],        # list of float (last 100)

        # ── GAN generation
        "total_generated":       0,

        # ── Daily activity (last 14 days)
        "daily_logins":          {},        # {"2025-07-01": 5, ...}
        "daily_attacks":         {},        # {"2025-07-01": 2, ...}
    }


# ─────────────────────────────────────────
# Load / Save helpers
# ─────────────────────────────────────────
def _load():
    if os.path.exists(STATS_FILE):
        try:
            with open(STATS_FILE, "r") as f:
                data = json.load(f)
            # Backfill any missing keys (in case file is from older version)
            defaults = _default_stats()
            for key, val in defaults.items():
                data.setdefault(key, val)
            return data
        except Exception:
            pass
    return _default_stats()


def _save(data):
    try:
        with open(STATS_FILE, "w") as f:
            json.dump(data, f, indent=2)
    except Exception as e:
        print(f"⚠️ Stats save failed: {e}")


def _today():
    return date.today().isoformat()   # "2025-07-01"


# ─────────────────────────────────────────
# Public update functions — call from app.py
# ─────────────────────────────────────────

def update_login(success: bool, score: float):
    """Call this every time /login is hit."""
    with _lock:
        data = _load()
        today = _today()

        data["total_logins"] += 1
        if success:
            data["successful_logins"] += 1
        else:
            data["failed_logins"] += 1

        # Keep only last 100 scores (to avoid file bloat)
        data["login_scores"].append(round(score, 4))
        data["login_scores"] = data["login_scores"][-100:]

        # Daily activity
        data["daily_logins"][today] = data["daily_logins"].get(today, 0) + 1

        _save(data)


def update_registration(person_id: str):
    """Call this every time /register succeeds."""
    with _lock:
        data = _load()
        today = _today()

        data["total_registrations"] += 1
        data["registrations_by_date"][today] = \
            data["registrations_by_date"].get(today, 0) + 1

        _save(data)


def update_phase1(is_iris: bool):
    """Call this every time /phase1 runs."""
    with _lock:
        data = _load()

        data["phase1_total"] += 1
        if is_iris:
            data["phase1_iris"] += 1
        else:
            data["phase1_non_iris"] += 1

        _save(data)


def update_phase2(is_real: bool, confidence: float):
    """Call this every time /phase2 runs."""
    with _lock:
        data = _load()
        today = _today()

        data["phase2_total"] += 1
        if is_real:
            data["phase2_real"] += 1
        else:
            data["phase2_fake"] += 1
            data["daily_attacks"][today] = data["daily_attacks"].get(today, 0) + 1

        data["phase2_confidences"].append(round(confidence, 4))
        data["phase2_confidences"] = data["phase2_confidences"][-100:]

        _save(data)


def update_generation(count: int):
    """Call this every time /generate runs."""
    with _lock:
        data = _load()
        data["total_generated"] += count
        _save(data)


# ─────────────────────────────────────────
# Main dashboard endpoint data
# ─────────────────────────────────────────

def get_dashboard_stats(gallery: dict) -> dict:
    """
    Returns all stats needed for frontend charts + cards.
    Pass in the live `gallery` dict from iris_recognition.py.
    """
    with _lock:
        data = _load()

    today = _today()

    # ── Gallery breakdown: samples per person
    gallery_sizes = {
        person: int(embeddings.shape[0])
        for person, embeddings in gallery.items()
    }

    # ── Last 14 days of login + attack activity
    from datetime import timedelta
    last_14 = [
        (date.today() - timedelta(days=i)).isoformat()
        for i in range(13, -1, -1)
    ]
    daily_login_trend  = [data["daily_logins"].get(d, 0)  for d in last_14]
    daily_attack_trend = [data["daily_attacks"].get(d, 0) for d in last_14]

    # ── Average login score
    scores = data["login_scores"]
    avg_score = round(sum(scores) / len(scores), 4) if scores else 0.0

    # ── PAD confidence average
    confs = data["phase2_confidences"]
    avg_pad_conf = round(sum(confs) / len(confs), 4) if confs else 0.0

    return {
        # ── Summary cards
        "gallery": {
            "total_persons":    len(gallery),
            "total_samples":    sum(gallery_sizes.values()),
            "samples_per_person": gallery_sizes,          # for bar chart
        },
        "logins": {
            "total":            data["total_logins"],
            "successful":       data["successful_logins"],
            "failed":           data["failed_logins"],
            "success_rate":     round(
                data["successful_logins"] / data["total_logins"] * 100, 1
            ) if data["total_logins"] else 0,
            "avg_score":        avg_score,
            "recent_scores":    data["login_scores"][-20:],   # for line chart
        },
        "registrations": {
            "total":            data["total_registrations"],
            "by_date":          data["registrations_by_date"],
        },
        "phase1": {
            "total":            data["phase1_total"],
            "iris_detected":    data["phase1_iris"],
            "non_iris":         data["phase1_non_iris"],
        },
        "phase2": {
            "total":            data["phase2_total"],
            "real":             data["phase2_real"],
            "fake_blocked":     data["phase2_fake"],
            "attack_rate":      round(
                data["phase2_fake"] / data["phase2_total"] * 100, 1
            ) if data["phase2_total"] else 0,
            "avg_confidence":   avg_pad_conf,
        },
        "gan": {
            "total_generated":  data["total_generated"],
        },
        # ── Time-series for charts (last 14 days)
        "trends": {
            "dates":            last_14,
            "daily_logins":     daily_login_trend,
            "daily_attacks":    daily_attack_trend,
        },
        "today": {
            "date":             today,
            "logins_today":     data["daily_logins"].get(today, 0),
            "attacks_today":    data["daily_attacks"].get(today, 0),
        }
    }