metric_policy
Metric policy for rolling performance baseline comparisons.
fastvideo.performance.metric_policy.MetricPolicy
MetricPolicy(key: str, label: str, precision: int, lower_is_better: bool, threshold_percent: float, threshold_absolute: float, gated: bool = True)
Source
@dataclass(frozen=True)
class MetricPolicy:
key: str
label: str
precision: int
lower_is_better: bool
threshold_percent: float
threshold_absolute: float
gated: bool = Truefastvideo.performance.metric_policy.MetricDelta
MetricDelta(absolute: float, percent: float, threshold_exceeded: bool, regressed: bool)
Source
@dataclass(frozen=True)
class MetricDelta:
absolute: float
percent: float
threshold_exceeded: bool
regressed: boolfastvideo.performance.metric_policy.resolve_metric_policies
resolve_metric_policies(threshold_overrides: Mapping[str, Any] | None) -> tuple[MetricPolicy, ...]
Return default metric policies with optional per-metric overrides.
Source
def resolve_metric_policies(threshold_overrides: Mapping[str, Any] | None, ) -> tuple[MetricPolicy, ...]:
"""Return default metric policies with optional per-metric overrides."""
if not isinstance(threshold_overrides, Mapping):
threshold_overrides = {}
policies: list[MetricPolicy] = []
for base_policy in DEFAULT_METRIC_POLICIES:
raw_override = threshold_overrides.get(base_policy.key, {})
if not isinstance(raw_override, Mapping):
raw_override = {}
threshold_percent = _optional_float(raw_override.get("threshold_percent"))
threshold_absolute = _optional_float(raw_override.get("threshold_absolute"))
gated = _optional_bool(raw_override.get("gated"))
policies.append(
MetricPolicy(
key=base_policy.key,
label=base_policy.label,
precision=base_policy.precision,
lower_is_better=base_policy.lower_is_better,
threshold_percent=(base_policy.threshold_percent if threshold_percent is None else threshold_percent),
threshold_absolute=(base_policy.threshold_absolute
if threshold_absolute is None else threshold_absolute),
gated=base_policy.gated if gated is None else gated,
))
return tuple(policies)</code></pre></details>
fastvideo.performance.metric_policy.serialize_metric_thresholds
serialize_metric_thresholds(policies: tuple[MetricPolicy, ...]) -> dict[str, dict[str, float | bool]]
Source
def serialize_metric_thresholds(policies: tuple[MetricPolicy, ...], ) -> dict[str, dict[str, float | bool]]:
return {
policy.key: {
"threshold_percent": policy.threshold_percent,
"threshold_absolute": policy.threshold_absolute,
"gated": policy.gated,
}
for policy in policies
}
fastvideo.performance.metric_policy.regression_delta
regression_delta(policy: MetricPolicy, current: float, baseline: float) -> MetricDelta | None
Source
def regression_delta(
policy: MetricPolicy,
current: float,
baseline: float,
) -> MetricDelta | None:
if baseline <= 0:
return None
absolute_delta = current - baseline if policy.lower_is_better else baseline - current
percent_delta = absolute_delta / baseline
threshold_exceeded = (percent_delta > policy.threshold_percent and absolute_delta > policy.threshold_absolute)
return MetricDelta(
absolute=absolute_delta,
percent=percent_delta,
threshold_exceeded=threshold_exceeded,
regressed=policy.gated and threshold_exceeded,
)