Source code for EvaluateFitness.EvalPulseCountFitness

 1"""Oscillation pulse-count fitness evaluator.
 2
 3Fitness is computed as the inverse distance from a target pulse count.
 4Used with :class:`~EvaluateFitness.EvaluateFitness.EvaluateFitness` via
 5the :class:`~EvaluateFitness.EvaluateFitness.FitnessEvaluator` protocol::
 6
 7    evaluator = EvalPulseCountFitness(target=1000, plot_data_recorder=recorder)
 8    EvaluateFitness(evaluator)
 9"""
10
11from PlotDataRecorder import PlotDataRecorder
12
13

[docs]
14class EvalPulseCountFitness:
15    """Evaluate fitness based on how close measured pulse counts are to a target.
16
17    For each individual the minimum fitness across all samples is used,
18    where fitness = 1 when pulses == target, 0 when pulses == 0, and
19    ``1 / abs(target - pulses)`` otherwise.
20    """

[docs]
21    def __init__(self, target: int, plot_data_recorder: PlotDataRecorder):
22        """Create an evaluator targeting *target* pulses."""
23        self.__target = target
24        self.__plot_data_recorder = plot_data_recorder

25

[docs]
26    def start_eval(self):
27        """Called before evaluating a generation (no-op for pulse count)."""
28        pass

29

[docs]
30    def end_eval(self):
31        """Called after evaluating a generation (no-op for pulse count)."""
32        pass

33

[docs]
34    def calculate_success(self, data: list[int], index: int, src_pop: str) -> float:
35        """Return the minimum fitness across all pulse-count samples in *data*."""
36        acc: list[float] = []
37        for p in data:
38            acc.append(self.__calculate_individual(p))
39        fit = min(acc)
40
41        min_idx = acc.index(fit)
42        pulses_at_min = data[min_idx]
43
44        self.__plot_data_recorder.record_all_live_data(index, pulses_at_min, src_pop)
45
46        return fit

47    

[docs]
48    def calculate_error(self, err: Exception, index: int, src_pop: str) -> float:
49        """Return zero fitness when measurement fails."""
50        self.__plot_data_recorder.record_all_live_data(index, -1, src_pop)
51        return 0

52    
53    def __calculate_individual(self, pulses: int) -> float:
54        if pulses == self.__target:
55            return 1
56        elif pulses == 0:
57            return 0
58        else:
59            return 1.0 / abs(self.__target - pulses)