summaryrefslogtreecommitdiff
path: root/script/cat.py
diff options
context:
space:
mode:
Diffstat (limited to 'script/cat.py')
-rwxr-xr-xscript/cat.py108
1 files changed, 108 insertions, 0 deletions
diff --git a/script/cat.py b/script/cat.py
new file mode 100755
index 0000000..330669e
--- /dev/null
+++ b/script/cat.py
@@ -0,0 +1,108 @@
+#!/usr/bin/env python3
+
+import re
+import serial
+import serial.threaded
+import sys
+import time
+
+cpu_freq = None
+timer_overflow = None
+
+
+class SerialReader(serial.threaded.Protocol):
+ """
+ Character- to line-wise data buffer for serial interfaces.
+
+ Reads in new data whenever it becomes available and exposes a line-based
+ interface to applications.
+ """
+
+ def __init__(self, callback):
+ """Create a new SerialReader object."""
+ self.callback = callback
+ self.recv_buf = ""
+
+ def __call__(self):
+ return self
+
+ def data_received(self, data):
+ """Append newly received serial data to the line buffer."""
+ try:
+ str_data = data.decode("UTF-8")
+ self.recv_buf += str_data
+
+ # We may get anything between \r\n, \n\r and simple \n newlines.
+ # We assume that \n is always present and use str.strip to remove leading/trailing \r symbols
+ # Note: Do not call str.strip on lines[-1]! Otherwise, lines may be mangled
+ lines = self.recv_buf.split("\n")
+ if len(lines) > 1:
+ self.recv_buf = lines[-1]
+ for line in lines[:-1]:
+ self.callback(str.strip(line))
+
+ except UnicodeDecodeError:
+ pass
+ # sys.stderr.write('UART output contains garbage: {data}\n'.format(data = data))
+
+
+class SerialMonitor:
+ """SerialMonitor captures serial output for a specific amount of time."""
+
+ def __init__(self, port: str, baud: int, callback):
+ """
+ Create a new SerialMonitor connected to port at the specified baud rate.
+
+ Communication uses no parity, no flow control, and one stop bit.
+ Data collection starts immediately.
+ """
+ self.ser = serial.serial_for_url(port, do_not_open=True)
+ self.ser.baudrate = baud
+ self.ser.parity = "N"
+ self.ser.rtscts = False
+ self.ser.xonxoff = False
+
+ try:
+ self.ser.open()
+ except serial.SerialException as e:
+ sys.stderr.write(
+ "Could not open serial port {}: {}\n".format(self.ser.name, e)
+ )
+ sys.exit(1)
+
+ self.reader = SerialReader(callback=callback)
+ self.worker = serial.threaded.ReaderThread(self.ser, self.reader)
+ self.worker.start()
+
+ def close(self):
+ """Close serial connection."""
+ self.worker.stop()
+ self.ser.close()
+
+
+def handle_line(line):
+ if cpu_freq is not None:
+ print_line = line
+ for match in re.finditer("(\d+)/(\d+) cycles", line):
+ cycles = int(match.group(1))
+ overflows = int(match.group(2))
+ ms = (cycles + timer_overflow * overflows) * 1000 / cpu_freq
+ match_line = f"{cycles}/{overflows} cycles"
+ new_line = f"{ms} ms ({cycles}/{overflows} cycles)"
+ print_line = re.sub(match_line, new_line, print_line)
+ print(print_line)
+ else:
+ print(line)
+
+
+if __name__ == "__main__":
+ tty = sys.argv[1]
+ baud = int(sys.argv[2])
+
+ if len(sys.argv) > 4:
+ cpu_freq = int(sys.argv[3])
+ timer_overflow = int(sys.argv[4])
+
+ monitor = SerialMonitor(tty, baud, handle_line)
+ time.sleep(5)
+ monitor.close()