Spaces:
Sleeping
Sleeping
File size: 3,979 Bytes
3da2703 6862340 3da2703 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 | """
Node 5: Clip Disposition Recommender.
Deterministic heuristic to recommend accept/reject/fix/defer for a video clip.
This module is standalone to Sub-env 2 and only consumes clip evidence plus
dataset context.
"""
from src.schemas.subenv2 import (
ClipDispositionObservation,
ClipDispositionAction,
)
def recommend_clip_disposition(obs: ClipDispositionObservation) -> ClipDispositionAction:
"""
Recommend a disposition for a clip using a rule-based heuristic.
"""
dossier = obs.evidence_dossier
# Step 1: Base quality score [0.0, 1.0]
drift_map = {"none": 1.0, "minor": 0.7, "moderate": 0.3, "severe": 0.0}
drift_score = drift_map.get(dossier.identity_drift_severity, 0.0)
stability = 0.0 if dossier.temporal_instability_flag else 0.3
sync = 0.3 if dossier.lip_sync_quality in ["good", "acceptable"] else 0.0
phoneme_val = 0.2 * dossier.unique_phoneme_value
redundancy = 0.2 * (1.0 - dossier.dataset_redundancy_score)
quality = (drift_score + stability + sync + phoneme_val + redundancy) / 2.0
# Step 2: Disposition
disposition = "accept"
if quality >= 0.7:
disposition = "accept"
elif quality < 0.3 and dossier.unique_phoneme_value <= 0.3:
disposition = "reject"
elif quality < 0.3 and dossier.unique_phoneme_value > 0.3:
disposition = "fix"
elif 0.3 <= quality < 0.5:
disposition = "fix"
elif 0.5 <= quality < 0.7 and dossier.estimated_training_impact == "neutral":
disposition = "defer"
else:
disposition = "accept"
# Field assembly
fix_instructions = None
estimated_fix_effort = None
if disposition == "fix":
fix_instructions = []
fix_count = 0
if dossier.temporal_instability_flag:
fix_instructions.append("trim frames with temporal instability in jaw landmark region")
fix_count += 1
if dossier.lip_sync_quality == "poor":
fix_instructions.append("re-record segment — lip sync confidence below threshold")
fix_count += 1
if dossier.dataset_redundancy_score > 0.7:
fix_instructions.append("clip is redundant — consider replacing with novel scenario")
fix_count += 1
fix_instructions.append(f"retained value: unique_phoneme_value={dossier.unique_phoneme_value:.2f}")
# estimated_fix_effort
if dossier.temporal_instability_flag and fix_count == 1:
estimated_fix_effort = "trivial"
elif fix_count >= 2:
estimated_fix_effort = "high"
else:
estimated_fix_effort = "moderate"
defer_reason = None
if disposition == "defer":
defer_reason = f"quality borderline ({quality:.2f}) — manual review recommended"
# Override handling
override_decision = "not_applicable"
override_justification = None
if disposition in ["accept", "fix"] and dossier.estimated_training_impact == "negative":
override_decision = "applied"
override_justification = f"accepting despite negative training impact: {dossier.primary_rejection_reason}"
# Dataset impact reasoning
dataset_impact_reasoning = (
f"Dataset phoneme gaps: {list(obs.phoneme_gap_severity.keys())}. "
f"Pose gaps: {list(obs.pose_gap_severity.keys())}. "
f"This clip {'addresses' if dossier.unique_phoneme_value > 0.3 else 'does not address'} critical gaps."
)
return ClipDispositionAction(
disposition=disposition,
confidence=float(quality),
rejection_reasons=[dossier.primary_rejection_reason] if disposition == "reject" and dossier.primary_rejection_reason else None,
fix_instructions=fix_instructions,
estimated_fix_effort=estimated_fix_effort,
defer_reason=defer_reason,
dataset_impact_reasoning=dataset_impact_reasoning,
override_decision=override_decision,
override_justification=override_justification,
)
|