conn_primitive_serial.py
Go to the documentation of this file.
#!/usr/bin/env python
"""
mbed SDK
Copyright (c) 2011-2016 ARM Limited
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
import time
from serial import Serial, SerialException
from mbed_host_tests import host_tests_plugins
from mbed_host_tests.host_tests_plugins.host_test_plugins import HostTestPluginBase
from .conn_primitive import ConnectorPrimitive, ConnectorPrimitiveException
class SerialConnectorPrimitive(ConnectorPrimitive):
def __init__(self, name, port, baudrate, config):
ConnectorPrimitive.__init__(self, name)
self.portport = port
self.baudratebaudrate = int(baudrate)
self.read_timeoutread_timeout = 0.01 # 10 milli sec
self.write_timeoutwrite_timeout = 5
self.configconfig = config
self.target_idtarget_id = self.configconfig.get('target_id', None)
self.polling_timeout polling_timeoutpolling_timeout = config.get('polling_timeout', 60)
self.forced_reset_timeoutforced_reset_timeout = config.get('forced_reset_timeout', 1)
self.skip_resetskip_reset = config.get('skip_reset', False)
self.serialserial = None
# Check if serial port for given target_id changed
# If it does we will use new port to open connections and make sure reset plugin
# later can reuse opened already serial port
#
# Note: This listener opens serial port and keeps connection so reset plugin uses
# serial port object not serial port name!
serial_port = HostTestPluginBase().check_serial_port_ready(self.portport, target_id=self.target_idtarget_id, timeout=self.polling_timeout polling_timeoutpolling_timeout)
if serial_port is None:
raise ConnectorPrimitiveException("Serial port not ready!")
if serial_port != self.portport:
# Serial port changed for given targetID
self.loggerlogger.prn_inf("serial port changed from '%s to '%s')"% (self.portport, serial_port))
self.portport = serial_port
startTime = time.time()
self.loggerlogger.prn_inf("serial(port=%s, baudrate=%d, read_timeout=%s, write_timeout=%d)"% (self.portport, self.baudratebaudrate, self.read_timeoutread_timeout, self.write_timeoutwrite_timeout))
while time.time() - startTime < self.polling_timeout polling_timeoutpolling_timeout:
try:
# TIMEOUT: While creating Serial object timeout is delibrately passed as 0. Because blocking in Serial.read
# impacts thread and mutliprocess functioning in Python. Hence, instead in self.read() s delay (sleep()) is
# inserted to let serial buffer collect data and avoid spinning on non blocking read().
self.serialserial = Serial(self.portport, baudrate=self.baudratebaudrate, timeout=0, write_timeout=self.write_timeoutwrite_timeout)
except SerialException as e:
self.serialserial = None
self.LAST_ERROR LAST_ERRORLAST_ERROR = "connection lost, serial.Serial(%s, %d, %d, %d): %s"% (self.portport,
self.baudratebaudrate,
self.read_timeoutread_timeout,
self.write_timeoutwrite_timeout,
str(e))
self.loggerlogger.prn_err(str(e))
self.loggerlogger.prn_err("Retry after 1 sec until %s seconds" % self.polling_timeout polling_timeoutpolling_timeout)
else:
if not self.skip_resetskip_reset:
self.reset_dev_via_serialreset_dev_via_serial(delay=self.forced_reset_timeoutforced_reset_timeout)
break
time.sleep(1)
def reset_dev_via_serial(self, delay=1):
"""! Reset device using selected method, calls one of the reset plugins """
reset_type = self.configconfig.get('reset_type', 'default')
if not reset_type:
reset_type = 'default'
disk = self.configconfig.get('disk', None)
self.loggerlogger.prn_inf("reset device using '%s' plugin..."% reset_type)
result = host_tests_plugins.call_plugin('ResetMethod',
reset_type,
serial=self.serialserial,
disk=disk,
target_id=self.target_idtarget_id,
polling_timeout=self.configconfig.get('polling_timeout'))
# Post-reset sleep
if delay:
self.loggerlogger.prn_inf("waiting %.2f sec after reset"% delay)
time.sleep(delay)
self.loggerlogger.prn_inf("wait for it...")
return result
def read(self, count):
"""! Read data from serial port RX buffer """
# TIMEOUT: Since read is called in a loop, wait for self.timeout period before calling serial.read(). See
# comment on serial.Serial() call above about timeout.
time.sleep(self.read_timeoutread_timeout)
c = str()
try:
if self.serialserial:
c = self.serialserial.read(count)
except SerialException as e:
self.serialserial = None
self.LAST_ERROR LAST_ERRORLAST_ERROR = "connection lost, serial.read(%d): %s"% (count, str(e))
self.loggerlogger.prn_err(str(e))
return c
def write(self, payload, log=False):
"""! Write data to serial port TX buffer """
try:
if self.serialserial:
self.serialserial.write(payload.encode('utf-8'))
if log:
self.loggerlogger.prn_txd(payload)
return True
except SerialException as e:
self.serialserial = None
self.LAST_ERROR LAST_ERRORLAST_ERROR = "connection lost, serial.write(%d bytes): %s"% (len(payload), str(e))
self.loggerlogger.prn_err(str(e))
return False
def flush(self):
if self.serialserial:
self.serialserial.flush()
def connected(self):
return bool(self.serialserial)
def finish(self):
if self.serialserial:
self.serialserial.close()
def reset(self):
self.reset_dev_via_serialreset_dev_via_serial(self.forced_reset_timeoutforced_reset_timeout)
def __del__(self):
self.finish finishfinish()
mbed_host_tests.host_tests_conn_proxy.conn_primitive.ConnectorPrimitiveException
Definition: conn_primitive.py:22
mbed_host_tests.host_tests_conn_proxy.conn_primitive.ConnectorPrimitive
Definition: conn_primitive.py:29
mbed_host_tests.host_tests_conn_proxy.conn_primitive.ConnectorPrimitive.LAST_ERROR
LAST_ERROR
Definition: conn_primitive.py:32
mbed_host_tests.host_tests_conn_proxy.conn_primitive.ConnectorPrimitive.finish
def finish(self)
Handle DUT dtor like (close resource) operations here.
Definition: conn_primitive.py:87
mbed_host_tests.host_tests_conn_proxy.conn_primitive.ConnectorPrimitive.polling_timeout
polling_timeout
Definition: conn_primitive.py:34
mbed_host_tests.host_tests_conn_proxy.conn_primitive.ConnectorPrimitive.logger
logger
Definition: conn_primitive.py:33
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive
Definition: conn_primitive_serial.py:27
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.write_timeout
write_timeout
Definition: conn_primitive_serial.py:33
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.reset
def reset(self)
Reset the dut.
Definition: conn_primitive_serial.py:140
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.reset_dev_via_serial
def reset_dev_via_serial(self, delay=1)
Reset device using selected method, calls one of the reset plugins.
Definition: conn_primitive_serial.py:79
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.finish
def finish(self)
Handle DUT dtor like (close resource) operations here.
Definition: conn_primitive_serial.py:136
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.port
port
Definition: conn_primitive_serial.py:30
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.baudrate
baudrate
Definition: conn_primitive_serial.py:31
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.connected
def connected(self)
Check if there is a connection to DUT.
Definition: conn_primitive_serial.py:133
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.polling_timeout
polling_timeout
Definition: conn_primitive_serial.py:36
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.flush
def flush(self)
Flush read/write channels of DUT.
Definition: conn_primitive_serial.py:129
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.__init__
def __init__(self, name, port, baudrate, config)
Definition: conn_primitive_serial.py:28
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.__del__
def __del__(self)
Definition: conn_primitive_serial.py:143
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.forced_reset_timeout
forced_reset_timeout
Definition: conn_primitive_serial.py:37
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.config
config
Definition: conn_primitive_serial.py:34
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.read_timeout
read_timeout
Definition: conn_primitive_serial.py:32
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.skip_reset
skip_reset
Definition: conn_primitive_serial.py:38
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.LAST_ERROR
LAST_ERROR
Definition: conn_primitive_serial.py:66
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.read
def read(self, count)
Read data from serial port RX buffer.
Definition: conn_primitive_serial.py:100
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.write
def write(self, payload, log=False)
Write data to serial port TX buffer.
Definition: conn_primitive_serial.py:115
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.serial
serial
Definition: conn_primitive_serial.py:39
mbed_host_tests.host_tests_conn_proxy.conn_primitive_serial.SerialConnectorPrimitive.target_id
target_id
Definition: conn_primitive_serial.py:35
mbed_host_tests.host_tests_plugins.host_test_plugins.HostTestPluginBase
Base class for all plugins used with host tests.
Definition: host_test_plugins.py:32
mbed_host_tests.host_tests_plugins.host_test_plugins
Definition: host_test_plugins.py:1