.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 from datetime import datetime,timezone
  26 import queue
  27 import serial
  28 import threading
  29
  30 class SerialBuffer:
  31     def __init__(self, dev, filename, prefix):
  32         self.f = open(filename, "wb+")
  33         self.dev = dev
  34         self.serial = serial.Serial(dev, 115200, timeout=10)
  35         self.byte_queue = queue.Queue()
  36         self.line_queue = queue.Queue()
  37         self.prefix = prefix
  38         self.sentinel = object()
  39
  40         self.read_thread = threading.Thread(target=self.serial_read_thread_loop, daemon=True)
  41         self.read_thread.start()
  42         self.lines_thread = threading.Thread(target=self.serial_lines_thread_loop, daemon=True)
  43         self.lines_thread.start()
  44
  45     # Thread that just reads the bytes from the serial device to try to keep from
  46     # buffer overflowing it.
  47     def serial_read_thread_loop(self):
  48         greet = "Serial thread reading from %s\n" % self.dev
  49         self.byte_queue.put(greet.encode())
  50
  51         while True:
  52             try:
  53                 self.byte_queue.put(self.serial.read())
  54             except Exception as err:
  55                 print(self.prefix + str(err))
  56                 self.byte_queue.put(self.sentinel)
  57                 break
  58
  59     # Thread that processes the stream of bytes to 1) log to stdout, 2) log to
  60     # file, 3) add to the queue of lines to be read by program logic
  61
  62     def serial_lines_thread_loop(self):
  63         line = bytearray()
  64         while True:
  65             bytes = self.byte_queue.get(block=True)
  66
  67             if bytes == self.sentinel:
  68                 self.read_thread.join()
  69                 self.line_queue.put(self.sentinel)
  70                 break;
  71
  72             # Write our data to the output file
  73             self.f.write(bytes)
  74             self.f.flush()
  75
  76             for b in bytes:
  77                 line.append(b)
  78                 if b == b'\n'[0]:
  79                     line = line.decode(errors="replace")
  80
  81                     time = datetime.now().strftime('%y-%m-%d %H:%M:%S')
  82                     print("{time} {prefix}{line}".format(time=time, prefix=self.prefix, line=line), flush=True, end='')
  83
  84                     self.line_queue.put(line)
  85                     line = bytearray()
  86
  87     def get_line(self):
  88         line = self.line_queue.get()
  89         if line == self.sentinel:
  90             self.lines_thread.join()
  91         return line
  92
  93     def lines(self):
  94         return iter(self.get_line, self.sentinel)
  95
  96 def main():
  97     parser = argparse.ArgumentParser()
  98
  99     parser.add_argument('--dev', type=str, help='Serial device', required=True)
 100     parser.add_argument('--file', type=str, help='Filename to output our serial data to')
 101     parser.add_argument('--prefix', type=str, help='Prefix for logging serial to stdout', nargs='?')
 102
 103     args = parser.parse_args()
 104
 105     ser = SerialBuffer(args.dev, args.file, args.prefix or "")
 106     for line in ser.lines():
 107         # We're just using this as a logger, so eat the produced lines and drop
 108         # them
 109         pass
 110
 111 if __name__ == '__main__':
 112     main()