1 # Copyright (c) 2014 The Chromium OS Authors. All rights reserved. 2 # Use of this source code is governed by a BSD-style license that can be 3 # found in the LICENSE file. 4 5 import logging 6 7 import common 8 from autotest_lib.client.cros.cellular.pseudomodem import modem_3gpp 9 from autotest_lib.client.cros.cellular.pseudomodem import pm_constants 10 from autotest_lib.client.cros.cellular.pseudomodem import state_machine 11 from autotest_lib.client.cros.cellular.pseudomodem import state_machine_factory 12 13 class InteractiveStateMachineFactory(state_machine_factory.StateMachineFactory): 14 """ Run relevant state machines in interactive mode. """ 15 def __init__(self): 16 super(InteractiveStateMachineFactory, self).__init__() 17 self.SetInteractive(pm_constants.STATE_MACHINE_ENABLE) 18 self.SetInteractive(pm_constants.STATE_MACHINE_REGISTER) 19 20 21 class ScanMachine(state_machine.StateMachine): 22 """ 23 Handle shill initiated 3GPP scan request. 24 25 A simple machine that allows the test to hook into the Scan asynchronous 26 call. 27 28 """ 29 # State machine states. 30 SCAN_STATE = 'Scan' 31 DONE_STATE = 'Done' 32 33 def __init__(self, modem): 34 super(ScanMachine, self).__init__(modem) 35 self._state = ScanMachine.SCAN_STATE 36 37 38 def _HandleScanState(self): 39 """ The only real state in this machine. """ 40 self._modem.DoScan() 41 self._state = ScanMachine.DONE_STATE 42 return True 43 44 45 def _GetCurrentState(self): 46 return self._state 47 48 49 def _GetModemStateFunctionMap(self): 50 return { 51 ScanMachine.SCAN_STATE: ScanMachine._HandleScanState, 52 # ScanMachine.DONE_STATE is the final state. So, no handler. 53 } 54 55 56 def _ShouldStartStateMachine(self): 57 return True 58 59 60 class ScanStateMachineFactory(state_machine_factory.StateMachineFactory): 61 """ Extend StateMachineFactory to create an interactive ScanMachine. """ 62 def ScanMachine(self, *args, **kwargs): 63 """ Create a ScanMachine when needed in the modem. """ 64 machine = ScanMachine(*args, **kwargs) 65 machine.EnterInteractiveMode(self._bus) 66 return machine 67 68 69 class AsyncScanModem(modem_3gpp.Modem3gpp): 70 """ 3GPP modem that uses ScanMachine for the Scan call. """ 71 def __init__(self): 72 super(AsyncScanModem, self).__init__( 73 state_machine_factory=ScanStateMachineFactory()) 74 75 76 def Scan(self, return_cb, raise_cb): 77 """ Overriden from Modem3gpp. """ 78 # Stash away the scan_ok callback for when the Scan finishes. 79 logging.debug('Network scan initiated.') 80 self._scan_ok_callback = return_cb 81 self._scan_failed_callback = raise_cb 82 self._scan_machine = self._state_machine_factory.ScanMachine(self) 83 self._scan_machine.Start() 84 85 86 def DoScan(self): 87 """ Defer to Modem3gpp to take the original |SyncScan| action. """ 88 # We're done scanning, drop |_scan_machine| reference. 89 self._scan_machine = None 90 try: 91 scan_result = super(AsyncScanModem, self).SyncScan() 92 except dbus.exceptions.DBusException as e: 93 logging.warning('Network scan failed') 94 self._scan_failed_callback(e) 95 return 96 97 logging.debug('Network scan completed.') 98 self._scan_ok_callback(scan_result) 99
