testtools/UART_interface/esp_arduino_uart_interface.py [53:81]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
    def serial_write(self, ser, message, file=None):

        # Check that the device is no longer sending bytes
        if ser.in_waiting:
            self.serial_read(ser, message, file)

        # Check that the serial connection is open
        if ser.writable():
            wait = serial_settings.mxchip_buf_pause  # wait for at least 50ms between 128 byte writes.
            buf = bytearray((message.strip() + '\r\n').encode('ascii'))
            buf_len = len(buf)  # needed for loop as buf is a destructed list
            bytes_written = 0
            timeout = time.time()

            while bytes_written < buf_len:
                temp_written = ser.write(buf[:128])
                buf = buf[temp_written:]
                bytes_written += temp_written
                time.sleep(wait)
                if (time.time() - timeout > serial_settings.serial_comm_timeout):
                    break

            return bytes_written
        else:
            try:
                time.sleep(2)
                ser.open()
                self.serial_write(ser, message, file)
            except:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -



testtools/UART_interface/mxchip_uart_interface.py [56:85]:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
    def serial_write(self, ser, message, file=None):

        # Check that the device is no longer sending bytes
        if ser.in_waiting:
            self.serial_read(ser, message, file)

        # Check that the serial connection is open
        if ser.writable():
            wait = serial_settings.mxchip_buf_pause  # wait for at least 50ms between 128 byte writes.
            buf = bytearray((message.strip() + '\r\n').encode('ascii'))
            buf_len = len(buf)  # needed for loop as buf is a destructed list
            bytes_written = 0
            timeout = time.time()

            while bytes_written < buf_len:
                temp_written = ser.write(buf[:128])
                buf = buf[temp_written:]
                bytes_written += temp_written
                # print("bytes written: %d" %bytes_written)
                time.sleep(wait)
                if (time.time() - timeout > serial_settings.serial_comm_timeout):
                    break
            # print("final written: %d" %bytes_written)
            return bytes_written
        else:
            try:
                time.sleep(2)
                ser.open()
                self.serial_write(ser, message, file)
            except:
- - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -