Add artifact.preflight_checker/src/artifact_preflight_checker/core.py

artifact.preflight_checker/src/artifact_preflight_checker/core.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+import logging
+from typing import List, Dict, Any
+
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+class InputValidationError(ValueError):
+    """Custom exception for invalid run input data."""
+    pass
+
+
+def _validate_run_structure(runs: List[Dict[str, Any]]) -> None:
+    if not isinstance(runs, list):
+        raise InputValidationError("valid_runs must be a list of dictionaries.")
+    for r in runs:
+        if not isinstance(r, dict):
+            raise InputValidationError("Each run must be a dictionary.")
+        if 'near_expiry_unpinned' not in r or not isinstance(r['near_expiry_unpinned'], (int, float)):
+            raise InputValidationError("Each run must contain a numeric 'near_expiry_unpinned' metric.")
+
+
+def run_preflight_check(valid_runs: List[Dict[str, Any]]) -> Dict[str, Any]:
+    """Performs a preflight check on given valid runs to determine freeze stability.
+
+    Args:
+        valid_runs (list[dict]): List of run metric dictionaries including 'near_expiry_unpinned'.
+
+    Returns:
+        dict: A JSON-compatible dictionary containing freeze_ok, freeze_target, and freeze_tol.
+    """
+    logger.info("Starting preflight check for %d runs", len(valid_runs))
+
+    _validate_run_structure(valid_runs)
+
+    # Simple logic for freeze assessment based on average of 'near_expiry_unpinned'
+    freeze_values = [float(run['near_expiry_unpinned']) for run in valid_runs]
+
+    if not freeze_values:
+        logger.warning("No freeze values found.")
+        return {"freeze_ok": False, "freeze_target": 0.0, "freeze_tol": 0.0}
+
+    avg_freeze = sum(freeze_values) / len(freeze_values)
+    freeze_target = 0.5  # Target proportion (example reference)
+    freeze_tol = 0.1     # Allowed deviation
+
+    lower_bound = freeze_target - freeze_tol
+    upper_bound = freeze_target + freeze_tol
+    freeze_ok = lower_bound <= avg_freeze <= upper_bound
+
+    result = {
+        "freeze_ok": freeze_ok,
+        "freeze_target": freeze_target,
+        "freeze_tol": freeze_tol
+    }
+
+    logger.info("Preflight check completed. freeze_ok=%s avg=%.3f target=%.3f tol=%.3f",
+                freeze_ok, avg_freeze, freeze_target, freeze_tol)
+
+    assert all(k in result for k in ("freeze_ok", "freeze_target", "freeze_tol")), \
+        "Result missing required keys."
+
+    return result