Formatted code using the Black formatter - https://black.readthedocs.io

Resolves issue 82:  #82

Author: ThomasAkam

com/data_logger.py

@@ -3,8 +3,9 @@
 from datetime import datetime
 from shutil import copyfile
 
-class Data_logger():
-    '''Class for logging data from a pyControl setup to disk'''
+
+class Data_logger:
+    """Class for logging data from a pyControl setup to disk"""
 
     def __init__(self, sm_info=None, print_func=None, data_consumers=[]):
         self.data_file = None
@@ -16,55 +17,58 @@ def __init__(self, sm_info=None, print_func=None, data_consumers=[]):
 
     def set_state_machine(self, sm_info):
         self.sm_info = sm_info
-        self.ID2name_fw = self.sm_info['ID2name'] # Dict mapping framework IDs to names.
-        
+        self.ID2name_fw = self.sm_info["ID2name"]  # Dict mapping framework IDs to names.
+
     def open_data_file(self, data_dir, experiment_name, setup_ID, subject_ID, datetime_now=None):
-        '''Open file tsv/txt file for event data and write header information.
-        If state machine uses analog inputs instantiate analog data writers.'''
+        """Open file tsv/txt file for event data and write header information.
+        If state machine uses analog inputs instantiate analog data writers."""
         self.data_dir = data_dir
         self.experiment_name = experiment_name
         self.subject_ID = subject_ID
         self.setup_ID = setup_ID
-        if datetime_now is None: datetime_now = datetime.now()
+        if datetime_now is None:
+            datetime_now = datetime.now()
         self.end_timestamp = -1
-        file_name = self.subject_ID + datetime_now.strftime('-%Y-%m-%d-%H%M%S') + '.tsv'
+        file_name = self.subject_ID + datetime_now.strftime("-%Y-%m-%d-%H%M%S") + ".tsv"
         self.file_path = os.path.join(self.data_dir, file_name)
-        self.data_file = open(self.file_path, 'w', newline = '\n')
-        self.data_file.write(self.tsv_row_str( # Write header with row names.
-            rtype='type', time='time', name='name', value='value'))
-        self.write_info_line('experiment_name', self.experiment_name)
-        self.write_info_line('task_name', self.sm_info['name'])
-        self.write_info_line('task_file_hash', self.sm_info['task_hash'])
-        self.write_info_line('setup_id', self.setup_ID)
-        self.write_info_line('framework_version', self.sm_info['framework_version'])
-        self.write_info_line('micropython_version', self.sm_info['micropython_version'])
-        self.write_info_line('subject_id', self.subject_ID)
-        self.write_info_line('start_time', datetime.utcnow().isoformat(timespec='milliseconds'))
-        self.analog_writers = {ID: 
-            Analog_writer(ai['name'], ai['fs'], ai['dtype'], self.file_path)
-            for ID, ai in self.sm_info['analog_inputs'].items()}
+        self.data_file = open(self.file_path, "w", newline="\n")
+        self.data_file.write(
+            self.tsv_row_str(rtype="type", time="time", name="name", value="value")  # Write header with row names.
+        )
+        self.write_info_line("experiment_name", self.experiment_name)
+        self.write_info_line("task_name", self.sm_info["name"])
+        self.write_info_line("task_file_hash", self.sm_info["task_hash"])
+        self.write_info_line("setup_id", self.setup_ID)
+        self.write_info_line("framework_version", self.sm_info["framework_version"])
+        self.write_info_line("micropython_version", self.sm_info["micropython_version"])
+        self.write_info_line("subject_id", self.subject_ID)
+        self.write_info_line("start_time", datetime.utcnow().isoformat(timespec="milliseconds"))
+        self.analog_writers = {
+            ID: Analog_writer(ai["name"], ai["fs"], ai["dtype"], self.file_path)
+            for ID, ai in self.sm_info["analog_inputs"].items()
+        }
 
     def write_info_line(self, name, value, time=0):
-        self.data_file.write(self.tsv_row_str('info', time=time, name=name, value=value))
+        self.data_file.write(self.tsv_row_str("info", time=time, name=name, value=value))
 
-    def tsv_row_str(self, rtype, time='', name='', value=''):
-        time_str = f'{time/1000:.3f}' if type(time) == int else time
-        return f'{time_str}\t{rtype}\t{name}\t{value}\n'
+    def tsv_row_str(self, rtype, time="", name="", value=""):
+        time_str = f"{time/1000:.3f}" if type(time) == int else time
+        return f"{time_str}\t{rtype}\t{name}\t{value}\n"
 
