Source code for PlotDataRecorder

  1"""Records live experiment data to workspace log files for real-time plotting."""
  2
  3

[docs]
  4class PlotDataRecorder:
  5    """Tracks overall best fitness and writes data to log files consumed by PlotEvolutionLive."""

[docs]
  6    def __init__(self):
  7        """Initialize the recorder with zero overall best fitness."""
  8        self.__ovr_best_fit = 0

  9

[docs]
 10    def reset(self):
 11        """Reset overall best fitness to zero."""
 12        self.__ovr_best_fit = 0

 13

[docs]
 14    def record_waveform(self, waveform: list[int]):
 15        """Write waveform sample points to the waveform live-data log.
 16
 17        Parameters
 18        ----------
 19        waveform : list[int]
 20            Sampled waveform data points.
 21        """
 22        with open("workspace/waveformlivedata.log", "w+") as waveLive:
 23            i = 1
 24            for points in waveform:
 25                waveLive.write(str(i) + ", " + str(points) + "\n")
 26                i += 1

 27    

[docs]
 28    def record_generation(self, fits: list[float], current_epoch: int, diversity: float):
 29        """Append generation summary (best, worst, avg, diversity) to log files.
 30
 31        Parameters
 32        ----------
 33        fits : list[float]
 34            Fitness values for the current generation.
 35        current_epoch : int
 36            Current generation number.
 37        diversity : float
 38            Population diversity measure.
 39        """
 40        fits.sort(reverse=True)
 41        fitness_sum = 0
 42        for f in fits:
 43            fitness_sum += f
 44            if f > self.__ovr_best_fit:
 45                self.__ovr_best_fit = f
 46        
 47        with open("workspace/bestlivedata.log", "a") as liveFile:
 48            avg = fitness_sum / len(fits)
 49            # Format: Epoch, Best Fitness, Worst Fitness, Average Fitness, Ovr Best Fitness, Diversity Measure
 50            liveFile.write("{}, {}, {}, {}, {}, {}\n".format(
 51                str(current_epoch),
 52                str(fits[0]),
 53                str(fits[-1]),
 54                str(avg),
 55                str(self.__ovr_best_fit),
 56                str(diversity)
 57            ))
 58
 59        fit_strs = list(map(lambda x: str(x), fits))
 60        with open("workspace/violinlivedata.log", "a") as live_file:
 61            live_file.write(("{}:{}\n").format(current_epoch, ",".join(fit_strs)))

 62

[docs]
 63    def record_waveform_heatmap(self, current_epoch: int, best_waveform: list[int]):
 64        """Append the best waveform of this epoch to the heatmap log.
 65
 66        Parameters
 67        ----------
 68        current_epoch : int
 69            Current generation number.
 70        best_waveform : list[int]
 71            Waveform data of the best individual.
 72        """
 73        best_waveform_str = list(map(lambda x: str(x), best_waveform))
 74        with open("workspace/heatmaplivedata.log", "a") as live_file2:
 75            live_file2.write(("{}:{}\n").format(current_epoch, ",".join(best_waveform_str)))

 76

[docs]
 77    def record_all_live_data(self, index: int, reported_value: float, src_population: str):
 78        """Update a specific line in the all-live-data log with a new value.
 79
 80        Parameters
 81        ----------
 82        index : int
 83            Line index to update.
 84        reported_value : float
 85            Fitness or metric value to record.
 86        src_population : str
 87            Name of the source population.
 88        """
 89        # Read in the file contents first
 90        lines = []
 91        with open("workspace/alllivedata.log", "r") as allLive:
 92            lines = allLive.readlines()
 93
 94        # Modify the content internally
 95        if len(lines) <= index:
 96            for i in range(index - len(lines) + 1):
 97                lines.append("\n")
 98
 99        lines[index] = "{},{},{}\n".format(
100            str(index),
101            str(reported_value),
102            src_population
103        )
104
105        # Write these new lines to the file
106        with open("workspace/alllivedata.log", "w+") as allLive:
107            allLive.writelines(lines)