1#!/usr/bin/env python3
2#
3#   Copyright 2018 - The Android Open Source Project
4#
5#   Licensed under the Apache License, Version 2.0 (the "License");
6#   you may not use this file except in compliance with the License.
7#   You may obtain a copy of the License at
8#
9#       http://www.apache.org/licenses/LICENSE-2.0
10#
11#   Unless required by applicable law or agreed to in writing, software
12#   distributed under the License is distributed on an "AS IS" BASIS,
13#   WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14#   See the License for the specific language governing permissions and
15#   limitations under the License.
16
17"""
18Class definition of B29 device for controlling the device.
19
20B29 is an engineering device with serial capabilities. It is almost like
21b20 except it has additional features that allow sending commands
22to b10 via one-wire and to pull logs from b10 via one-wire.
23
24Please see https://docs.google.com/document/d/17yJeJRNWxv5E9
25fBvw0sXkgwCBkshU_l4SxWkKgAxVmk/edit for details about available operations.
26"""
27
28import os
29import re
30import time
31
32from acts import tracelogger
33from acts import utils
34from logging import Logger
35
36logging = tracelogger.TakoTraceLogger(Logger(__file__))
37DEVICE_REGEX = (
38    r'_(?P<device_serial>[A-Z0-9]+)-(?P<interface>\w+)\s->\s'
39    r'(\.\./){2}(?P<port>\w+)'
40)
41# TODO: automate getting the latest version from x20
42DEBUG_BRIDGE = ('/google/data/ro/teams/wearables/apollo/ota/jenkins-presubmit/'
43                'ovyalov/master/apollo-sw/CL14060_v2-build13686/v13686/'
44                'automation/apollo_debug_bridge/linux2/apollo_debug_bridge')
45B29_CHIP = 'Cypress_Semiconductor_USBUART'
46
47
48# TODO:
49# as the need arises, additional functionalities of debug_bridge should be
50# integrated
51# TODO:
52# https://docs.google.com/document/d/17yJeJRNWxv5E9fBvw0sXkgwCBkshU_
53# l4SxWkKgAxVmk/edit
54
55class B29Error(Exception):
56    """Module Level Error."""
57
58
59def get_b29_devices():
60    """ Get all available B29 devices.
61
62    Returns:
63      (list) A list of available devices (ex: ['/dev/ttyACM4',...]) or empty
64      list if none found
65    """
66    devices = []
67    result = os.popen('ls -l /dev/serial/by-id/*%s*' % B29_CHIP).read()
68    for line in result.splitlines():
69        match = re.search(DEVICE_REGEX, line)
70        device_serial = match.group('device_serial')
71        log_port = None
72        commander_port = '/dev/' + match.group('port')
73        device = {
74            'commander_port': commander_port,
75            'log_port': log_port,
76            'serial_number': device_serial
77        }
78        devices.append(device)
79    return devices
80
81
82class B29Device(object):
83    """Class to control B29 device."""
84
85    def __init__(self, b29_serial):
86        """ Class to control B29 device
87        Args: String type of serial number (ex: 'D96045152F121B00'
88        """
89        self.serial = b29_serial
90        b29_port = [d['commander_port'] for d in get_b29_devices() if
91                    d['serial_number'] == b29_serial]
92        if not b29_port:
93            logging.error("unable to find b29 with serial number %s" %
94                          b29_serial)
95            raise B29Error(
96                "Recovery failed because b29_serial specified in device "
97                "manifest file is not found or invalid")
98        self.port = b29_port[0]
99        self.ping_match = {'psoc': r'Pings: tx=[\d]* rx=[1-9][0-9]',
100                           'csr': r'count=100, sent=[\d]*, received=[1-9][0-9]',
101                           'charger': r'Pings: tx=[\d]* rx=[1-9][0-9]'}
102        self.fw_version = self._get_version('fw')
103        self.app_version = self._get_version('app')
104
105    def _get_version(self, type='fw'):
106        """ Method to get version of B29
107        Returns:
108            String version if found (ex: '0006'), None otherwise
109        """
110        command = '--serial={}'.format(self.port)
111        debug_bridge_process = self._send_command(command=command)
112        if type == 'fw':
113            version_match = re.compile(r'CHARGER app version: version=([\d]*)')
114        elif type == 'app':
115            version_match = re.compile(r'APP VERSION: ([\d]*)')
116        version_str = self._parse_output_of_running_process(
117            debug_bridge_process, version_match)
118        debug_bridge_process.kill()
119        if version_str:
120            match = version_match.search(version_str)
121            version = match.groups()[0]
122            return version
123        return None
124
125    def _parse_output_of_running_process(self, subprocess, match, timeout=30):
126        """ Parses the logs from subprocess objects and checks to see if a
127        match is found within the allotted time
128        Args:
129            subprocess: object returned by _send_command (which is the same as
130            bject returned by subprocess.Popen()) match: regex match object
131            (what is returned by re.compile(r'<regex>') timeout: int - time to
132            keep retrying before bailing
133
134        """
135        start_time = time.time()
136        success_match = re.compile(match)
137        while start_time + timeout > time.time():
138            out = subprocess.stderr.readline()
139            if success_match.search(out):
140                return out
141            time.sleep(.5)
142        return False
143
144    def _send_command(self, command):
145        """ Send command to b29 using apollo debug bridge
146        Args:
147          command: The command for apollo debug to execute
148        Returns:
149          subprocess object
150        """
151        return utils.start_standing_subprocess(
152            '{} {} {}'.format(DEBUG_BRIDGE, '--rpc_port=-1', command),
153            shell=True)
154
155    def restore_golden_image(self):
156        """ Start a subprocess that calls the debug-bridge executable with
157        options that restores golden image of b10 attached to the b29. The
158        recovery restores the 'golden image' which is available in b10 partition
159         8. The process runs for 120 seconds which is adequate time for the
160         recovery to have completed.
161        """
162        # TODO:
163        # because we are accessing x20, we need to capture error resulting from
164        #  expired prodaccess and report it explicitly
165        # TODO:
166        # possibly file not found error?
167
168        # start the process, wait for two minutes and kill it
169        logging.info('Restoring golden image...')
170        command = '--serial=%s --debug_spi=dfu --sqif_partition=8' % self.port
171        debug_bridge_process = self._send_command(command=command)
172        success_match = re.compile('DFU on partition #8 successfully initiated')
173        if self._parse_output_of_running_process(debug_bridge_process,
174                                                 success_match):
175            logging.info('Golden image restored successfully')
176            debug_bridge_process.kill()
177            return True
178        logging.warning('Failed to restore golden image')
179        debug_bridge_process.kill()
180        return False
181
182    def ping_component(self, component, timeout=30):
183        """ Send ping to the specified component via B290
184        Args:
185            component = 'csr' or 'psoc' or 'charger'
186        Returns:
187            True if successful and False otherwise
188        """
189        if component not in ('csr', 'psoc', 'charger'):
190            raise B29Error('specified parameter for component is not valid')
191        logging.info('Pinging %s via B29...' % component)
192        command = '--serial={} --ping={}'.format(self.port, component)
193        debug_bridge_process = self._send_command(command=command)
194        if self._parse_output_of_running_process(debug_bridge_process,
195                                                 self.ping_match[component],
196                                                 timeout):
197            logging.info('Ping passes')
198            debug_bridge_process.kill()
199            return True
200        else:
201            logging.warning('Ping failed')
202            debug_bridge_process.kill()
203            return False
204
205    def reset_charger(self):
206        """ Send reset command to B29
207        Raises: TimeoutError (lib.utils.TimeoutError) if the device does not
208        come back within 120 seconds
209        """
210        # --charger_reset
211        if int(self.fw_version) >= 6:
212            logging.info('Resetting B29')
213            command = '--serial={} --charger_reset'.format(self.port)
214            reset_charger_process = self._send_command(command=command)
215            time.sleep(2)
216            reset_charger_process.kill()
217            logging.info('Waiting for B29 to become available..')
218            utils.wait_until(lambda: self.ping_component('charger'), 120)
219        else:
220            logging.warning('B20 firmware version %s does not support '
221                            'charger_reset argument' % self.fw_version)
222