1"""Logging and monitoring for evolutionary experiments.
2
3Provides console and file-based logging with ANSI color formatting,
4live plot launching, and workspace archival for experiment results.
5"""
6from dataclasses import dataclass
7from pathlib import Path
8from sys import stdout
9from datetime import datetime
10from subprocess import CalledProcessError, run
11from os.path import exists
12from os.path import join
13from os import mkdir
14from shutil import copytree
15from shutil import rmtree
16from datetime import datetime
17
18# The window dimensions
19LINE_WIDTH = 112
20WIN_DIM="105x29"
21DOUBLE_HLINE = "=" * LINE_WIDTH
22
23MONITOR_FILE = None
24
25TERM_CMD=["gnome-terminal", "--geometry={}".format(WIN_DIM), "--"]
26
27# The time between animation frames in milliseconds
28FRAME_DELAY = 200
29
30# The spacing between matplotlib subplots as a percentage of the average
31# axis height
32SUBPLOT_SPACING = 0.5
33
34# Formatting constants
35HEADER = '\033[95m'
36OKBLUE = '\033[94m'
37OKGREEN = '\033[92m'
38WARNING = '\033[93m'
39FAIL = '\033[91m'
40ENDC = '\033[0m'
41BOLD = '\033[1m'
42UNDERLINE = '\033[4m'
43
44README_FILE_HEADER = "FPGA/MCU [1] \n"
45
[docs]
46@dataclass
47class LoggerConfig:
48 """Configuration for Logger output, plot launching, and file paths.
49
50 Consumed by ``Logger.__init__`` to control verbosity, plot behavior,
51 and workspace log destinations.
52 """
53
54 plots_dir: Path
55 launch_plots: bool
56 is_sensitivity: bool
57 frame_interval: int
58 log_file: Path
59 log_level: int
60 save_log: bool
61 datetime_format: str
62
63# TODO Utilize Python logging library
[docs]
64class Logger:
65 """Level-gated logger with monitor file output and live plot integration.
66
67 Writes timestamped entries to both stdout and a monitor file, supports
68 ANSI-colored severity prefixes, and can launch gnome-terminal-based
69 live plotting windows for experiment visualization.
70 """
71 def __init_monitor(self):
72 """Initialize the monitor file header, plots directory, and live plot windows.
73
74 Writes the experiment banner to the monitor file, resets the plots
75 directory, and optionally launches a live plotting terminal.
76 """
77 # Start the monitor
78 # self.log_event(1, "Creating the monitor file...")
79
80 self.log_monitor(1, "{}{}".format(
81 "Evolutionary Experiment Monitor".center(LINE_WIDTH),
82 "\n"
83 ))
84 self.log_monitor("", "{}".format(DOUBLE_HLINE))
85 # self.log_monitor(1, "Parameters and updates load during circuit evaluation")
86 # self.log_monitor(1, ".\n" * 23)
87 self.log_monitor("", str(self.__experiment_explanation))
88 self.log_monitor("", "{}".format(DOUBLE_HLINE))
89 # self.log_monitor("", self.__config.get_raw_data())
90 self.log_monitor("", "{}".format(DOUBLE_HLINE))
91 self.__monitor_file.flush()
92
93 # args = TERM_CMD + ["python3", "src/Monitor.py"]
94 # try:
95 # run(args, check=True, capture_output=True)
96 # except OSError as e:
97 # self.log_error(1, "An error occured while launching Monitor.py")
98 # except CalledProcessError as e:
99 # self.log_error(1, "An error occured in Monitor.py")
100
101 #set up directory for saving files
102 plots_dir = self.__config.plots_dir
103 try:
104 rmtree(plots_dir)
105 except OSError as error:
106 print(error)
107
108 if not plots_dir.exists():
109 plots_dir.mkdir()
110
111 if self.__config.launch_plots:
112 if self.__config.is_sensitivity:
113 args = TERM_CMD + ["python3", "src/PlotSensitivityLive.py"]
114 else:
115 args = TERM_CMD + ["python3", "src/PlotEvolutionLive.py", "--frame-interval", str(self.__config.frame_interval)]
116
117 try:
118 run(args, check=True, capture_output=True)
119 except OSError as e:
120 self.log_error(1, "An error occured while launching PlotEvolutionLive.py")
121 except CalledProcessError as e:
122 self.log_error(1, "An error occured in PlotEvolutionLive.py")
123 self.log_error(1, e)
124
[docs]
125 def __init__(self, explanation: str, logger_config: LoggerConfig):
126 """Set up log files, clear workspace data logs, and start the monitor.
127
128 Parameters
129 ----------
130 explanation : str
131 Human-readable description of the experiment.
132 logger_config : LoggerConfig
133 Configuration controlling log paths, verbosity, and plot behavior.
134 """
135 self.__monitor_file = open(logger_config.log_file, "w")
136 self.__log_file = stdout
137 self.__config = logger_config
138 self.__experiment_explanation = explanation
139
140 # Ensure the logs exists and have been cleared. Not happy with
141 # this method, but couldn't find a better way to do it.
142 open("workspace/alllivedata.log", "w").close()
143 open("workspace/bestlivedata.log", "w").close()
144 open("workspace/waveformlivedata.log", "w").close()
145 open("workspace/maplivedata.log", "w").close()
146 open("workspace/heatmaplivedata.log", "w").close()
147 open("workspace/pulselivedata.log", "w").close()
148 open("workspace/violinlivedata.log", "w").close()
149 open("workspace/poplivedata.log", "w").close()
150 open("workspace/randomizationdata.log", "w").close()
151 open("workspace/fitnesssensitivity.log", "w").close()
152 open("workspace/bitstream_avg.log", "w").close()
153 if not exists("workspace/template"):
154 mkdir("workspace/template")
155
156 if exists("workspace/plots"):
157 rmtree("workspace/plots")
158 if not exists("workspace/plots"):
159 mkdir("workspace/plots")
160 # Determine if we need to the to initialize the analysis and
161 # if so, do so.
162 # currently removed since we're not currently storing any data, so there's a bunch of empty files and directories
163 # if explanation != "test":
164 # self.__init_analysis()
165
166 # Determine whether we need to launch the monitor and launch it
167 # if so.
168 # if config.get_launch_monitor():
169 # self.__init_monitor()
170 self.__init_monitor()
171
[docs]
172 def log_generation(self, population, epoch_time):
173 """Log a generation summary including current and overall best fitness.
174
175 Parameters
176 ----------
177 population : Population
178 The population whose best circuits are reported.
179 epoch_time : float
180 Wall-clock seconds elapsed during the epoch.
181 """
182 self.log_event(2, DOUBLE_HLINE)
183 self.log_event(2, DOUBLE_HLINE)
184 self.log_event(2, DOUBLE_HLINE)
185
186 current_best_circuit = population.get_current_best_circuit()
187 overall_best_circuit = population.get_overall_best_circuit_info()
188
189 self.log_event(2, "CURRENT BEST: {} : EPOCH {} : FITNESS {}".format(
190 str(overall_best_circuit.name),
191 str(population.get_best_epoch()),
192 str(overall_best_circuit.fitness)
193 ))
194
195 self.log_event(2, "HIGHEST FITNESS OF EPOCH {} IS: {} = {} over {} seconds".format(
196 str(population.get_current_epoch()),
197 str(current_best_circuit),
198 str(current_best_circuit.get_fitness()),
199 str(epoch_time)
200 ))
201
202 self.log_event(2, DOUBLE_HLINE)
203 self.log_event(2, DOUBLE_HLINE)
204 self.log_event(2, DOUBLE_HLINE)
205
[docs]
206 def log_monitor(self, prefix, *msg):
207 """Write a timestamped message directly to the monitor log file."""
208 if self.__config.save_log:
209 now = datetime.now()
210 print(now, prefix, *msg, file=self.__monitor_file)
211
[docs]
212 def log_event(self, level, *msg):
213 """Log a general event to stdout and the monitor file if level permits."""
214 if self.__config.log_level >= level:
215 print(*msg, file=self.__log_file)
216 self.log_monitor("", *msg)
217
[docs]
218 def log_info(self, level, *msg):
219 """Log an informational message with blue ANSI formatting."""
220 if self.__config.log_level >= level:
221 print("INFO: ", OKBLUE, *msg, ENDC, file=self.__log_file)
222 self.log_monitor("INFO: ", *msg)
223
[docs]
224 def log_warning(self, level, *msg):
225 """Log a warning message with yellow ANSI formatting."""
226 if self.__config.log_level >= level:
227 print("WARNING: ", WARNING, *msg, ENDC, file=self.__log_file)
228 self.log_monitor("WARNING: ", *msg)
229
[docs]
230 def log_error(self, level, *msg):
231 """Log an error message with red ANSI formatting."""
232 if self.__config.log_level >= level:
233 print("ERROR: ", FAIL, *msg, ENDC, file=self.__log_file)
234 self.log_monitor("ERROR: ", *msg)
235
[docs]
236 def log_critical(self, level, *msg):
237 """Log a critical message with red ANSI formatting."""
238 if self.__config.log_level >= level:
239 print("CRITICAL: ", FAIL, *msg, ENDC, file=self.__log_file)
240 self.log_monitor("CRITICAL: ", *msg)
241
[docs]
242 def save_workspace(self, directory):
243 """Close the monitor file and archive the workspace to a timestamped directory."""
244 self.__monitor_file.close()
245 current_time = str(datetime.now().strftime(self.__config.datetime_format))
246 current_time = current_time.replace('/', '-')
247 copytree("./workspace", join(directory, current_time))
