.gitlab-ci/bare-metal/serial_buffer.py

   1 #!/usr/bin/env python3
   2 #
   3 # Copyright © 2020 Google LLC
   4 #
   5 # Permission is hereby granted, free of charge, to any person obtaining a
   6 # copy of this software and associated documentation files (the "Software"),
   7 # to deal in the Software without restriction, including without limitation
   8 # the rights to use, copy, modify, merge, publish, distribute, sublicense,
   9 # and/or sell copies of the Software, and to permit persons to whom the
  10 # Software is furnished to do so, subject to the following conditions:
  11 #
  12 # The above copyright notice and this permission notice (including the next
  13 # paragraph) shall be included in all copies or substantial portions of the
  14 # Software.
  15 #
  16 # THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
  17 # IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
  18 # FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT.  IN NO EVENT SHALL
  19 # THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
  20 # LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
  21 # FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS
  22 # IN THE SOFTWARE.
  23
  24 import argparse
  25 import queue
  26 import serial
  27 import threading
  28
  29 class SerialBuffer:
  30     def __init__(self, dev, filename, prefix):
  31         self.f = open(filename, "wb+")
  32         self.dev = dev
  33         self.serial = serial.Serial(dev, 115200, timeout=10)
  34         self.byte_queue = queue.Queue()
  35         self.line_queue = queue.Queue()
  36         self.prefix = prefix
  37         self.sentinel = object()
  38
  39         self.read_thread = threading.Thread(target=self.serial_read_thread_loop, daemon=True)
  40         self.read_thread.start()
  41         self.lines_thread = threading.Thread(target=self.serial_lines_thread_loop, daemon=True)
  42         self.lines_thread.start()
  43
  44     # Thread that just reads the bytes from the serial device to try to keep from
  45     # buffer overflowing it.
  46     def serial_read_thread_loop(self):
  47         greet = "Serial thread reading from %s\n" % self.dev
  48         self.byte_queue.put(greet.encode())
  49
  50         while True:
  51             try:
  52                 self.byte_queue.put(self.serial.read())
  53             except Exception as err:
  54                 print(self.prefix + str(err))
  55                 self.byte_queue.put(self.sentinel)
  56                 break
  57
  58     # Thread that processes the stream of bytes to 1) log to stdout, 2) log to
  59     # file, 3) add to the queue of lines to be read by program logic
  60
  61     def serial_lines_thread_loop(self):
  62         line = bytearray()
  63         while True:
  64             bytes = self.byte_queue.get(block=True)
  65
  66             if bytes == self.sentinel:
  67                 self.read_thread.join()
  68                 self.line_queue.put(self.sentinel)
  69                 break;
  70
  71             # Write our data to the output file
  72             self.f.write(bytes)
  73             self.f.flush()
  74
  75             for b in bytes:
  76                 line.append(b)
  77                 if b == b'\n'[0]:
  78                     line = line.decode(errors="replace")
  79                     print(self.prefix + line, flush=True, end='')
  80                     self.line_queue.put(line)
  81                     line = bytearray()
  82
  83     def get_line(self):
  84         line = self.line_queue.get()
  85         if line == self.sentinel:
  86             self.lines_thread.join()
  87         return line
  88
  89     def lines(self):
  90         return iter(self.get_line, self.sentinel)
  91
  92 def main():
  93     parser = argparse.ArgumentParser()
  94
  95     parser.add_argument('--dev', type=str, help='Serial device', required=True)
  96     parser.add_argument('--file', type=str, help='Filename to output our serial data to')
  97     parser.add_argument('--prefix', type=str, help='Prefix for logging serial to stdout', nargs='?')
  98
  99     args = parser.parse_args()
 100
 101     ser = SerialBuffer(args.dev, args.file, args.prefix or "")
 102     for line in ser.lines():
 103         # We're just using this as a logger, so eat the produced lines and drop
 104         # them
 105         pass
 106
 107 if __name__ == '__main__':
 108     main()