-    def copy_task_file(self, data_dir, tasks_dir, dir_name='task_files'):
-        '''If not already present, copy task file to data_dir/dir_name
-        appending the files djb2 hash to the file name.'''
+    def copy_task_file(self, data_dir, tasks_dir, dir_name="task_files"):
+        """If not already present, copy task file to data_dir/dir_name
+        appending the files djb2 hash to the file name."""
         exp_tasks_dir = os.path.join(data_dir, dir_name)
         if not os.path.exists(exp_tasks_dir):
             os.mkdir(exp_tasks_dir)
-        task_file_path = os.path.join(tasks_dir, self.sm_info['name']+'.py')
-        task_save_name = os.path.split(self.sm_info['name'])[1] +'_{}.py'.format(self.sm_info['task_hash'])
+        task_file_path = os.path.join(tasks_dir, self.sm_info["name"] + ".py")
+        task_save_name = os.path.split(self.sm_info["name"])[1] + "_{}.py".format(self.sm_info["task_hash"])
         if not task_save_name in os.listdir(exp_tasks_dir):
             copyfile(task_file_path, os.path.join(exp_tasks_dir, task_save_name))
-            
+
     def close_files(self):
         if self.data_file:
-            self.write_info_line('end_time', self.end_datetime.isoformat(timespec='milliseconds'), self.end_timestamp)
+            self.write_info_line("end_time", self.end_datetime.isoformat(timespec="milliseconds"), self.end_timestamp)
             self.data_file.close()
             self.data_file = None
             self.file_path = None
@@ -73,12 +77,12 @@ def close_files(self):
         self.analog_writers = {}
 
     def process_data(self, new_data):
-        '''If data _file is open new data is written to file.  If print_func is specified
-        human readable data strings are passed to it.'''
+        """If data _file is open new data is written to file.  If print_func is specified
+        human readable data strings are passed to it."""
         if self.data_file:
             self.write_to_file(new_data)
         if self.print_func:
-            self.print_func(self.data_to_string(new_data).replace('\t\t', '\t'), end='')
+            self.print_func(self.data_to_string(new_data).replace("\t\t", "\t"), end="")
         if self.data_consumers:
             for data_consumer in self.data_consumers:
                 data_consumer.process_data(new_data)
@@ -89,35 +93,35 @@ def write_to_file(self, new_data):
             self.data_file.write(data_string)
             self.data_file.flush()
         for nd in new_data:
-            if nd.type == 'A':
+            if nd.type == "A":
                 self.analog_writers[nd.ID].save_analog_chunk(timestamp=nd.time, data_array=nd.data)
 
     def data_to_string(self, new_data):
-        '''Convert list of data tuples into a string.  If verbose=True state and event names are used,
-        if verbose=False state and event IDs are used.'''
-        data_string = ''
+        """Convert list of data tuples into a string.  If verbose=True state and event names are used,
+        if verbose=False state and event IDs are used."""
+        data_string = ""
         for nd in new_data:
-            if nd.type == 'D':  # State entry or event.
-                if nd.ID in self.sm_info['states'].values():
-                    data_string += self.tsv_row_str('state', time=nd.time, name=self.ID2name_fw[nd.ID])
+            if nd.type == "D":  # State entry or event.
+                if nd.ID in self.sm_info["states"].values():
+                    data_string += self.tsv_row_str("state", time=nd.time, name=self.ID2name_fw[nd.ID])
                 else:
-                    data_string += self.tsv_row_str('event', time=nd.time, name=self.ID2name_fw[nd.ID])
-            elif nd.type == 'P': # User print output.
-                data_string += self.tsv_row_str('print', time=nd.time, value=nd.data)
-            elif nd.type == 'V': # Variable.
-                data_string += self.tsv_row_str('variable', time=nd.time, name=nd.ID, value=nd.data)
-            elif nd.type == '!': # Warning
-                data_string += self.tsv_row_str('warning', value=nd.data)
-            elif nd.type == '!!': # Error
-                data_string += self.tsv_row_str('error', value=nd.data.replace('\n','|').replace('\r','|'))
-            elif nd.type == 'S': # Framework stop.
+                    data_string += self.tsv_row_str("event", time=nd.time, name=self.ID2name_fw[nd.ID])
+            elif nd.type == "P":  # User print output.
+                data_string += self.tsv_row_str("print", time=nd.time, value=nd.data)
+            elif nd.type == "V":  # Variable.
+                data_string += self.tsv_row_str("variable", time=nd.time, name=nd.ID, value=nd.data)
+            elif nd.type == "!":  # Warning
+                data_string += self.tsv_row_str("warning", value=nd.data)
+            elif nd.type == "!!":  # Error
+                data_string += self.tsv_row_str("error", value=nd.data.replace("\n", "|").replace("\r", "|"))
+            elif nd.type == "S":  # Framework stop.
                 self.end_datetime = datetime.utcnow()
                 self.end_timestamp = nd.time
         return data_string
 
 
