Skip to content

gds_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.

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

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

    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

    def run(self) -> SweepResults:
        """Execute the sweep and return results."""
        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__,
        )

run()

Execute the sweep and return results.

Source code in packages/gds-psuu/gds_psuu/sweep.py 
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
def run(self) -> SweepResults:
    """Execute the sweep and return results."""
    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__,
    )