|
| 1 | +"""Derive a wait timeout from observed step durations instead of guessing. |
| 2 | +
|
| 3 | +Hard-coded waits are a perennial source of flakiness: too short and a slow |
| 4 | +machine races the UI; too long and every failure pays the full timeout. The |
| 5 | +durable fix is to *learn* the timeout from how long the step has actually taken. |
| 6 | +``adaptive_timeout`` turns a sample of observed durations into a robust timeout: |
| 7 | +a high percentile (the slow-but-real case) scaled by a safety ``factor``, then |
| 8 | +clamped to a sane ``[min_s, max_s]`` band. |
| 9 | +
|
| 10 | +* :func:`recommend_timeout` — the single number to feed a wait / ``GateConfig``. |
| 11 | +* :func:`timeout_stats` — the same with the percentiles and clamp flags exposed |
| 12 | + for logging / tuning. |
| 13 | +
|
| 14 | +Both are pure and reuse :func:`stats.percentile`; with no samples they fall back |
| 15 | +to ``default_s`` (or ``min_s``). Imports no ``PySide6``. |
| 16 | +""" |
| 17 | +from typing import Any, Dict, List, Optional, Sequence |
| 18 | + |
| 19 | +from je_auto_control.utils.stats.stats import percentile |
| 20 | + |
| 21 | + |
| 22 | +def _clamp(value: float, min_s: float, max_s: Optional[float]) -> float: |
| 23 | + """Clamp ``value`` to ``[min_s, max_s]`` (``max_s`` None = no upper cap).""" |
| 24 | + bounded = max(float(min_s), float(value)) |
| 25 | + if max_s is not None: |
| 26 | + bounded = min(float(max_s), bounded) |
| 27 | + return bounded |
| 28 | + |
| 29 | + |
| 30 | +def _fallback(default_s: Optional[float], min_s: float) -> float: |
| 31 | + """The timeout to use when there are no duration samples.""" |
| 32 | + return float(default_s) if default_s is not None else float(min_s) |
| 33 | + |
| 34 | + |
| 35 | +def recommend_timeout(durations: Sequence[float], *, percentile_q: float = 95.0, |
| 36 | + factor: float = 1.5, min_s: float = 1.0, |
| 37 | + max_s: Optional[float] = 60.0, |
| 38 | + default_s: Optional[float] = None) -> float: |
| 39 | + """Recommend a wait timeout (seconds) from observed ``durations``. |
| 40 | +
|
| 41 | + Takes the ``percentile_q``-th percentile of the samples, scales it by |
| 42 | + ``factor``, and clamps to ``[min_s, max_s]``. With no samples returns |
| 43 | + ``default_s`` (or ``min_s``). |
| 44 | + """ |
| 45 | + samples = [float(d) for d in durations if d is not None] |
| 46 | + if not samples: |
| 47 | + return _fallback(default_s, min_s) |
| 48 | + scaled = percentile(samples, float(percentile_q)) * float(factor) |
| 49 | + return _clamp(scaled, min_s, max_s) |
| 50 | + |
| 51 | + |
| 52 | +def timeout_stats(durations: Sequence[float], *, percentile_q: float = 95.0, |
| 53 | + factor: float = 1.5, min_s: float = 1.0, |
| 54 | + max_s: Optional[float] = 60.0, |
| 55 | + default_s: Optional[float] = None) -> Dict[str, Any]: |
| 56 | + """Recommend a timeout and expose the percentiles and clamp decisions. |
| 57 | +
|
| 58 | + Returns ``{n, p50, p_high, percentile_q, recommended, floored, capped}``. |
| 59 | + """ |
| 60 | + samples: List[float] = [float(d) for d in durations if d is not None] |
| 61 | + recommended = recommend_timeout( |
| 62 | + samples, percentile_q=percentile_q, factor=factor, min_s=min_s, |
| 63 | + max_s=max_s, default_s=default_s) |
| 64 | + if not samples: |
| 65 | + return {"n": 0, "p50": None, "p_high": None, |
| 66 | + "percentile_q": float(percentile_q), |
| 67 | + "recommended": recommended, "floored": False, "capped": False} |
| 68 | + p_high = percentile(samples, float(percentile_q)) |
| 69 | + scaled = p_high * float(factor) |
| 70 | + return { |
| 71 | + "n": len(samples), |
| 72 | + "p50": percentile(samples, 50.0), |
| 73 | + "p_high": p_high, |
| 74 | + "percentile_q": float(percentile_q), |
| 75 | + "recommended": recommended, |
| 76 | + "floored": scaled < float(min_s), |
| 77 | + "capped": max_s is not None and scaled > float(max_s), |
| 78 | + } |
0 commit comments