telemetry.py

#!/usr/bin/env python3
"""
telemetry.py — лёгкий лог прогонов MakeFaces в SQLite.

Зачем: копить данные для подстройки границ зон идентичности — в частности,
решения юзера по серой зоне (оставил/выкинул кадр с такой-то близостью).
Это честная разметка: появляется истина "для этого человека близость X = ещё он".

Две таблицы:
  runs   — одна строка на прогон (параметры + агрегаты)
  grays  — одна строка на кадр серой зоны (близость, резкость, решение юзера)
           связаны по run_id

На Spaces (эфемерный диск) логирование смысла не имеет — там это no-op.
Локально пишет в makefaces_runs.db рядом со скриптом.
"""

import os
import sqlite3
import time
from pathlib import Path

IS_SPACE = bool(os.environ.get("SPACE_ID"))
DB_PATH = Path(__file__).parent / "makefaces_runs.db"


def _enabled() -> bool:
    # на Spaces не логируем (диск всё равно сотрётся), локально — да
    return not IS_SPACE


def _conn():
    c = sqlite3.connect(str(DB_PATH))
    c.execute("""CREATE TABLE IF NOT EXISTS runs (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        ts REAL, video TEXT, frames_total INTEGER,
        fps REAL, num INTEGER, target INTEGER, size INTEGER,
        min_sharp REAL, min_face INTEGER,
        n_good INTEGER, n_small INTEGER,
        floor REAL, ceil REAL,
        n_take INTEGER, n_pending INTEGER,
        drop_blur INTEGER, drop_noface INTEGER, drop_identity INTEGER,
        final_size INTEGER, hit_target INTEGER
    )""")
    c.execute("""CREATE TABLE IF NOT EXISTS grays (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id INTEGER, sim REAL, sharp REAL, kept INTEGER,
        FOREIGN KEY(run_id) REFERENCES runs(id)
    )""")
    c.execute("""CREATE TABLE IF NOT EXISTS slots (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        run_id INTEGER, slot TEXT, frames INTEGER,
        n_good INTEGER, n_small INTEGER, quota INTEGER,
        drop_blur INTEGER, drop_no_face INTEGER, drop_edge INTEGER, multi_handled INTEGER,
        FOREIGN KEY(run_id) REFERENCES runs(id)
    )""")
    return c


def log_run(**fields) -> int | None:
    """Записать строку прогона. Возвращает run_id (или None, если логирование off)."""
    if not _enabled():
        return None
    cols = ["ts", "video", "frames_total", "fps", "num", "target", "size",
            "min_sharp", "min_face", "n_good", "n_small", "floor", "ceil",
            "n_take", "n_pending", "drop_blur", "drop_noface", "drop_identity",
            "final_size", "hit_target"]
    fields.setdefault("ts", time.time())
    vals = [fields.get(c) for c in cols]
    with _conn() as c:
        cur = c.execute(
            f"INSERT INTO runs ({','.join(cols)}) VALUES ({','.join('?' * len(cols))})",
            vals,
        )
        return cur.lastrowid


def log_slots(run_id, slot_rows):
    """Записать пер-слотовую разбивку. slot_rows — список dict с полями:
    slot, frames, n_good, n_small, quota, drop_blur, drop_no_face, drop_edge, multi_handled."""
    if not _enabled() or run_id is None:
        return
    rows = [(run_id, r["slot"], r["frames"], r["n_good"], r["n_small"], r["quota"],
             r["drop_blur"], r["drop_no_face"], r.get("drop_edge", 0),
             r["multi_handled"]) for r in slot_rows]
    with _conn() as c:
        c.executemany(
            "INSERT INTO slots (run_id, slot, frames, n_good, n_small, quota, "
            "drop_blur, drop_no_face, drop_edge, multi_handled) "
            "VALUES (?,?,?,?,?,?,?,?,?,?)", rows
        )


def log_grays(run_id, pending, kept_paths):
    """Записать решения по серой зоне: для каждого pending — оставлен он или нет.

    pending     — список dict(path, sim, sharp) из rescue_smalls_zoned
    kept_paths  — set путей, которые юзер решил оставить
    """
    if not _enabled() or run_id is None:
        return
    kept = set(kept_paths)
    rows = [(run_id, p["sim"], p["sharp"], 1 if p["path"] in kept else 0) for p in pending]
    with _conn() as c:
        c.executemany(
            "INSERT INTO grays (run_id, sim, sharp, kept) VALUES (?,?,?,?)", rows
        )