Skip to content

gds_analysis.psuu.sweep

Sweep orchestrator -- the main entry point for parameter search.

Sweep orchestrator — the main entry point for parameter search.

Sweep

Bases: BaseModel

Orchestrates parameter space search.

Drives the optimizer suggest/observe loop, delegating evaluation to the Evaluator which bridges to gds-sim.

If parameter_schema is provided, the sweep validates the parameter space against the declared schema before starting. Validation errors raise ValueError.

Source code in packages/gds-analysis/gds_analysis/psuu/sweep.py 
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
class Sweep(BaseModel):
    """Orchestrates parameter space search.

    Drives the optimizer suggest/observe loop, delegating evaluation
    to the Evaluator which bridges to gds-sim.

    If ``parameter_schema`` is provided, the sweep validates the
    parameter space against the declared schema before starting.
    Validation errors raise ``ValueError``.
    """

    model_config = ConfigDict(arbitrary_types_allowed=True, frozen=True)

    model: Model
    space: ParameterSpace
    kpis: list[KPI]
    optimizer: Optimizer
    objective: Objective | None = None
    timesteps: int = 100
    runs: int = 1
    parameter_schema: Any = None

    def run(self) -> SweepResults:
        """Execute the sweep and return results."""
        if self.parameter_schema is not None:
            self._validate_schema()

        kpi_names = [k.name for k in self.kpis]
        self.optimizer.setup(self.space, kpi_names)

        evaluator = Evaluator(
            base_model=self.model,
            kpis=list(self.kpis),
            timesteps=self.timesteps,
            runs=self.runs,
        )

        evaluations: list[EvaluationResult] = []
        while not self.optimizer.is_exhausted():
            params = self.optimizer.suggest()
            result = evaluator.evaluate(params)
            self.optimizer.observe(params, result.scores)
            evaluations.append(result)

        return SweepResults(
            evaluations=evaluations,
            kpi_names=kpi_names,
            optimizer_name=type(self.optimizer).__name__,
        )

    def _validate_schema(self) -> None:
        """Validate the parameter space against the declared schema."""
        from gds.verification.findings import Severity

        from gds_analysis.psuu.checks import check_parameter_space_compatibility

        findings = check_parameter_space_compatibility(
            self.space, self.parameter_schema
        )
        errors = [f for f in findings if f.severity == Severity.ERROR]
        if errors:
            messages = "; ".join(f.message for f in errors)
            raise ValueError(f"Parameter space violates declared schema: {messages}")

run()

Execute the sweep and return results.

Source code in packages/gds-analysis/gds_analysis/psuu/sweep.py 
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
def run(self) -> SweepResults:
    """Execute the sweep and return results."""
    if self.parameter_schema is not None:
        self._validate_schema()

    kpi_names = [k.name for k in self.kpis]
    self.optimizer.setup(self.space, kpi_names)

    evaluator = Evaluator(
        base_model=self.model,
        kpis=list(self.kpis),
        timesteps=self.timesteps,
        runs=self.runs,
    )

    evaluations: list[EvaluationResult] = []
    while not self.optimizer.is_exhausted():
        params = self.optimizer.suggest()
        result = evaluator.evaluate(params)
        self.optimizer.observe(params, result.scores)
        evaluations.append(result)

    return SweepResults(
        evaluations=evaluations,
        kpi_names=kpi_names,
        optimizer_name=type(self.optimizer).__name__,
    )