client/site_tests/cellular_ScanningProperty/files/modems.py

# Copyright (c) 2014 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 logging

import common
from autotest_lib.client.cros.cellular.pseudomodem import modem_3gpp
from autotest_lib.client.cros.cellular.pseudomodem import pm_constants
from autotest_lib.client.cros.cellular.pseudomodem import state_machine
from autotest_lib.client.cros.cellular.pseudomodem import state_machine_factory

class InteractiveStateMachineFactory(state_machine_factory.StateMachineFactory):
    """ Run relevant state machines in interactive mode. """
    def __init__(self):
        super(InteractiveStateMachineFactory, self).__init__()
        self.SetInteractive(pm_constants.STATE_MACHINE_ENABLE)
        self.SetInteractive(pm_constants.STATE_MACHINE_REGISTER)


class ScanMachine(state_machine.StateMachine):
    """
    Handle shill initiated 3GPP scan request.

    A simple machine that allows the test to hook into the Scan asynchronous
    call.

    """
    # State machine states.
    SCAN_STATE = 'Scan'
    DONE_STATE = 'Done'

    def __init__(self, modem):
        super(ScanMachine, self).__init__(modem)
        self._state = ScanMachine.SCAN_STATE


    def _HandleScanState(self):
        """ The only real state in this machine. """
        self._modem.DoScan()
        self._state = ScanMachine.DONE_STATE
        return True


    def _GetCurrentState(self):
        return self._state


    def _GetModemStateFunctionMap(self):
        return {
                ScanMachine.SCAN_STATE: ScanMachine._HandleScanState,
                # ScanMachine.DONE_STATE is the final state. So, no handler.
        }


    def _ShouldStartStateMachine(self):
        return True


class ScanStateMachineFactory(state_machine_factory.StateMachineFactory):
    """ Extend StateMachineFactory to create an interactive ScanMachine. """
    def ScanMachine(self, *args, **kwargs):
        """ Create a ScanMachine when needed in the modem. """
        machine = ScanMachine(*args, **kwargs)
        machine.EnterInteractiveMode(self._bus)
        return machine


class AsyncScanModem(modem_3gpp.Modem3gpp):
    """ 3GPP modem that uses ScanMachine for the Scan call. """
    def __init__(self):
        super(AsyncScanModem, self).__init__(
                state_machine_factory=ScanStateMachineFactory())


    def Scan(self, return_cb, raise_cb):
        """ Overriden from Modem3gpp. """
        # Stash away the scan_ok callback for when the Scan finishes.
        logging.debug('Network scan initiated.')
        self._scan_ok_callback = return_cb
        self._scan_failed_callback = raise_cb
        self._scan_machine = self._state_machine_factory.ScanMachine(self)
        self._scan_machine.Start()


    def DoScan(self):
        """ Defer to Modem3gpp to take the original |SyncScan| action. """
        # We're done scanning, drop |_scan_machine| reference.
        self._scan_machine = None
        try:
            scan_result = super(AsyncScanModem, self).SyncScan()
        except dbus.exceptions.DBusException as e:
            logging.warning('Network scan failed')
            self._scan_failed_callback(e)
            return

        logging.debug('Network scan completed.')
        self._scan_ok_callback(scan_result)