Move UART support to PTY driver
Different servo interfaces inherit from pty_driver, and all of them
could potentially provide access to the UART channel available over
this driver.
This change refactors the code such that the UART support is provided
by the base class (pty_driver.py), and as such is available to all
devices deriving from it.
The configuration changes add the uart_cmd functionality for the CPU
UART port.
BUG=chromium-os:15610
TEST=manual
. build new servod image and start it up on a servo connected to a
Snow device running in u-boot console mode
enter the following commands:
$ dut-control cpu_uart_regexp:'["mmcinfo",]'
$ dut-control cpu_uart_cmd:help
$ dut-control cpu_uart_regexp:'["mmcinfx",]'
$ dut-control cpu_uart_cmd:help
Problem setting 'cpu_uart_cmd' to 'help' :: Timeout waiting for EC response.
observe proper target behavior: it shows the 'help' output on the screen
. verify that EC keys are still processed properly (the following
test succeeds)
$ ~/trunk/src/scripts/run_remote_tests.sh --board=link --servo --remote=192.168.1.23 \
firmware_ECKeyboard --args='board=link'
Change-Id: I16fe0d9472dbc3c653f5c3f24eca6c665e966ea7
Signed-off-by: Vadim Bendebury <vbendeb@chromium.org>
Reviewed-on: https://gerrit.chromium.org/gerrit/42454
Reviewed-by: Tom Wai-Hong Tam <waihong@chromium.org>
Reviewed-by: Vic Yang <victoryang@chromium.org>
Reviewed-by: Todd Broch <tbroch@chromium.org>
diff --git a/servo/data/servo_v2_r0.xml b/servo/data/servo_v2_r0.xml
index a5bd139..1356ee7 100644
--- a/servo/data/servo_v2_r0.xml
+++ b/servo/data/servo_v2_r0.xml
@@ -645,4 +645,25 @@
<params drv="uart" subtype="props" line_prop="bits"
interface="8" map="uart_bits"></params>
</control>
+ <control>
+ <name>cpu_uart_cmd</name>
+ <doc>Set to send command to CPU UART. Get to obtain the matched
+ results with the regular expression of cpu_uart_regexp.</doc>
+ <params subtype="uart_cmd" interface="8" drv="uart"
+ input_type="str"></params>
+ </control>
+ <control>
+ <name>cpu_uart_regexp</name>
+ <doc>List of regular expressions which matches the response of
+ cpu_uart_cmd.</doc>
+ <params subtype="uart_regexp" interface="8" drv="uart"
+ input_type="str"></params>
+ </control>
+ <control>
+ <name>cpu_uart_timeout</name>
+ <doc>Timeout value for waiting CPU UART response of issuing an
+ EC command.</doc>
+ <params subtype="uart_timeout" interface="8" drv="uart"
+ input_type="float"></params>
+ </control>
</root>
diff --git a/servo/drv/ec.py b/servo/drv/ec.py
index e98da8e..3d704f6 100644
--- a/servo/drv/ec.py
+++ b/servo/drv/ec.py
@@ -12,20 +12,17 @@
kbd_m2_a1
dev_mode (Temporary. See crosbug.com/p/9341)
"""
-import ast
import logging
import pty_driver
# Default setting values
-DEFAULT_DICT = {'kbd_en': 0,
+EXTRA_PARAMS = {'kbd_en': 0,
'kbd_m1_a0': 1,
'kbd_m1_a1': 1,
'kbd_m2_a0': 1,
'kbd_m2_a1': 1,
- 'uart_cmd': None,
- 'uart_regexp': None,
- 'uart_timeout': pty_driver.DEFAULT_UART_TIMEOUT}
+ }
# Key matrix row and column mapped from kbd_m*_a*
KEY_MATRIX = [[[(0,4), (11,4)], [(2,4), None]],
@@ -63,8 +60,8 @@
"""
super(ec, self).__init__(interface, params)
self._logger.debug("")
- # A dictionary that stores current setting values
- self._dict = DEFAULT_DICT
+ # Add locals to the values dictionary.
+ self._dict.update(EXTRA_PARAMS)
def _limit_channel(self, name):
"""
@@ -343,70 +340,3 @@
else:
# "-1" is treated as max fan RPM in EC, so we don't need to handle that
self._issue_cmd("fanset %d" % value)
-
- def _Set_uart_timeout(self, timeout):
- """Set timeout value for waiting EC UART response.
-
- Args:
- timeout: Timeout value in second.
- """
- self._dict['uart_timeout'] = timeout
-
- def _Get_uart_timeout(self):
- """Get timeout value for waiting EC UART response.
-
- Returns:
- Timeout value in second.
- """
- return self._dict['uart_timeout']
-
- def _Set_uart_regexp(self, regexp):
- """Set the list of regular expressions which matches the command response.
-
- Args:
- regexp: A string which contains a list of regular expressions.
- """
- if not isinstance(regexp, str):
- raise ecError('The argument regexp should be a string.')
- self._dict['uart_regexp'] = ast.literal_eval(regexp)
-
- def _Get_uart_regexp(self):
- """Get the list of regular expressions which matches the command response.
-
- Returns:
- A string which contains a list of regular expressions.
- """
- return str(self._dict['uart_regexp'])
-
- def _Set_uart_cmd(self, cmd):
- """Set the UART command and send it to EC UART.
-
- If ec_uart_regexp is 'None', the command is just sent and it doesn't care
- about its response.
-
- If ec_uart_regexp is not 'None', the command is send and its response,
- which matches the regular expression of ec_uart_regexp, will be kept.
- Use its getter to obtain this result. If no match after ec_uart_timeout
- seconds, a timeout error will be raised.
-
- Args:
- cmd: A string of UART command.
- """
- if self._dict['uart_regexp']:
- self._dict['uart_cmd'] = self._issue_cmd_get_results(
- cmd,
- self._dict['uart_regexp'],
- self._dict['uart_timeout'])
- else:
- self._dict['uart_cmd'] = None
- self._issue_cmd(cmd)
-
- def _Get_uart_cmd(self):
- """Get the result of the latest UART command.
-
- Returns:
- A string which contains a list of tuples, each of which contains the
- entire matched string and all the subgroups of the match. 'None' if
- the ec_uart_regexp is 'None'.
- """
- return str(self._dict['uart_cmd'])
diff --git a/servo/drv/pty_driver.py b/servo/drv/pty_driver.py
index 06c1b00..8985ba1 100644
--- a/servo/drv/pty_driver.py
+++ b/servo/drv/pty_driver.py
@@ -1,6 +1,8 @@
# Copyright (c) 2012 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
+
+import ast
import fdpexpect
import os
import pexpect
@@ -12,6 +14,10 @@
class ptyError(Exception):
"""Exception class for ec."""
+UART_PARAMS = {'uart_cmd': None,
+ 'uart_regexp': None,
+ 'uart_timeout': DEFAULT_UART_TIMEOUT
+ }
class ptyDriver(hw_driver.HwDriver):
"""."""
@@ -21,20 +27,21 @@
self._child = None
self._fd = None
self._pty_path = self._interface.get_pty()
+ self._dict = UART_PARAMS
def _open(self):
- """Open EC console and create pexpect interface."""
+ """Connect to serial device and create pexpect interface."""
self._fd = os.open(self._pty_path, os.O_RDWR | os.O_NONBLOCK)
self._child = fdpexpect.fdspawn(self._fd)
def _close(self):
- """Close EC console."""
+ """Close serial device connection."""
os.close(self._fd)
self._fd = None
self._child = None
def _flush(self):
- """Flush EC console output to prevent previous messages interfering."""
+ """Flush device output to prevent previous messages interfering."""
self._child.sendline("")
while True:
try:
@@ -45,22 +52,22 @@
def _send(self, cmd):
"""Send command to EC.
- This function always flush EC console before sending, and is used as
- a wrapper function to make sure EC console is always flushed before
+ This function always flushes serial device before sending, and is used as
+ a wrapper function to make sure the channel is always flushed before
sending commands.
Args:
- cmd: The command string to send to EC.
+ cmd: The command string to send to the device.
Raises:
- ptyError: Raised when writing to EC fails.
+ ptyError: Raised when writing to the device fails.
"""
self._flush()
if self._child.sendline(cmd) != len(cmd) + 1:
raise ptyError("Failed to send command.")
def _issue_cmd(self, cmd):
- """Send command to EC and do not wait for response.
+ """Send command to the device and do not wait for response.
Args:
cmd: The command string to send to EC.
@@ -69,7 +76,7 @@
def _issue_cmd_get_results(self, cmd,
regex_list, timeout=DEFAULT_UART_TIMEOUT):
- """Send command to EC and wait for response.
+ """Send command to the device and wait for response.
This function waits for response message matching a regular
expressions.
@@ -93,7 +100,7 @@
[('High temp: 37.2', '37', '2'), ('Low temp: 36.4', '36', '4')]
Raises:
- ptyError: If timed out waiting for EC response
+ ptyError: If timed out waiting for a response
"""
result_list = []
self._open()
@@ -112,13 +119,13 @@
except pexpect.TIMEOUT:
self._logger.debug("Before: ^%s^" % self._child.before)
self._logger.debug("After: ^%s^" % self._child.after)
- raise ptyError("Timeout waiting for EC response.")
+ raise ptyError("Timeout waiting for response.")
finally:
self._close()
return result_list
def _issue_cmd_get_multi_results(self, cmd, regex):
- """Send command to EC and wait for multiple response.
+ """Send command to the device and wait for multiple response.
This function waits for arbitary number of response message
matching a regular expression.
@@ -151,3 +158,70 @@
finally:
self._close()
return result_list
+
+ def _Set_uart_timeout(self, timeout):
+ """Set timeout value for waiting for the device response.
+
+ Args:
+ timeout: Timeout value in second.
+ """
+ self._dict['uart_timeout'] = timeout
+
+ def _Get_uart_timeout(self):
+ """Get timeout value for waiting for the device response.
+
+ Returns:
+ Timeout value in second.
+ """
+ return self._dict['uart_timeout']
+
+ def _Set_uart_regexp(self, regexp):
+ """Set the list of regular expressions which matches the command response.
+
+ Args:
+ regexp: A string which contains a list of regular expressions.
+ """
+ if not isinstance(regexp, str):
+ raise ecError('The argument regexp should be a string.')
+ self._dict['uart_regexp'] = ast.literal_eval(regexp)
+
+ def _Get_uart_regexp(self):
+ """Get the list of regular expressions which matches the command response.
+
+ Returns:
+ A string which contains a list of regular expressions.
+ """
+ return str(self._dict['uart_regexp'])
+
+ def _Set_uart_cmd(self, cmd):
+ """Set the UART command and send it to the device.
+
+ If ec_uart_regexp is 'None', the command is just sent and it doesn't care
+ about its response.
+
+ If ec_uart_regexp is not 'None', the command is send and its response,
+ which matches the regular expression of ec_uart_regexp, will be kept.
+ Use its getter to obtain this result. If no match after ec_uart_timeout
+ seconds, a timeout error will be raised.
+
+ Args:
+ cmd: A string of UART command.
+ """
+ if self._dict['uart_regexp']:
+ self._dict['uart_cmd'] = self._issue_cmd_get_results(
+ cmd,
+ self._dict['uart_regexp'],
+ self._dict['uart_timeout'])
+ else:
+ self._dict['uart_cmd'] = None
+ self._issue_cmd(cmd)
+
+ def _Get_uart_cmd(self):
+ """Get the result of the latest UART command.
+
+ Returns:
+ A string which contains a list of tuples, each of which contains the
+ entire matched string and all the subgroups of the match. 'None' if
+ the ec_uart_regexp is 'None'.
+ """
+ return str(self._dict['uart_cmd'])
diff --git a/servo/drv/uart.py b/servo/drv/uart.py
index 934f753..c79ec3d 100644
--- a/servo/drv/uart.py
+++ b/servo/drv/uart.py
@@ -6,14 +6,14 @@
import logging
-import hw_driver
+import pty_driver
class uartError(Exception):
"""Error class for uart class."""
-class uart(hw_driver.HwDriver):
+class uart(pty_driver.ptyDriver):
"""Object to access type=uart controls.
Note, instances of this object get dispatched via base class,