-class Analog_writer():
-    '''Class for writing data from one analog input to disk.'''
+class Analog_writer:
+    """Class for writing data from one analog input to disk."""
 
     def __init__(self, name, sampling_rate, data_type, session_filepath):
         self.name = name
@@ -127,36 +131,35 @@ def __init__(self, name, sampling_rate, data_type, session_filepath):
 
     def open_data_files(self, session_filepath):
         ses_path_stem, file_ext = os.path.splitext(session_filepath)
-        self.path_stem = ses_path_stem + f'_{self.name}'
-        self.t_tempfile_path = self.path_stem + '.time.temp'
-        self.d_tempfile_path = self.path_stem + f'.data-1{self.data_type}.temp'
-        self.time_tempfile  = open(self.t_tempfile_path, 'wb')
-        self.data_tempfile = open(self.d_tempfile_path, 'wb')
+        self.path_stem = ses_path_stem + f"_{self.name}"
+        self.t_tempfile_path = self.path_stem + ".time.temp"
+        self.d_tempfile_path = self.path_stem + f".data-1{self.data_type}.temp"
+        self.time_tempfile = open(self.t_tempfile_path, "wb")
+        self.data_tempfile = open(self.d_tempfile_path, "wb")
         self.next_chunk_start_time = 0
 
     def close_files(self):
-        '''Close data files. Convert temp files to numpy.'''
+        """Close data files. Convert temp files to numpy."""
         self.time_tempfile.close()
         self.data_tempfile.close()
-        with open(self.t_tempfile_path, 'rb') as f:
-            times = np.frombuffer(f.read(), dtype='float64')
-            np.save(self.path_stem + '.time.npy', times)
-        with open(self.d_tempfile_path, 'rb') as f:
+        with open(self.t_tempfile_path, "rb") as f:
+            times = np.frombuffer(f.read(), dtype="float64")
+            np.save(self.path_stem + ".time.npy", times)
+        with open(self.d_tempfile_path, "rb") as f:
             data = np.frombuffer(f.read(), dtype=self.data_type)
-            np.save(self.path_stem + '.data.npy', data)
+            np.save(self.path_stem + ".data.npy", data)
         os.remove(self.t_tempfile_path)
         os.remove(self.d_tempfile_path)
 
     def save_analog_chunk(self, timestamp, data_array):
-        '''Save a chunk of analog data to .pca data file.'''
-        if np.abs(self.next_chunk_start_time - timestamp/1000)<0.001:
+        """Save a chunk of analog data to .pca data file."""
+        if np.abs(self.next_chunk_start_time - timestamp / 1000) < 0.001:
             chunk_start_time = self.next_chunk_start_time
         else:
-            chunk_start_time = timestamp/1000
-        times = (np.arange(len(data_array), dtype='float64') 
-                 / self.sampling_rate) + chunk_start_time # Seconds
+            chunk_start_time = timestamp / 1000
+        times = (np.arange(len(data_array), dtype="float64") / self.sampling_rate) + chunk_start_time  # Seconds
         self.time_tempfile.write(times.tobytes())
         self.data_tempfile.write(data_array.tobytes())
         self.time_tempfile.flush()
         self.data_tempfile.flush()
-        self.next_chunk_start_time = chunk_start_time+len(data_array)/self.sampling_rate
+        self.next_chunk_start_time = chunk_start_time + len(data_array) / self.sampling_rate

com/pyboard.py

@@ -23,13 +23,16 @@
 import time
 import serial
 
+
 def stdout_write_bytes(b):
     sys.stdout.buffer.write(b)
     sys.stdout.buffer.flush()
 
+
 class PyboardError(BaseException):
     pass
 
+
 class Pyboard:
     def __init__(self, serial_device, baudrate=115200):
         self.serial = serial.Serial(serial_device, baudrate=baudrate, interCharTimeout=1)
@@ -59,37 +62,37 @@ def read_until(self, min_num_bytes, ending, timeout=10, data_consumer=None):
         return data
 
     def enter_raw_repl(self):
-        self.serial.write(b'\r\x03\x03') # ctrl-C twice: interrupt any running program
+        self.serial.write(b"\r\x03\x03")  # ctrl-C twice: interrupt any running program
         # flush input (without relying on serial.flushInput())
         n = self.serial.in_waiting
         while n > 0:
             self.serial.read(n)
             n = self.serial.in_waiting
