1"""
2Microcontroller.py
3------------------
4
5This file has a class that is used to interact with a microcontroller to extract information about the running circuit.
6This interacts with serial.
7
8.. todo::
9 Make a Mock of the Microcontroller to test standard operation without connection to the microcontroller.
10
11.. todo::
12 Make a Testing program that would allow you to directly get values from the micrcocontroller class. Maybe a terminal input or something would be good.
13
14.. todo::
15 Figure out how to mock Serial connection to allow this class to be tested.
16
17"""
18from serial import Serial
19from time import time
20import numpy as np
21
22from Circuit.CircuitLegacy import CircuitLegacy
23import typing
24
25from Config import Config
26from Logger import Logger
27
[docs]
28class Microcontroller:
29 """
30 This is a class that represents the microcontroller connected to the FPGA.
31 It is primarily used to interpret the serial output into values that are useful for the rest of the program.
32 It mostly deals with values for fitness functions.
33 """
34 def __log_event(self, level, *event):
35 self.__logger.log_event(level, *event)
36
37 def __log_info(self, level, *info):
38 self.__logger.log_info(level, *info)
39
40 def __log_error(self, level, *error):
41 self.__logger.log_error(level, *error)
42
43 def __log_warning(self, level, *warning):
44 self.__logger.log_warning(level, *warning)
45
[docs]
46 def __init__(self, config: Config, logger: Logger):
47 """
48 Initializes Microcontroller Object
49
50 Parameters
51 ----------
52 config : Config
53 Configuration object that determines what Microcontroller does.
54 logger : Logger
55 Logger object where this object stores its logging information.
56 """
57 self.__logger = logger
58 self.__config = config
59 if config.get_simulation_mode() == "FULLY_INTRINSIC" or config.get_simulation_mode() == "INTRINSIC_SENSITIVITY":
60 self.__log_event(1, "MCU SETTINGS ================================", config.get_usb_path(), config.get_serial_baud())
61 self.__serial = Serial(
62 config.get_usb_path(),
63 config.get_serial_baud(),
64 timeout=config.get_mcu_read_timeout()
65 )
66 self.__serial.dtr = False
67 if(config.reading_temp_humidity()):
68 self.__env_serial = Serial(
69 config.get_env_usb_path(),
70 config.get_serial_baud(),
71 timeout=config.get_mcu_read_timeout()
72 )
73 self.__env_serial.dtr = False
74 self.__fpga = config.get_fpga()
75
[docs]
76 def switch_fpga(self):
77 """
78 If there are multiple FPGAs, switch which FPGA the microcontroller is acting on and reading from.
79
80 .. todo::
81 Allyn, Is this the correct interpretation of what this does? Do you want to add additional detail?
82 Please also check that the other multi-fpga logic is properly represented as well.
83 """
84 self.__serial.reset_input_buffer()
85 self.__serial.reset_output_buffer()
86 self.__serial.write(b'4')
87 if self.__fpga == self.__config.get_fpga():
88 self.__log_event(2, "Switching to FPGA 2")
89 self.__fpga = self.__config.get_fpga2()
90 else :
91 self.__log_event(2, "Switching to FPGA 1")
92 self.__fpga = self.__config.get_fpga()
93 self.__log_event(2, "Done switching FPGAs")
94
[docs]
95 def get_fpga(self):
96 """
97 Get __fpga string
98
99 .. todo::
100 Allyn, another one for you. What is this?
101
102 Returns
103 -------
104 str
105 FPGA Identifier
106 """
107 return self.__fpga
108
[docs]
109 def simple_measure_pulses(self, data_filepath):
110 """
111 This measure pulses function will poll the MCU,
112 and just put the raw pulse count recorded into the circuit's data file
113
114 Parameters
115 ----------
116 circuit : Circuit
117 The circuit we will measure pulses of.
118 """
119 data_file = open(data_filepath, "w")
120 lines = []
121 buf = []
122 # Poll serial line until START signal
123 self.__log_event(3, f"Starting loop for reading")
124
125 self.__serial.reset_input_buffer()
126 self.__serial.reset_output_buffer()
127 # NOTE The MCU is expecting a string '1' if fitness isn't measured this may be why
128 self.__serial.write(b'1')
129 start = time()
130 self.__log_event(3, f"Starting MCU loop...")
131
132 max_attempts = 5
133 attempts = 0
134 while True:
135 attempts = attempts + 1
136 self.__log_event(3, f"Serial reading...")
137 p = self.__serial.read_until()
138 self.__log_event(3, f"Serial read done")
139 if (time() - start) >= self.__config.get_mcu_read_timeout():
140 self.__log_warning(1, f"Time Exceeded")
141 if attempts >= max_attempts:
142 self.__log_warning(3, f"Exceeded max attempts ({max_attempts}). Halting MCU reading")
143 buf.append(-1)
144 break
145 # TODO We should be able to do whatever this line does better
146 # This is currently doing a poor job at REGEXing the MCU serial return - can be done better
147 # It's supposed to handle exceptions from transmission loss (i.e. dropped or additional spaces, shifted colons, etc)
148 self.__log_event(3, "Pulled", p, f"from MCU")
149 if (p != b"" and b":" not in p and b"START" not in p and b"FINISH" not in p and b" " not in p):
150 p = p.translate(None, b"\r\n")
151 buf.append(p)
152 break
153
154 end = time() - start
155
156 # if the transfer interval is "SAMPLE", switch to the other fpga between samples
157 # if self.__config.get_transfer_sample():
158 # self.switch_fpga()
159
160 # buf now has `samples` entries
161 self.__log_event(2, 'Length of buffer:', len(buf))
162 if len(buf) == 0:
163 buf.append(-1000) # This should never happen
164 for i in range(len(buf)):
165 self.__log_event(2, f'Buffer entry {i}:', buf[i])
166 try:
167 buf[i] = int(buf[i])
168 except ValueError:
169 buf[i] = -1
170 lines.append(str(buf[i]) + "\n")
171
172 data_file.writelines(lines)
173 data_file.close()
174
[docs]
175 def measure_pulses(self, circuit: CircuitLegacy):
176 """
177 Measures the number of pulses generated by the circuit provided.
178
179 Parameters
180 ----------
181 circuit : Circuit
182 The circuit we are measuring pulses of
183 """
184 samples = 2
185 if self.__config.get_simulation_mode() == "INTRINSIC_SENSITIVITY":
186 samples = 1
187
188 # TODO Use pathlib here
189 # Begin monitoring on load
190 data_file = open(circuit.get_data_filepath(), "wb")
191
192 buf = []
193 for i in range(0,samples):
194 # Poll serial line until START signal
195 self.__log_event(3, "Starting loop for reading")
196
197 self.__serial.reset_input_buffer()
198 self.__serial.reset_output_buffer()
199 # NOTE The MCU is expecting a string '1' if fitness isn't measured this may be why
200 self.__serial.write(b'1')
201 start = time()
202 self.__log_event(3, "Starting MCU loop...")
203
204 while True:
205 self.__log_event(3, "Serial reading...")
206 p = self.__serial.read_until()
207 self.__log_event(3, "Serial read done")
208 if (time() - start) >= self.__config.get_mcu_read_timeout():
209 self.__log_warning(1, "Time Exceeded. Halting MCU Reading")
210 buf.append(0)
211 break
212 # TODO We should be able to do whatever this line does better
213 # This is currently doing a poor job at REGEXing the MCU serial return - can be done better
214 # It's supposed to handle exceptions from transmission loss (i.e. dropped or additional spaces, shifted colons, etc)
215 self.__log_event(3, "Pulled", p, "from MCU")
216 if (p != b"" and b":" not in p and b"START" not in p and b"FINISH" not in p and b" " not in p):
217 p = p.translate(None, b"\r\n")
218 buf.append(p)
219 break
220
221 end = time() - start
222
223 buf_dif = 0
224 weighted_count = int(buf[0])
225
226 for i in range(0, len(buf)):
227 if buf[i] == b'':
228 buf[i] = 0
229 else:
230 buf[i] = int(buf[i])
231 if i + 1 < len(buf):
232 buf_dif = abs(int(buf[i]) - int(buf[i+1]))
233
234 if len(buf) > 0:
235 if samples > 1:
236 weighted_count = abs(buf[0] - buf_dif)
237 data_file.write(bytes(str(weighted_count) + "\n", "utf-8"))
238
239 if len(buf) > 0:
240 freq = sum(buf)/len(buf)
241 else:
242 freq = 0.0
243
244 self.__log_event(2, "Length of Buffer:", len(buf))
245 self.__log_event(2, "Number Pulses:", sum(buf))
246 self.__log_event(2, "Average Frequency: ~", freq, "Hz")
247 self.__log_event(2, "Sampling Duration:", end)
248 self.__log_event(2, "Completed writing to data file")
249
250 data_file.close()
251
[docs]
252 def measure_signal(self, data_filepath):
253 """
254 Measures the signal, writing the waveform data to the provided data file
255
256 .. todo::
257 Preexisting todo: This whole section can probably be optimized.
258
259 Parameters
260 ----------
261 circuit : Circuit
262 The circuit we are measuring the signal of
263 """
264 buf = []
265
266 # Begin monitoring on load
267 data_file = open(data_filepath, "wb")
268
269 self.__serial.reset_input_buffer()
270 self.__serial.reset_output_buffer()
271 self.__log_event(1, "Reading microcontroller.")
272 # The MCU is expecting a string '2' to initiate the ADC capture from the FPGA (waveform as opposed to pulses)
273 self.__serial.write(b'2')
274 line = self.__serial.read()
275
276 start = time()
277
278 # The MCU returns a START line followed by many lines of data (500 currently) followed by a FINISHED line
279 while b"START\n" not in line:
280 self.__serial.write(b'2')
281 line = self.__serial.read_until()
282
283 if (time() - start) >= self.__config.get_mcu_read_timeout():
284 self.__log_warning(1, "Did not read START from MCU")
285 self.__log_warning(1, "Time Exceeded. Halting MCU Reading.")
286 break
287
288 # TODO This whole section can probably be optimized
289 # Reads in 500 samples from MCU, with each being 10microseconds apart
290 # Then, dumps into a file
291 while (b"FINISHED\n" not in line):
292 line = self.__serial.read_until()
293 if line != b"\n" and line != b"START\n" and line != b"FINISHED\n" and line != b"FINISHED\n":
294 buf.append(line)
295 if (time() - start) >= self.__config.get_mcu_read_timeout():
296 self.__log_warning(1, "Time Exceeded. Halting MCU Reading.")
297 break
298
299 self.__log_event(2, "Finished reading microcontroller. Logging data to file.")
300
301 for i in buf:
302 if b"FINISHED" not in i:
303 data_file.write(bytes(i))
304
305 data_file.close()
306 self.__log_event(2, "Completed writing to data file")
307
[docs]
308 def measure_signal_td(self, data_filepath):
309 """
310 Measures (1) the FPGA waveform directly from FPGA output pin and (2) the "state"/frequency waveform
311 directly from the signal-generating Nano. Writes 1000 sample points' data to a file.
312
313 .. todo::
314 Preexisting todo: This whole section can probably be optimized.
315
316 Parameters
317 ----------
318 circuit : Circuit
319 The circuit we are measuring the signal of
320 """
321 # TODO This whole section can probably be optimized
322
323 buf = []
324
325 # Begin monitoring on load
326 data_file = open(data_filepath, "wb")
327
328 self.__serial.reset_input_buffer()
329 self.__serial.reset_output_buffer()
330 self.__log_event(1, "Reading microcontroller.")
331 # The MCU is expecting a string '5' to initiate the ADC capture from the FPGA (waveform & state as opposed to pulses)
332 self.__serial.write(b'5')
333 line = self.__serial.read()
334
335 start = time()
336
337 # The MCU returns a START line followed by many lines of data (1000 currently) followed by a FINISHED line
338 while b"START\n" not in line:
339 # Avoid spamming serial link. Experiment works without this.
340 # self.__serial.write(b'5')
341 line = self.__serial.read_until()
342 if (time() - start) >= self.__config.get_mcu_read_timeout():
343 self.__log_warning(1, "Did not read START from MCU")
344 self.__log_warning(1, "Time Exceeded. Halting MCU Reading.")
345 break
346
347 # TODO This whole section can probably be optimized
348 # Reads in 1000 samples from MCU, with each being 2.5 ms apart
349 # Then, dumps into a file
350 while (b"FINISHED\n" not in line):
351 line = self.__serial.read_until()
352 if line != b"\n" and line != b"START\n" and line != b"FINISHED\n" and line != b"FINISHED\n":
353 buf.append(line)
354 if (time() - start) >= self.__config.get_mcu_read_timeout():
355 self.__log_warning(1, "Time Exceeded. Halting MCU Reading.")
356 break
357
358 self.__log_event(2, "Finished reading microcontroller. Logging data to file.")
359
360 for i in buf:
361 if b"FINISHED" not in i:
362 data_file.write(bytes(i))
363
364 data_file.close()
365 self.__log_event(2, "Completed writing to data file")
366
367
[docs]
368 def measure_temp(self):
369 """
370 Measures the temperature using a DHT22 sensor conected to the Arduino.
371 """
372 self.__log_event(3, "Measuring temperature")
373
374 self.__env_serial.reset_input_buffer()
375 self.__env_serial.reset_output_buffer()
376 self.__env_serial.write(b'5')
377 start = time()
378
379 self.__log_event(3, "Serial reading...")
380 p = self.__env_serial.read_until()
381 self.__log_event(3, "Serial read done")
382 if (time() - start) >= self.__config.get_mcu_read_timeout():
383 self.__log_warning(1, "Time Exceeded. Halting MCU Reading of temperature")
384 return -1;
385 # TODO We should be able to do whatever this line does better
386 # This is currently doing a poor job at REGEXing the MCU serial return - can be done better
387 # It's supposed to handle exceptions from transmission loss (i.e. dropped or additional spaces, shifted colons, etc)
388 self.__log_event(3, "Pulled", p, "from MCU")
389 if (p != b"" and b":" not in p and b"START" not in p and b"FINISH" not in p and b" " not in p):
390 p = p.translate(None, b"\r\n")
391 return(float(p))
392
[docs]
393 def measure_humidity(self):
394 """
395 Measures the humidity using a DHT22 sensor conected to the Arduino.
396 """
397 self.__log_event(3, "Measuring humidity")
398
399 self.__env_serial.reset_input_buffer()
400 self.__env_serial.reset_output_buffer()
401 self.__env_serial.write(b'6')
402 start = time()
403
404 self.__log_event(3, "Serial reading...")
405 p = self.__env_serial.read_until()
406 self.__log_event(3, "Serial read done")
407 if (time() - start) >= self.__config.get_mcu_read_timeout():
408 self.__log_warning(1, "Time Exceeded. Halting MCU Reading of humidity")
409 return -1;
410 # TODO We should be able to do whatever this line does better
411 # This is currently doing a poor job at REGEXing the MCU serial return - can be done better
412 # It's supposed to handle exceptions from transmission loss (i.e. dropped or additional spaces, shifted colons, etc)
413 self.__log_event(3, "Pulled", p, "from MCU")
414 if (p != b"" and b":" not in p and b"START" not in p and b"FINISH" not in p and b" " not in p):
415 p = p.translate(None, b"\r\n")
416 return(float(p))
417
418
419
420
421