Helper object to buffer serial and setup baud
| 65 | |
| 66 | |
| 67 | class SerialTester(object): |
| 68 | """Helper object to buffer serial and setup baud""" |
| 69 | |
| 70 | def __init__(self, port): |
| 71 | self.raw_serial = serial.Serial(port=port, bytesize=serial.EIGHTBITS, parity=serial.PARITY_NONE, stopbits=serial.STOPBITS_ONE, timeout=None, xonxoff=False, rtscts=False, write_timeout=None, dsrdtr=False, inter_byte_timeout=None, exclusive=None) |
| 72 | self.raw_serial.write_timeout = ERROR_TIMEOUT_SECONDS |
| 73 | self._queue = queue.Queue() |
| 74 | self._write_thread = threading.Thread(target=self._serial_main) |
| 75 | self._write_thread.start() |
| 76 | |
| 77 | def __enter__(self): |
| 78 | return self |
| 79 | |
| 80 | def __exit__(self, exception_type, value, traceback): |
| 81 | self._queue.put(None) |
| 82 | self._write_thread.join(ERROR_TIMEOUT_SECONDS) |
| 83 | assert not self._write_thread.is_alive(), "Thread join failed" |
| 84 | self.raw_serial.close() |
| 85 | self.raw_serial = None |
| 86 | return False |
| 87 | |
| 88 | def new_session_with_baud(self, baud, parent_test): |
| 89 | """Start a new session by restarting target and setting baud""" |
| 90 | test_info = parent_test.create_subtest("Set Baud") |
| 91 | |
| 92 | # Set baud to 115200 |
| 93 | self.raw_serial.baudrate = 115200 |
| 94 | self.raw_serial.timeout = 1.0 |
| 95 | |
| 96 | # Reset the target |
| 97 | self.raw_serial.sendBreak() |
| 98 | self.raw_serial.reset_output_buffer() |
| 99 | self.raw_serial.reset_input_buffer() |
| 100 | |
| 101 | # Wait until the target is initialized |
| 102 | expected_resp = "{init}" |
| 103 | resp = self.read(len(expected_resp)) |
| 104 | if not _same(resp.decode(), expected_resp): |
| 105 | test_info.failure("Fail on init: %s" % resp) |
| 106 | return False |
| 107 | |
| 108 | # Change baudrate to that of the first test |
| 109 | command = "{baud:%i}" % baud |
| 110 | self.write(command.encode()) |
| 111 | resp = self.read(len(command)) |
| 112 | if not _same(resp.decode(), command): |
| 113 | test_info.failure("Fail on baud command: %s" % resp) |
| 114 | return False |
| 115 | |
| 116 | # Update baud of local serial port |
| 117 | self.raw_serial.baudrate = baud |
| 118 | |
| 119 | # Read the response indicating that the baudrate |
| 120 | # on the target has changed |
| 121 | expected_resp = "{change}" |
| 122 | resp = self.read(len(expected_resp)) |
| 123 | if not _same(resp.decode(), expected_resp): |
| 124 | test_info.failure("Fail on baud change %s" % resp) |