-        self.serial.write(b'\r\x01') # ctrl-A: enter raw REPL
-        data = self.read_until(1, b'to exit\r\n>')
-        if not data.endswith(b'raw REPL; CTRL-B to exit\r\n>'):
+        self.serial.write(b"\r\x01")  # ctrl-A: enter raw REPL
+        data = self.read_until(1, b"to exit\r\n>")
+        if not data.endswith(b"raw REPL; CTRL-B to exit\r\n>"):
             print(data)
-            raise PyboardError('could not enter raw repl')
-        self.serial.write(b'\x04') # ctrl-D: soft reset
-        data = self.read_until(1, b'to exit\r\n>')
-        if not data.endswith(b'raw REPL; CTRL-B to exit\r\n>'):
+            raise PyboardError("could not enter raw repl")
+        self.serial.write(b"\x04")  # ctrl-D: soft reset
+        data = self.read_until(1, b"to exit\r\n>")
+        if not data.endswith(b"raw REPL; CTRL-B to exit\r\n>"):
             print(data)
-            raise PyboardError('could not enter raw repl')
+            raise PyboardError("could not enter raw repl")
 
     def exit_raw_repl(self):
-        self.serial.write(b'\r\x02') # ctrl-B: enter friendly REPL
+        self.serial.write(b"\r\x02")  # ctrl-B: enter friendly REPL
 
     def follow(self, timeout, data_consumer=None):
         # wait for normal output
-        data = self.read_until(1, b'\x04', timeout=timeout, data_consumer=data_consumer)
-        if not data.endswith(b'\x04'):
-            raise PyboardError('timeout waiting for first EOF reception')
+        data = self.read_until(1, b"\x04", timeout=timeout, data_consumer=data_consumer)
+        if not data.endswith(b"\x04"):
+            raise PyboardError("timeout waiting for first EOF reception")
         data = data[:-1]
 
         # wait for error output
-        data_err = self.read_until(2, b'\x04>', timeout=timeout)
-        if not data_err.endswith(b'\x04>'):
-            raise PyboardError('timeout waiting for second EOF reception')
+        data_err = self.read_until(2, b"\x04>", timeout=timeout)
+        if not data_err.endswith(b"\x04>"):
+            raise PyboardError("timeout waiting for second EOF reception")
         data_err = data_err[:-2]
 
         # return normal and error output
@@ -99,47 +102,48 @@ def exec_raw_no_follow(self, command):
         if isinstance(command, bytes):
             command_bytes = command
         else:
-            command_bytes = bytes(command, encoding='utf8')
+            command_bytes = bytes(command, encoding="utf8")
 
         # write command
         for i in range(0, len(command_bytes), 256):
-            self.serial.write(command_bytes[i:min(i + 256, len(command_bytes))])
+            self.serial.write(command_bytes[i : min(i + 256, len(command_bytes))])
             time.sleep(0.01)
-        self.serial.write(b'\x04')
+        self.serial.write(b"\x04")
 
         # check if we could exec command
         data = self.serial.read(2)
-        if data != b'OK':
-            raise PyboardError('could not exec command')
+        if data != b"OK":
+            raise PyboardError("could not exec command")
 
     def exec_raw(self, command, timeout=10, data_consumer=None):
-        self.exec_raw_no_follow(command);
+        self.exec_raw_no_follow(command)
         return self.follow(timeout, data_consumer)
 
     def eval(self, expression):
-        ret = self.exec('print({})'.format(expression))
+        ret = self.exec("print({})".format(expression))
         ret = ret.strip()
         return ret
 
     def exec(self, command):
         ret, ret_err = self.exec_raw(command)
         if ret_err:
-            raise PyboardError('exception', ret, ret_err)
+            raise PyboardError("exception", ret, ret_err)
         return ret
 
     def execfile(self, filename):
-        with open(filename, 'rb') as f:
+        with open(filename, "rb") as f:
             pyfile = f.read()
         return self.exec(pyfile)
 
     def get_time(self):
-        t = str(self.eval('pyb.RTC().datetime()'), encoding='utf8')[1:-1].split(', ')
+        t = str(self.eval("pyb.RTC().datetime()"), encoding="utf8")[1:-1].split(", ")
         return int(t[4]) * 3600 + int(t[5]) * 60 + int(t[6])
 
-def execfile(filename, device='/dev/ttyACM0'):
+
+def execfile(filename, device="/dev/ttyACM0"):
     pyb = Pyboard(device)
     pyb.enter_raw_repl()
     output = pyb.execfile(filename)
     stdout_write_bytes(output)
     pyb.exit_raw_repl()
-    pyb.close()
+    pyb.close()