dfod_config.py

from __future__ import annotations

import copy
import hashlib
import json
import warnings
from typing import Any, Dict, Iterable, Tuple


ConfigDict = Dict[str, Any]


DEFAULT_CONFIG: ConfigDict = {
    "perception": {
        "detector_model": "yolov8n",
        "objectness_enabled": True,
        "objectness_iou_threshold": 0.3,
        "objectness_rerank_weight": 0.35,
        "min_objectness": 0.0,
        "rerank_by_combined_score": True,
    },
    "support_quality": {
        "yolo_weight": 0.45,
        "objectness_weight": 0.30,
        "area_weight": 0.15,
        "edge_weight": 0.10,
        "weight_temperature": 0.2,
    },
    "memory": {
        "max_prototypes_per_class": 2,
        "cluster_min_supports": 6,
        "min_supports_per_mode": 3,
        "clustering_seed": 0,
        "variance_alpha": 0.3,
        "variance_floor": 1e-4,
    },
    "reasoning": {
        "metrics": ["mahalanobis"],
        "primary_metric": "mahalanobis",
        "enable_metric_fusion": False,
        "metric_weights": {
            "mahalanobis": 1.0,
            "cosine": 1.0,
            "euclidean": 1.0,
            "manhattan": 1.0,
        },
        "softmax_temperature": 1.0,
        "score_threshold": float("-inf"),
        "margin_threshold": 0.0,
    },
    "calibration": {
        "enabled": False,
        "calibration_path": None,
        "known_score_quantile": 0.05,
        "known_margin_quantile": 0.05,
        "rejection_known_floor": 0.80,
        "update_precision_floor": 0.98,
    },
    "updates": {
        "enabled": True,
        "score_threshold_min": float("-inf"),
        "margin_threshold_min": 0.0,
        "update_percentile": 90.0,
        "margin_percentile": 75.0,
        "det_conf_threshold": 0.70,
        "objectness_threshold": 0.20,
        "quality_threshold": 0.20,
        "max_updates_per_image": 3,
        "max_memory_per_class": 20,
        "memory_keep_policy": "recent",
        "persist_updated_memory": False,
        "dynamic_threshold_min_decisions": 5,
    },
    "adapter": {
        "enabled": True,
        "allow_training": True,
        "bottleneck_dim": 256,
        "image_size": 224,
        "weights_path": "adapter.pt",
        "train_on_support_build": True,
        "online_enabled": True,
        "online_min_examples": 3,
        "replay_support_per_class": 2,
        "finetune_epochs": 5,
        "finetune_lr": 1e-3,
        "reembed_after_update": True,
    },
    "diagnostics": {
        "include": True,
    },
}


_ALIAS_MAP: Dict[str, Tuple[str, str]] = {
    "tau": ("reasoning", "score_threshold"),
    "margin": ("reasoning", "margin_threshold"),
    "ll_threshold": ("reasoning", "score_threshold"),
    "ll_margin": ("reasoning", "margin_threshold"),
    "temperature": ("reasoning", "softmax_temperature"),
    "alpha": ("memory", "variance_alpha"),
    "variance_floor": ("memory", "variance_floor"),
    "tau_update_min": ("updates", "score_threshold_min"),
    "margin_update_min": ("updates", "margin_threshold_min"),
    "update_percentile": ("updates", "update_percentile"),
    "margin_percentile": ("updates", "margin_percentile"),
    "det_conf_update": ("updates", "det_conf_threshold"),
    "max_updates_per_image": ("updates", "max_updates_per_image"),
    "max_memory_per_class": ("updates", "max_memory_per_class"),
    "memory_keep_policy": ("updates", "memory_keep_policy"),
}


def _deep_update(dst: ConfigDict, src: ConfigDict) -> None:
    for key, value in src.items():
        if isinstance(value, dict) and isinstance(dst.get(key), dict):
            _deep_update(dst[key], value)
        else:
            dst[key] = value


def _set_nested(cfg: ConfigDict, path: Tuple[str, str], value: Any) -> None:
    section, key = path
    if section not in cfg or not isinstance(cfg[section], dict):
        cfg[section] = {}
    cfg[section][key] = value


def _normalize_metrics(cfg: ConfigDict) -> None:
    reasoning = cfg["reasoning"]
    metrics = reasoning.get("metrics", [])
    if isinstance(metrics, str):
        metrics = [metrics]
    metrics = [str(metric).strip().lower() for metric in metrics if str(metric).strip()]
    primary = str(reasoning.get("primary_metric", "mahalanobis")).strip().lower()
    if primary not in metrics:
        metrics.insert(0, primary)
    if not metrics:
        metrics = [primary]
    reasoning["metrics"] = list(dict.fromkeys(metrics))
    reasoning["primary_metric"] = primary


def normalize_config(config: Dict[str, Any] | None = None) -> ConfigDict:
    cfg = copy.deepcopy(DEFAULT_CONFIG)
    if not config:
        _normalize_metrics(cfg)
        return cfg

    for key, value in config.items():
        if key in cfg and isinstance(cfg[key], dict) and isinstance(value, dict):
            _deep_update(cfg[key], value)
            continue
        if key in _ALIAS_MAP:
            warnings.warn(
                f"Config key '{key}' is deprecated; use '{_ALIAS_MAP[key][0]}.{_ALIAS_MAP[key][1]}' instead.",
                DeprecationWarning,
                stacklevel=2,
            )
            _set_nested(cfg, _ALIAS_MAP[key], value)
            continue
        cfg[key] = value

    _normalize_metrics(cfg)
    return cfg


def config_fingerprint(config: Dict[str, Any] | None, sections: Iterable[str] | None = None) -> str:
    cfg = normalize_config(config)
    if sections is not None:
        payload = {section: cfg.get(section) for section in sections}
    else:
        payload = cfg
    blob = json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
    return hashlib.sha1(blob).hexdigest()