DeepNeural-AutoExploration/interaction_evaluator_bridge.py at main · sunghunkwag/DeepNeural-AutoExploration · GitHub

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
"""Bridge from interaction residues to evaluator-evolution decisions."""

from __future__ import annotations

from typing import Dict, List, Sequence

from evaluator_evolution import EvaluatorCandidate, EvaluatorGenome, score_decision
from interaction_residue_layer import InteractionResidue


class InteractionEvaluatorBridge:
    """Exports interaction residue evidence for the bounded evaluator path."""

    def __init__(self, *, generation: int = 0) -> None:
        self.generation = int(generation)

    def to_decision_records(
        self,
        residues: Sequence[InteractionResidue],
        *,
        split: str = "validation",
        generation: int | None = None,
        control_policy_scores: Dict[str, float] | None = None,
    ) -> List[Dict[str, object]]:
        if split == "test":
            raise ValueError("interaction evaluator bridge cannot use test split data")
        controls = dict(control_policy_scores or {})
        controls.setdefault("no_op", 0.0)
        controls.setdefault("random", 0.0)
        records: List[Dict[str, object]] = []
        for residue in residues:
            decision = residue.to_evaluator_decision(self.generation if generation is None else int(generation))
            decision["used_test_results"] = False
            decision["candidate_metadata"] = {
                "source": "interaction_operating_scaffold",
                "split": split,
                "interaction_failure_type": residue.failure_type,
                "timestamp_ms": residue.timestamp_ms,
                "candidate_family": "interaction_policy",
            }
            decision["control_policy_scores"] = {
                "no_op": float(controls["no_op"]),
                "random": float(controls["random"]),
            }
            decision.setdefault("no_op_control_delta", float(controls["no_op"]))
            decision.setdefault("random_control_delta", float(controls["random"]))
            self.validate_decision_record(decision)
            records.append(decision)
        return records

    @staticmethod
    def validate_decision_record(decision: Dict[str, object]) -> None:
        if decision.get("used_test_results") is not False:
            raise ValueError("interaction decisions must set used_test_results = False")
        metadata = decision.get("candidate_metadata", {})
        if isinstance(metadata, dict) and metadata.get("split") == "test":
            raise ValueError("interaction decisions cannot be exported from test split data")
        if "no_op_control_delta" not in decision:
            raise ValueError("interaction decision missing no-op control")
        if "random_control_delta" not in decision:
            raise ValueError("interaction decision missing random control")
        controls = decision.get("control_policy_scores", {})
        if not isinstance(controls, dict) or "no_op" not in controls or "random" not in controls:
            raise ValueError("interaction decision missing control policy scores")

    def score_records(
        self,
        genome: EvaluatorGenome,
        decisions: Sequence[Dict[str, object]],
        *,
        failure_rule_count: int = 1,
    ) -> List[float]:
        for decision in decisions:
            self.validate_decision_record(decision)
        return [score_decision(genome.active, decision, failure_rule_count=failure_rule_count) for decision in decisions]

    def validate_candidate(
        self,
        genome: EvaluatorGenome,
        decisions: Sequence[Dict[str, object]],
        *,
        candidate: EvaluatorCandidate | None = None,
        generation: int | None = None,
        failure_rule_count: int = 1,
    ) -> Dict[str, object]:
        if not decisions:
            raise ValueError("candidate validation requires at least one interaction decision")
        for decision in decisions:
            self.validate_decision_record(decision)
        selected = candidate or genome.propose(
            self.generation if generation is None else int(generation),
            failure_rule_count=failure_rule_count,
        )[0]
        return genome.validate_candidate(
            selected,
            decisions,
            generation=self.generation if generation is None else int(generation),
            failure_rule_count=failure_rule_count,
        )