app_enhancements.py

import json
import math
import os
import time

import torch


PRESETS = {
    "Fast": {
        "magnification": 1.0,
        "confidence_threshold": 0.55,
        "nms_radius": 10.0,
        "tracker_max_distance": 65.0,
        "inference_batch_size": 16,
        "frame_skip": 5,
        "patch_overlap": 0.0,
    },
    "Balanced": {
        "magnification": 1.5,
        "confidence_threshold": 0.5,
        "nms_radius": 8.0,
        "tracker_max_distance": 50.0,
        "inference_batch_size": 8,
        "frame_skip": 2,
        "patch_overlap": 0.25,
    },
    "Accurate": {
        "magnification": 2.0,
        "confidence_threshold": 0.45,
        "nms_radius": 6.0,
        "tracker_max_distance": 40.0,
        "inference_batch_size": 4,
        "frame_skip": 1,
        "patch_overlap": 0.5,
    },
}


def load_config(config_path, defaults):
    if os.path.exists(config_path):
        with open(config_path, "r", encoding="utf-8") as f:
            return {**defaults, **json.load(f)}
    return defaults


def save_config(config_path, config):
    with open(config_path, "w", encoding="utf-8") as f:
        json.dump(config, f, indent=2)


def run_with_oom_recovery(process_fn, *args, batch_size=8, min_batch_size=1, **kwargs):
    current_batch = max(min_batch_size, int(batch_size))
    while current_batch >= min_batch_size:
        try:
            return process_fn(*args, batch_size=current_batch, **kwargs), current_batch
        except RuntimeError as exc:
            if "out of memory" not in str(exc).lower():
                raise
            if torch.cuda.is_available():
                torch.cuda.empty_cache()
            if current_batch == min_batch_size:
                raise
            current_batch = max(min_batch_size, current_batch // 2)
    raise RuntimeError("CUDA OOM recovery failed.")


def confidence_interval(values, z=1.96):
    if not values:
        return 0.0, 0.0
    if len(values) == 1:
        return float(values[0]), float(values[0])
    mean = sum(values) / len(values)
    variance = sum((x - mean) ** 2 for x in values) / (len(values) - 1)
    margin = z * math.sqrt(variance) / math.sqrt(len(values))
    return mean - margin, mean + margin


def result_summary(start_time, frames_read, frames_analyzed, counts):
    elapsed = max(time.perf_counter() - start_time, 1e-9)
    ci_low, ci_high = confidence_interval(counts)
    return {
        "elapsed_sec": round(elapsed, 3),
        "frames_read": frames_read,
        "frames_analyzed": frames_analyzed,
        "effective_fps": round(frames_read / elapsed, 3),
        "analysis_fps": round(frames_analyzed / elapsed, 3),
        "avg_count": round(sum(counts) / len(counts), 3) if counts else 0,
        "peak_count": max(counts) if counts else 0,
        "count_95ci_low": round(ci_low, 3),
        "count_95ci_high": round(ci_high, 3),
    }


APP_PASTE_SECTION = r'''
CONFIG_PATH = os.path.join(BASE_DIR, "civic_pulse_config.json")
PRESETS = {
    "Fast": {"magnification": 1.0, "confidence_threshold": 0.55, "nms_radius": 10.0, "tracker_max_distance": 65.0, "inference_batch_size": 16, "frame_skip": 5, "patch_overlap": 0.0},
    "Balanced": {"magnification": 1.5, "confidence_threshold": 0.5, "nms_radius": 8.0, "tracker_max_distance": 50.0, "inference_batch_size": 8, "frame_skip": 2, "patch_overlap": 0.25},
    "Accurate": {"magnification": 2.0, "confidence_threshold": 0.45, "nms_radius": 6.0, "tracker_max_distance": 40.0, "inference_batch_size": 4, "frame_skip": 1, "patch_overlap": 0.5},
}

def process_with_oom_recovery(*args, batch_size, **kwargs):
    while batch_size >= 1:
        try:
            return process_frame(*args, batch_size=batch_size, **kwargs), batch_size
        except RuntimeError as exc:
            if "out of memory" not in str(exc).lower():
                raise
            if torch.cuda.is_available():
                torch.cuda.empty_cache()
            if batch_size == 1:
                raise
            batch_size = max(1, batch_size // 2)
'''