| # Copyright (c) 2013 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 collections |
| import inspect |
| import logging |
| |
| import at_channel |
| import task_loop |
| import wardmodem_exceptions as wme |
| |
| MODEM_RESPONSE_TIMEOUT_MILLISECONDS = 30000 |
| ARG_PLACEHOLDER = '*' |
| |
| class ATTransceiverMode(object): |
| """ |
| Enum to specify what mode the ATTransceiver is operating in. |
| |
| There are three modes. These modes determine how the commands to/from |
| the modemmanager are routed. |
| WARDMODEM: modemmanager interacts with wardmodem alone. |
| SPLIT_VERIFY: modemmanager commands are sent to both the wardmodem |
| and the physical modem on the device. Responses from |
| wardmodem are verified against responses from the physical |
| modem. In case of a mismatch, wardmodem's response is |
| chosen, and a warning is issued. |
| PASS_THROUGH: modemmanager commands are routed to/from the physical |
| modem. Frankly, wardmodem isn't running in this mode. |
| |
| """ |
| WARDMODEM = 0 |
| SPLIT_VERIFY = 1 |
| PASS_THROUGH = 2 |
| |
| MODE_NAME = { |
| WARDMODEM: 'WARDMODEM', |
| SPLIT_VERIFY: 'SPLIT_VERIFY', |
| PASS_THROUGH: 'PASS_THROUGH' |
| } |
| |
| |
| @classmethod |
| def to_string(cls, value): |
| """ |
| A class method to obtain string representation of the enum values. |
| |
| @param value: the enum value to stringify. |
| |
| """ |
| return "%s.%s" % (cls.__name__, cls.MODE_NAME[value]) |
| |
| |
| class ATTransceiver(object): |
| """ |
| A world facing multiplexer class that orchestrates the communication between |
| modem manager, the physical modem, and wardmodem back-end. |
| |
| """ |
| |
| def __init__(self, mm_at_port, modem_conf, |
| modem_at_port=None): |
| """ |
| @param mm_at_port: File descriptor for AT port used by modem manager. |
| Can not be None. |
| |
| @param modem_conf: A ModemConfiguration object containing the |
| configuration data for the current modem. |
| |
| @param modem_at_port: File descriptor for AT port used by the modem. May |
| be None, but that forces ATTransceiverMode.WARDMODEM. Default: |
| None. |
| |
| """ |
| super(ATTransceiver, self).__init__() |
| assert mm_at_port is not None |
| |
| self._logger = logging.getLogger(__name__) |
| self._task_loop = task_loop.get_instance() |
| self._mode = ATTransceiverMode.WARDMODEM |
| # The time we wait for any particular response from physical modem. |
| self._modem_response_timeout_milliseconds = ( |
| MODEM_RESPONSE_TIMEOUT_MILLISECONDS) |
| # We keep a queue of responses from the wardmodem and physical modem, |
| # so that we can verify they match. |
| self._cached_modem_responses = collections.deque() |
| self._cached_wardmodem_responses = collections.deque() |
| |
| # When a wardmodem response has been received but the corresponding |
| # physical modem response hasn't arrived, we post a task to wait for the |
| # response. |
| self._modem_response_wait_task = None |
| |
| # We use a map from a set of well known state machine names to actual |
| # objects to dispatch state machine calls. This allows tests to provide |
| # alternative implementations of any state machine to wardmodem. |
| self._state_machines = {} |
| |
| # If registered with a non-None machine, the fallback machine is used to |
| # service all AT commands that are not matched with any other machine. |
| self._fallback_state_machine = None |
| self._fallback_machine_function = None |
| |
| # Maps an incoming AT command from modemmanager to an internal wardmodem |
| # action. |
| self._at_to_wm_action_map = {} |
| # Maps an internal response from wardmodem to an AT command to be sent |
| # to modemmanager. |
| self._wm_response_to_at_map = {} |
| |
| # Load mapping between AT commands and wardmodem actions. |
| self._update_at_to_wm_action_map(modem_conf.base_at_to_wm_action_map) |
| self._update_at_to_wm_action_map(modem_conf.plugin_at_to_wm_action_map) |
| self._update_wm_response_to_at_map( |
| modem_conf.base_wm_response_to_at_map) |
| self._update_wm_response_to_at_map( |
| modem_conf.plugin_wm_response_to_at_map) |
| self._logger.debug('Finished loading AT --> wardmodem configuration.') |
| self._logger.debug(self._at_to_wm_action_map) |
| self._logger.debug('Finished loading wardmodem --> AT configuration.') |
| self._logger.debug(self._wm_response_to_at_map) |
| |
| # Initialize channels -- let the session begin. |
| if modem_at_port is not None: |
| self._modem_channel = at_channel.ATChannel( |
| self._process_modem_at_command, |
| modem_at_port, |
| 'modem_primary_channel') |
| self._modem_channel.at_prefix = modem_conf.mm_to_modem_at_prefix |
| self._modem_channel.at_suffix = modem_conf.mm_to_modem_at_suffix |
| else: |
| self._modem_channel = None |
| |
| self._mm_channel = at_channel.ATChannel(self._process_mm_at_command, |
| mm_at_port, |
| 'mm_primary_channel') |
| self._mm_channel.at_prefix = modem_conf.modem_to_mm_at_prefix |
| self._mm_channel.at_suffix = modem_conf.modem_to_mm_at_suffix |
| |
| |
| # Verification failure reasons |
| VERIFICATION_FAILED_MISMATCH = 1 |
| VERIFICATION_FAILED_TIME_OUT = 2 |
| |
| |
| @property |
| def mode(self): |
| """ |
| ATTranscieverMode value. Determines how commands are routed. |
| |
| @see ATTransceiverMode |
| |
| """ |
| return self._mode |
| |
| |
| @mode.setter |
| def mode(self, value): |
| """ |
| Set mode. |
| |
| @param value: The value to set. Type: ATTransceiverMode. |
| |
| """ |
| if value != ATTransceiverMode.WARDMODEM and self._modem_channel is None: |
| self._logger.warning( |
| 'Can not switch to %s mode. No modem port provided.', |
| ATTransceiverMode.to_string(value)) |
| return |
| self._logger.info('Set mode to %s', |
| ATTransceiverMode.to_string(value)) |
| self._mode = value |
| |
| |
| def get_state_machine(self, well_known_name): |
| """ |
| Get the registered state machine for the given well known name. |
| |
| @param well_known_name: The name of the desired machine. |
| |
| @return: The machine. None if not found. |
| |
| """ |
| return self._state_machines.get(well_known_name, None) |
| |
| |
| def register_state_machine(self, state_machine): |
| """ |
| Register a new state machine. |
| |
| We maintain a map from the well known name of the state machine to the |
| object. Any older object mapped to the same name will be replaced. |
| |
| @param state_machine: [StateMachine object] The state machine |
| object to be used to dispatch calls. |
| |
| """ |
| state_machine_name = state_machine.get_well_known_name() |
| self._state_machines[state_machine_name] = state_machine |
| |
| |
| def register_fallback_state_machine(self, state_machine_name, function): |
| """ |
| Register the fallback state machine to forward AT commands to. |
| |
| If this machine is registered, all AT commands for which no matching |
| rule is found will result in the call |state_machine|.|function|(at). |
| where |at| is the actual AT command that could not be matched. |
| |
| @param state_machine_name: Well known name of the machine to fallback on |
| if no machine matches an incoming AT command. |
| |
| @param function: The function in |state_machine| to call. |
| |
| """ |
| if state_machine_name not in self._state_machines: |
| self._setup_error('Machine %s, set as fallback, has not been ' |
| 'registered. ' % state_machine_name) |
| self._fallback_state_machine = state_machine_name |
| self._fallback_machine_function = function |
| |
| |
| def process_wardmodem_response(self, response, *args): |
| """ |
| Convert responses from the wardmodem into AT commands and send them to |
| modemmanager. |
| |
| @param response: wardmodem response to be translated to AT response to |
| the modem manager. |
| |
| @param *args: arguments to the wardmodem response. |
| |
| @raises: ATTransceiverError if the response can not be translated into |
| an AT command. |
| |
| """ |
| self._logger.debug('Processing wardmodem response %s%s', |
| response, str(args) if args else '') |
| if response not in self._wm_response_to_at_map: |
| self._runtime_error('Unknown wardmodem response |%s|' % response) |
| at_response = self._construct_at_response( |
| self._wm_response_to_at_map[response], *args) |
| self._process_wardmodem_at_command(at_response) |
| |
| # ########################################################################## |
| # Callbacks -- These are the functions that process events from the |
| # ATChannel or the TaskLoop. These functions are either |
| # (1) set as callbacks in the ATChannel, or |
| # (2) called internally to process the AT command to/from the TaskLoop. |
| |
| def _process_modem_at_command(self, command): |
| """ |
| Callback called by the physical modem channel when an AT response is |
| received. |
| |
| @param command: AT command sent by the physical modem. |
| |
| """ |
| assert self.mode != ATTransceiverMode.WARDMODEM |
| self._logger.debug('Command {modem ==> []}: |%s|', command) |
| if self.mode == ATTransceiverMode.PASS_THROUGH: |
| self._logger.debug('Command {[] ==> mm}: |%s|' , command) |
| self._mm_channel.send(command) |
| else: |
| self._cached_modem_responses.append(command) |
| self._verify_and_send_mm_commands() |
| |
| |
| def _process_mm_at_command(self, command): |
| """ |
| Callback called by the modem manager channel when an AT command is |
| received. |
| |
| @param command: AT command sent by modem manager. |
| |
| """ |
| self._logger.debug('Command {mm ==> []}: |%s|', command) |
| if(self.mode == ATTransceiverMode.PASS_THROUGH or |
| self.mode == ATTransceiverMode.SPLIT_VERIFY): |
| self._logger.debug('Command {[] ==> modem}: |%s|', command) |
| self._modem_channel.send(command) |
| if(self.mode == ATTransceiverMode.WARDMODEM or |
| self.mode == ATTransceiverMode.SPLIT_VERIFY): |
| self._logger.debug('Command {[] ==> wardmodem}: |%s|', command) |
| self._post_wardmodem_request(command) |
| |
| |
| def _process_wardmodem_at_command(self, command): |
| """ |
| Function called to process an AT command response of wardmodem. |
| |
| This function is called after the response from the task loop has been |
| converted to an AT command. |
| |
| @param command: The AT command response of wardmodem. |
| |
| """ |
| assert self.mode != ATTransceiverMode.PASS_THROUGH |
| self._logger.debug('Command {wardmodem ==> []: |%s|', command) |
| if self.mode == ATTransceiverMode.WARDMODEM: |
| self._logger.debug('Command {[] ==> mm}: |%s|', command) |
| self._mm_channel.send(command) |
| else: |
| self._cached_wardmodem_responses.append(command) |
| self._verify_and_send_mm_commands() |
| |
| |
| def _post_wardmodem_request(self, command): |
| """ |
| For an AT command, find out the action to be taken on wardmodem and post |
| the action. |
| |
| @param command: AT command for which a request must be posted to |
| wardmodem. |
| |
| @raises: ATTransceiverException if no valid action exists for the given |
| AT command. |
| |
| """ |
| action = self._find_wardmodem_action_for_at(command) |
| state_machine_name, function_name, args = action |
| try: |
| state_machine = self._state_machines[state_machine_name] |
| except KeyError: |
| self._runtime_error( |
| 'Malformed action registered for AT command -- Unknown ' |
| 'state machine. AT command: |%s|. Action: |%s|' % |
| (command, action)) |
| try: |
| function = getattr(state_machine, function_name) |
| except AttributeError: |
| self._runtime_error( |
| 'Malformed action registered for AT command -- Unkonwn ' |
| 'function name. AT command: |%s|. Action: |%s|. Object ' |
| 'dictionary: %s.' % (command, action, dir(state_machine))) |
| |
| self._task_loop.post_task( |
| self._execute_state_machine_function, command, action, function, |
| *args) |
| |
| # ########################################################################## |
| # Helper functions |
| |
| def _execute_state_machine_function(self, at_command, action, function, |
| *args): |
| """ |
| A thin wrapper to execute state_machine.function(args). Instead of |
| posting the call directly, this method is posted for better error |
| reporting in case of failure. |
| |
| @param at_command: The AT command for which this function was called. |
| |
| @param action: The matching wardmodem action which led to this function |
| call. |
| |
| @param function: The function to call. |
| |
| @param *args: Arguments to be passed to function. |
| |
| """ |
| try: |
| function(*args) |
| except TypeError as e: |
| self._logger.error( |
| 'Possible malformed action registered for AT command -- ' |
| 'Incorrect arguments. AT command: |%s|. Action: |%s|. ' |
| 'Expected function signature: %s. ' |
| 'Original error raised: |%s|', |
| at_command, action, inspect.getargspec(function), str(e)) |
| # use 'raise' here to preserve the original backtrace. |
| raise |
| |
| |
| def _update_at_to_wm_action_map(self, raw_map): |
| """ |
| Update the dictionary that maps AT commands and their arguments to the |
| action to be taken by wardmodem. |
| |
| The internal map updated is |
| {at_command, {(arg1, arg2, ...), (state_machine_name, |
| function, |
| (idx1, idx2, ...))}} |
| Here, |
| - at_command [string] is the AT Command received, |
| - (arg1, arg2, ...) [tuple of string] is possibly empty, and |
| specifies the arguments that need to be matched. It may contain |
| the special symbol '*' to mean ignore that argument while |
| matching. |
| - state_machine_name [string] is name of a state machine in the |
| state machine map. |
| - function [string] is a function exported by the state machine |
| mapped to by state_machine_name |
| - (idx1, idx2, ...) [tuple of int] lists the (string) arguments that |
| should be passed on from the AT command to the called function. |
| |
| @param raw_map: The raw map from AT command to function read in from the |
| configuration file. For the format of this map, see the comment |
| at the head of a configuration file. |
| |
| @raises WardModemSetupException if raw_map was not well-formed, and the |
| update failed. Absolutely no guarantees about the state of the |
| map if the update fails. |
| |
| """ |
| for atcom in raw_map: |
| try: |
| at, args = self._parse_at_command(atcom) |
| except wme.ATTransceiverException as e: |
| self._setup_error(e.args) |
| action = self._sanitize_wardmodem_action(raw_map[atcom]) |
| |
| if at not in self._at_to_wm_action_map: |
| self._at_to_wm_action_map[at] = {} |
| if args in self._at_to_wm_action_map[at]: |
| self._logger.debug('Updated at_to_wm_action_map: ' |
| '|%s(%s): [%s --> %s]|', |
| at, args, |
| str(self._at_to_wm_action_map[at][args]), |
| str(action)) |
| else: |
| self._logger.debug('Added to at_to_wm_action_map: |%s(%s): %s|', |
| at, args, str(action)) |
| self._at_to_wm_action_map[at][args] = action |
| |
| |
| def _update_wm_response_to_at_map(self, raw_map): |
| """ |
| Update the dictionary that maps wardmodem responses to AT commands. |
| |
| The internal map updated is of the same form as raw_map: |
| {response_function: at_response} |
| where both response_function and at_response are of type string. |
| at_resposne may contain special placeholder charachters '*'. |
| |
| @param raw_map: The map read in from the configuration file. |
| |
| """ |
| for response_function, at_response in raw_map.iteritems(): |
| if response_function in self._wm_response_to_at_map: |
| self._logger.debug( |
| 'Updated wm_response_to_at_map: |%s: [%s --> %s]|', |
| response_function, |
| self._wm_response_to_at_map[response_function], |
| at_response) |
| else: |
| self._logger.debug( |
| 'Added to wm_response_to_at_map: |%s: %s|', |
| response_function, at_response) |
| self._wm_response_to_at_map[response_function] = at_response |
| |
| |
| def _sanitize_wardmodem_action(self, action): |
| """ |
| Test that the action specified in the AT command --> wardmodem action |
| map is sane and normalize to simplify handling later. |
| |
| Currently, this only checks that the action consists of tuples of the |
| right size / type. It might make sense to make this check a lot stricter |
| so that ill-formed configuration files are caught early. |
| |
| Returns the normalized form: 3-tuple with the last item being a tuple of |
| integers. |
| |
| @param action: The action tuple to check. |
| |
| @return action: Sanitized action tuple. Normalized form is (string, |
| string, (int*)). |
| |
| @raises: WardModemSetupException if action is ill-formed. |
| |
| """ |
| errstr = ('Ill formed action |%s|. Action must be of the form: ' |
| '(state_machine_name, function_name, (index_tuple)) ' |
| 'Here, index_tuple is a tuple of integers.' % str(action)) |
| sanitized_action = [] |
| if type(action) is not tuple: |
| self._setup_error(errstr) |
| if len(action) != 2 and len(action) != 3: |
| self._setup_error(errstr) |
| if type(action[0]) != str or type(action[1]) != str: |
| self._setup_error(errstr) |
| sanitized_action.append(action[0]) |
| sanitized_action.append(action[1]) |
| if len(action) != 3: |
| sanitized_action.append(()) |
| else: |
| if type(action[2]) == tuple: |
| for idx in action[2]: |
| if type(idx) != int: |
| self._setup_error(errstr) |
| sanitized_action.append(action[2]) |
| else: |
| if type(action[2]) != int: |
| self._setup_error(errstr) |
| sanitized_action.append((action[2],)) |
| return tuple(sanitized_action) |
| |
| |
| def _parse_at_command(self, atcom): |
| """ |
| Parse an AT command into the command and its arguments |
| |
| Examples: |
| 'AT?' --> ('AT?', ()) |
| 'AT+XX' --> ('AT+XX', ()) |
| 'AT%SCF=1,2' --> ('AT%SCF=', ('1', '2')) |
| 'ATX=*' --> ('ATX=', ('*',)) |
| |
| @param atcom: [string] the AT command to parse |
| |
| @return: [(string, (string))] A tuple of the AT command proper and a |
| tuple of arguments. If no arguments are present, an empty argument |
| tuple is included. |
| |
| @raises ATTransceiverError if atcom is not well-formed. |
| |
| """ |
| parts = atcom.split('=') |
| if len(parts) > 2: |
| self._runtime_error('Parsing error: |%s|' % atcom) |
| if len(parts) == 1: |
| return (atcom, ()) |
| # Note: Include the trailing '=' in the AT commmand. |
| at = parts[0] + '=' |
| if parts[1] == '': |
| # This was a command of the form 'ATXXX='. |
| # Treat this as having no arguments, instead of a single '' |
| # argument. |
| return (at, ()) |
| else: |
| return (at, tuple(parts[1].split(','))) |
| |
| |
| def _find_wardmodem_action_for_at(self, atcom): |
| """ |
| For the given AT command, find the appropriate action from wardmodem. |
| This will attempt to find a rule matching |atcom|. If that fails, and if |
| |_fallback_state_machine| exists, the default action from this machine |
| is returned. |
| |
| @param atcom: The AT command to find action for. Type: str. |
| |
| @return: Returns the tuple of (state_machine_name, function, |
| (arguments,)) for the corresponding action. The action to be |
| taken is roughly |
| state_machine.function(arguments) |
| Type: (string, string, (string,)) |
| |
| @raises: ATTransceiverException if the at command is ill-formed or we |
| don't have a corresponding action. |
| |
| """ |
| try: |
| at, args = self._parse_at_command(atcom) |
| except wme.ATTransceiverException as e: |
| self._runtime_error( |
| 'Ill formed AT command received. %s' % str(e.args)) |
| if at not in self._at_to_wm_action_map: |
| if self._fallback_state_machine: |
| return (self._fallback_state_machine, |
| self._fallback_machine_function, |
| (atcom,)) |
| self._runtime_error('Unknown AT command: |%s|' % atcom) |
| |
| for candidate_args in self._at_to_wm_action_map[at]: |
| candidate_action = self._at_to_wm_action_map[at][candidate_args] |
| if self._args_match(args, candidate_args): |
| # Found corresponding entry, now replace the indices of the |
| # arguments in the action with actual arguments. |
| machine, function, idxs = candidate_action |
| fargs = [] |
| for idx in idxs: |
| fargs.append(args[idx]) |
| return machine, function, tuple(fargs) |
| |
| if self._fallback_state_machine: |
| return (self._fallback_state_machine, |
| self._fallback_machine_function, |
| (atcom,)) |
| self._runtime_error('Unhandled arguments: |%s|' % atcom) |
| |
| |
| def _args_match(self, args, matches): |
| """ |
| Check whether args are captured by regexp. |
| |
| @param args: A tuple of strings, the arguments to check for inclusion. |
| |
| @param matches: A similar tuple, but may contain the wild-card '*'. |
| |
| @return True if args is represented by regexp, False otherwise. |
| |
| """ |
| if len(args) != len(matches): |
| return False |
| for i in range(len(args)): |
| arg = args[i] |
| match = matches[i] |
| if match == ARG_PLACEHOLDER: |
| return True |
| if arg != match: |
| return False |
| return True |
| |
| def _construct_at_response(self, raw_at, *args): |
| """ |
| Replace palceholders in an AT command template with actual arguments. |
| |
| @param raw_at: An AT command with '*' placeholders where arguments |
| should be provided. |
| |
| @param *args: Arguments to fill in the placeholders in |raw_at|. |
| |
| @return: AT command with placeholders replaced by arguments. |
| |
| @raises: ATTransceiverException if the number of arguments does not |
| match the number of placeholders. |
| |
| """ |
| parts = raw_at.split(ARG_PLACEHOLDER) |
| if len(args) < (len(parts) - 1): |
| self._runtime_error( |
| 'Failed to construct AT response from |%s|. Expected %d ' |
| 'arguments, found %d.' % |
| (raw_at, len(parts) - 1, len(args))) |
| if len(args) > (len(parts) - 1): |
| self._logger.warning( |
| 'Number of arguments in wardmodem response greater than ' |
| 'expected. Some of the arguments from %s will not be used ' |
| 'in the reconstruction of %s', str(args), raw_at) |
| |
| ret = [] |
| for i in range(len(parts) - 1): |
| ret += parts[i] |
| ret += str(args[i]) |
| ret += parts[len(parts) - 1] |
| return ''.join(ret) |
| |
| |
| def _verify_and_send_mm_commands(self): |
| """ |
| While there are corresponding responses from wardmodem and physical |
| modem, verify that they match and respond to modem manager. |
| |
| """ |
| if not self._cached_wardmodem_responses: |
| return |
| elif not self._cached_modem_responses: |
| if self._modem_response_wait_task is not None: |
| return |
| self._modem_response_wait_task = ( |
| self._task_loop.post_task_after_delay( |
| self._modem_response_timed_out, |
| self._modem_response_timeout_milliseconds)) |
| else: |
| if self._modem_response_wait_task is not None: |
| self._task_loop.cancel_posted_task( |
| self._modem_response_wait_task) |
| self._modem_response_wait_task = None |
| self._verify_and_send_mm_command( |
| self._cached_modem_responses.popleft(), |
| self._cached_wardmodem_responses.popleft()) |
| self._verify_and_send_mm_commands() |
| |
| |
| def _verify_and_send_mm_command(self, modem_response, wardmodem_response): |
| """ |
| Verify that the two AT commands match and respond to modem manager. |
| |
| @param modem_response: AT command response of the physical modem. |
| |
| @param wardmodem_response: AT command response of wardmodem. |
| |
| """ |
| # TODO(pprabhu) This can not handle unsolicited commands yet. |
| # Unsolicited commands from either of the modems will push the lists out |
| # of sync. |
| if wardmodem_response != modem_response: |
| self._logger.warning('Response verification failed.') |
| self._logger.warning('modem response: |%s|', modem_response) |
| self._logger.warning('wardmodem response: |%s|', wardmodem_response) |
| self._logger.warning('wardmodem response takes precedence.') |
| self._report_verification_failure( |
| self.VERIFICATION_FAILED_MISMATCH, |
| modem_response, |
| wardmodem_response) |
| self._logger.debug('Command {[] ==> mm}: |%s|' , wardmodem_response) |
| self._mm_channel.send(wardmodem_response) |
| |
| |
| def _modem_response_timed_out(self): |
| """ |
| Callback called when we time out waiting for physical modem response for |
| some wardmodem response. Can't do much -- log physical modem failure and |
| forward wardmodem response anyway. |
| |
| """ |
| assert (not self._cached_modem_responses and |
| self._cached_wardmodem_responses) |
| wardmodem_response = self._cached_wardmodem_responses.popleft() |
| self._logger.warning('modem response timed out. ' |
| 'Forwarding wardmodem response |%s| anyway.', |
| wardmodem_response) |
| self._logger.debug('Command {[] ==> mm}: |%s|' , wardmodem_response) |
| self._report_verification_failure( |
| self.VERIFICATION_FAILED_TIME_OUT, |
| None, |
| wardmodem_response) |
| self._mm_channel.send(wardmodem_response) |
| self._modem_response_wait_task = None |
| self._verify_and_send_mm_commands() |
| |
| |
| def _report_verification_failure(self, failure, modem_response, |
| wardmodem_response): |
| """ |
| Failure to verify the wardmodem response will call this non-public |
| method. |
| |
| At present, it is only used by unittests to detect failure. |
| |
| @param failure: The cause of failure. Must be one of |
| VERIFICATION_FAILED_MISMATCH or VERIFICATION_FAILED_TIME_OUT. |
| |
| @param modem_response: The received modem response (if any). |
| |
| @param wardmodem_response: The received wardmodem response. |
| |
| """ |
| pass |
| |
| |
| def _runtime_error(self, error_message): |
| """ |
| Log the message at error level and raise ATTransceiverException. |
| |
| @param error_message: The error message. |
| |
| @raises: ATTransceiverException. |
| |
| """ |
| self._logger.error(error_message) |
| raise wme.ATTransceiverException(error_message) |
| |
| |
| def _setup_error(self, error_message): |
| """ |
| Log the message at error level and raise WardModemSetupException. |
| |
| @param error_message: The error message. |
| |
| @raises: WardModemSetupException. |
| |
| """ |
| self._logger.error(error_message) |
| raise wme.WardModemSetupException(error_message) |
