# 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.

# DESCRIPTION:
# This test is used to verify the functionality of bluetooth device.
# The functionality under test are:
# 1. Detect the specified number of bluetooth adapter on dut.
# 2. Scan remote bluetooth device and try to find at least one device.
# 3. If a remote device keyword is given, the test will only care
#    the devices whose 'Name' contains keyword. This applies to item 4 as well.
# 4. If an RSSI threshold value is given, check that the largest average RSSI
#    among all scanned devices >= threshold.
# 5. Try to pair and connect with the bluetooth input device. Now it supports
#    mouse.
# Check the ARGS in BluetoothTest for the detail of arguments.

import glob
import logging
import os
import sys
import threading
import time
import unittest

import factory_common  # pylint: disable=unused-import
from cros.factory.device import device_utils
from cros.factory.test import event_log
from cros.factory.test import factory
from cros.factory.test import factory_task
from cros.factory.test.i18n import test_ui as i18n_test_ui
from cros.factory.test import shopfloor
from cros.factory.test import test_ui
from cros.factory.test import ui_templates
from cros.factory.test.utils import bluetooth_utils
from cros.factory.utils.arg_utils import Arg
from cros.factory.utils import process_utils
from cros.factory.utils import sync_utils


_TEST_TITLE = i18n_test_ui.MakeI18nLabel('Bluetooth functional Test')
_MSG_DETECT_ADAPTER = i18n_test_ui.MakeI18nLabel('Detect bluetooth adapter')
_MSG_TURN_ON_DEVICE = i18n_test_ui.MakeI18nLabelWithClass(
    'Enable the connection ability of bluetooth device and press Enter',
    'start-font-size')
_MSG_INTO_FIXTURE = i18n_test_ui.MakeI18nLabelWithClass(
    'Place the base into the fixture, '
    'and press the space key on the test host.', 'start-font-size')
_MSG_RESET_MAGNET = i18n_test_ui.MakeI18nLabelWithClass(
    'Please re-attach the magnet, and press the space key on the test host.',
    'start-font-size')
_MSG_START_CHARGE = i18n_test_ui.MakeI18nLabelWithClass(
    'Turn on charging by pressing the green button, '
    'take the keyboard out and put it back, '
    'and press the space key on the test host.', 'start-font-size')
_MSG_READ_BATTERY_1 = i18n_test_ui.MakeI18nLabelWithClass(
    'Read battery level for the 1st time.', 'start-font-size')
_MSG_READ_BATTERY_2 = i18n_test_ui.MakeI18nLabelWithClass(
    'Read battery level for the 2nd time.', 'start-font-size')
_MSG_BATTERY_CHARGE_TEST = i18n_test_ui.MakeI18nLabelWithClass(
    'Check if the battery has charged to a higher percentage',
    'start-font-size')
_MSG_CHECK_BATTERY_LEVEL = i18n_test_ui.MakeI18nLabelWithClass(
    'Check battery level.', 'start-font-size')
_MSG_STOP_CHARGE = i18n_test_ui.MakeI18nLabelWithClass(
    'Press the green button again to stop charging, '
    'and press the space key on the test host.', 'start-font-size')
_MSG_OUT_OF_FIXTURE = i18n_test_ui.MakeI18nLabelWithClass(
    'Take the base out of the fixture, '
    'and press the space key on the test host.', 'start-font-size')
_MSG_READ_FIRMWARE_REVISION_STRING = i18n_test_ui.MakeI18nLabelWithClass(
    'Read firmware revision string.', 'start-font-size')
_MSG_SCAN_DEVICE = i18n_test_ui.MakeI18nLabelWithClass(
    'Scanning...', 'start-font-size')
_MSG_DETECT_RSSI = lambda count, total: (
    i18n_test_ui.MakeI18nLabelWithClass(
        'Detect RSSI (count {count}/{total})',
        'start-font-size', count=count, total=total))
_MSG_TURN_ON_INPUT_DEVICE = i18n_test_ui.MakeI18nLabelWithClass(
    'Enable the connection ability of input bluetooth device and press Enter',
    'start-font-size')
_MSG_PAIR_INPUT_DEVICE = i18n_test_ui.MakeI18nLabelWithClass(
    'Pairing to input device now...', 'start-font-size')
_MSG_UNPAIR = i18n_test_ui.MakeI18nLabelWithClass(
    'Press shift-p-a-i-r simultaneously on the base.', 'start-font-size')
_MSG_CONNECT_INPUT_DEVICE = i18n_test_ui.MakeI18nLabelWithClass(
    'Connecting to input device now...', 'start-font-size')
_MSG_TEST_INPUT = i18n_test_ui.MakeI18nLabelWithClass(
    'Please test input. Press Escape to fail and Enter to pass',
    'start-font-size')
_MSG_UNPAIRING = i18n_test_ui.MakeI18nLabelWithClass(
    'Unpairing', 'start-font-size')
_MSG_AUTH_FAILED = i18n_test_ui.MakeI18nLabelWithClass(
    'Authentication failed, retrying...', 'start-font-size')


INPUT_MAX_RETRY_TIMES = 10
INPUT_RETRY_INTERVAL = 1
RESET_ADAPTER_SLEEP_TIME = 5
READ_BATTERY_MAX_RETRY_TIMES = 10

BATTERY_LEVEL_KEY = 'battery_level'
READ_BATTERY_STEP_1 = 'read_battery_1'
READ_BATTERY_STEP_2 = 'read_battery_2'


def GetCurrentTime():
  """Get the current time."""
  return time.strftime('%Y-%m-%d %H:%M:%S')


def ColonizeMac(mac):
  """ Given a MAC address, normalize its colons.

  Example: ABCDEF123456 -> AB:CD:EF:12:34:56
  """
  mac_no_colons = ''.join(mac.strip().split(':'))
  groups = [mac_no_colons[x:x+2] for x in range(0, len(mac_no_colons), 2)]
  return ':'.join(groups)


def MakePasskeyLabelPrompt(passkey):
  """Creates a label prompting the operator to enter a passkey"""
  return i18n_test_ui.MakeI18nLabelWithClass(
      'Enter passkey {key} then press enter on the base.',
      'start-font-size',
      key=passkey)


def CheckInputCount():
  """Returns the number of input devices from probing /dev/input/event*."""
  number_input = len(glob.glob('/dev/input/event*'))
  logging.info('Found %d input devices.', number_input)
  return number_input


def WaitForInputCount(task, expected_input_count, timeout=10):
  """ Waits for the number of input devices to reach the given count.

  Returns true if the input count reaches the given amount, false
  otherwise. On failure it fails the task.

  Args:
   task: The task that may be Failed
   expected_input_count: The number of input devices that determines success
   timeout: The maximum time in seconds that we will wait
  """
  end_time = time.time() + timeout
  while time.time() < end_time:
    input_count = CheckInputCount()
    if input_count == expected_input_count:
      return True
    time.sleep(0.2)
  task.Fail('Input device count %d is different than expected %d.' %
            (input_count, expected_input_count))
  return False


def _ResetAdapter():
  """Reset the adapter every time when using the BT device.
  This is because the adapter may be down anytime for some unknown reason.
  """
  cmd = 'hciconfig hci0 reset'
  factory.console.info('Reset adapter and wait 5 seconds....: %s', cmd)
  process_utils.Spawn(cmd.split(), log=True, check_call=True)
  time.sleep(RESET_ADAPTER_SLEEP_TIME)


def _SaveLocalLog(log_file, data):
  """Save the log locally on a test host."""
  log_dir = os.path.dirname(log_file)
  if not os.path.isdir(log_dir):
    os.makedirs(log_dir)
  with open(log_file, 'a') as log:
    log.write(str(data))


def _SaveAuxLogOnShopfloor(aux_log_file, data):
  """Save the local log file to shopfloor."""
  try:
    shopfloor_client = shopfloor.GetShopfloorConnection()
    shopfloor_client.SaveAuxLog(aux_log_file, str(data))
  except Exception as e:
    # It is only a logging error. Do not fail the test.
    logging.warning('Save aux log failure: %s', e)


def _SaveLogs(log_file, aux_log_file, data):
  """Save the log files on the local test host and on the shopfloor."""
  # Prepend the current timestamp to each line.
  data = ''.join([(GetCurrentTime() + ' ' + line + '\n') if line else '\n'
                  for line in data.splitlines()])
  if log_file:
    _SaveLocalLog(log_file, data)
    if aux_log_file:
      with open(log_file) as log:
        _SaveAuxLogOnShopfloor(aux_log_file, log.read())


def RetryWithProgress(template, template_message, action_string,
                      max_retry_times, retry_interval, target, *args, **kwargs):
  """Runs target function with retries and shows retry times on progress bar.

  Args:
    template: an ui_template
    template_message: The message to show when target function is running.
    action_string: The string to describe the action in logging.
    max_retry_times: the maximal retry times
    retry_interval: the interval between retries
    target: The target function. *args and **kwargs will be passed to target.

  Returns:
    Return the return value of the target function.
  """
  def _UpdateProgressBar(retry_time, max_retry_time):
    """Updates the progress bar according to retry_time and max_retry_time."""
    msg = 'Update progress bar with retry time: %d, max retry time: %d.'
    logging.info(msg, retry_time, max_retry_time)
    template.SetProgressBarValue(int(100 * retry_time / max_retry_time))

  template.SetState(template_message)
  template.DrawProgressBar()
  target_result = sync_utils.Retry(max_retry_times, retry_interval,
                                   _UpdateProgressBar, target, *args, **kwargs)
  template.SetProgressBarValue(100)
  log_msg = '%s was done.' if target_result else '%s failed.'
  logging.info(log_msg, action_string)
  return target_result


class DetectAdapterTask(factory_task.FactoryTask):
  """The task checking number of adapters.

  Detects adapters from dbus and checks if the number of adapters matches the
  expected number.

  Args:
     expected_adapter_count: The expected number of bluetooth adapters.
  """

  def __init__(self, test, expected_adapter_count):
    super(DetectAdapterTask, self).__init__()
    self._test = test
    self._expected_adapter_count = expected_adapter_count

  def Run(self):
    self._test.template.SetState(_MSG_DETECT_ADAPTER)
    adapters = self._test.dut.bluetooth.GetAdapters(
        self._test.args.detect_adapters_retry_times,
        self._test.args.detect_adapters_interval_secs)
    if len(adapters) == self._expected_adapter_count:
      self.Pass()
    else:
      self.Fail('DetectAdapter: expect %d and find %d adapter(s).' %
                (self._expected_adapter_count, len(adapters)))


class TurnOnTask(factory_task.FactoryTask):
  """The task to ask operator to turn on bluetooth device and press enter.

  Args:
    message: Html code containing message to show on the screen.
  """

  def __init__(self, test, message, key=test_ui.ENTER_KEY):
    super(TurnOnTask, self).__init__()
    self._test = test
    self._message = message
    self._key = key

  def Cleanup(self):
    """Unbinds the Enter key after this task is done."""
    self._test.ui.UnbindKey(self._key)

  def Run(self):
    self._test.template.SetState(self._message)
    self._test.ui.AppendCSS('.start-font-size {font-size: 2em;}')
    self._test.ui.BindKey(self._key, lambda _: self.Pass())
    logging.info('wait for the user to press a key')


def SetAndStartScanProgressBar(template, timeout_secs, scan_event=None):
  """Control progress bar fo a duration of timeout_secs."""
  def UpdateProgressBar():
    """Updates progress bar for a duration of timeout_secs"""
    start_time = time.time()
    end_time = start_time + timeout_secs
    while time.time() < end_time:
      if scan_event and scan_event.isSet():
        break
      template.SetProgressBarValue(int(
          100 * (time.time() - start_time) / timeout_secs))
      time.sleep(0.2)
    template.SetProgressBarValue(100)
    logging.debug('UpdateProgressBar stopped.')

  template.DrawProgressBar()
  return process_utils.StartDaemonThread(
      target=UpdateProgressBar, name='ProgressThread')


class ScanDevicesTask(factory_task.FactoryTask):
  """The task to scan bluetooth devices around.

  In this task, the test will control the first adapter from BluetoothManager
  and scan devices around for timeout_secs. The task passed if there is at least
  one device.
  If target_addresses is provided, the test will also check if it can find
  at least one device specified in target_addresses list.
  This passes the strongest matching device mac to _test.SetStrongestRssiMac

  Note: this task is intended to be executed on a DUT, i.e., a chromebook,
  to test its bluetooth module. A typical test case is to see if it can detect
  a bluetooth mouse placed around it.
  """

  def __init__(self, test):
    super(ScanDevicesTask, self).__init__()
    self._test = test
    self._keyword = test.args.keyword
    self._average_rssi_threshold = test.args.average_rssi_threshold
    self._mac_to_scan = test.GetInputDeviceMac()
    self._scan_counts = test.args.scan_counts
    self._timeout_secs = test.args.scan_timeout_secs

    self._progress_thread = None

  def FilterByKeyword(self, devices):
    """Returns the devices filtered by self._keyword.

    If self._keyword is None, leave devices as it is.
    """
    if self._keyword is None:
      return devices

    filtered_devices = {}
    for mac, props in devices.iteritems():
      if 'Name' not in props:
        logging.warning('Device %s: %s does not have "Name" property.',
                        mac, props)
        continue

      if self._keyword in props['Name']:
        filtered_devices[mac] = props
        logging.info('Device %s: "Name" property %s matches keyword %s.',
                     mac, props['Name'], self._keyword)
    return filtered_devices

  def UpdateRssi(self, devices_rssis, devices):
    """Updates devices_rssis using RSSI property in devices.

    Args:
      devices_rssis: A dict. Keys are mac addresses and values are lists of
        scanned RSSI value.
      devices: A dict. Keys are mac addresses and values are dicts of
        properties.
    """
    for mac, props in devices.iteritems():
      if 'RSSI' not in props:
        logging.warning('Device %s: %s does not have "RSSI" property.',
                        mac, props)
        continue
      if mac in devices_rssis:
        devices_rssis[mac].append(props['RSSI'])
      else:
        devices_rssis[mac] = [props['RSSI']]
    logging.info('UpdateRssi: %s', devices_rssis)

  def Run(self):
    bluetooth_manager = self._test.dut.bluetooth
    adapter = bluetooth_manager.GetFirstAdapter(self._test.host_mac)

    # Records RSSI of each scan and calculates average rssi.
    candidate_rssis = {}

    # Helper to check if the target MAC has been scanned
    def has_scanned_target_mac():
      return self._mac_to_scan and self._mac_to_scan in candidate_rssis

    for _ in xrange(self._scan_counts):
      self._test.template.SetState(_MSG_SCAN_DEVICE)
      self._progress_thread = SetAndStartScanProgressBar(self._test.template,
                                                         self._timeout_secs)
      devices = bluetooth_manager.ScanDevices(adapter, self._timeout_secs)
      self._progress_thread.join()

      logging.info('Found %d device(s).', len(devices))
      for mac, props in devices.iteritems():
        try:
          logging.info('Device found: %s. Name: %s, RSSI: %d',
                       mac, props['Name'], props['RSSI'])
        except KeyError:
          logging.exception('Name or RSSI is not available in %s', mac)

      self.UpdateRssi(candidate_rssis, self.FilterByKeyword(devices))
      # Optimization: if we are only interested in one particular address,
      # then we can early-out as soon as we find it
      if self._average_rssi_threshold is None and has_scanned_target_mac():
        logging.info("Address found, ending scan early")
        break

    logging.info('Found %d candidate device(s) in %s scans.',
                 len(candidate_rssis), self._scan_counts)
    factory.console.info('Candidate devices scan results: %s',
                         dict((str(k), [int(r) for r in v])
                              for k, v in candidate_rssis.iteritems()))

    if len(candidate_rssis) == 0:
      self.Fail('ScanDevicesTask: Fail to find any candidate device.')
      return

    # Calculates maximum average RSSI.
    max_average_rssi_mac, max_average_rssi = None, -sys.float_info.max
    for mac, rssis in candidate_rssis.iteritems():
      # typecast to str to avoid the weird dbus.String type
      mac = str(mac)
      average_rssi = float(sum(rssis)) / len(rssis)
      logging.info('Device %s has average RSSI: %f', mac, average_rssi)
      event_log.Log('avg_rssi', mac=mac, average_rssi=average_rssi)
      if average_rssi > max_average_rssi:
        max_average_rssi_mac, max_average_rssi = mac, average_rssi

    logging.info('Device %s has the largest average RSSI: %f',
                 max_average_rssi_mac, max_average_rssi)

    event_log.Log('bluetooth_scan_device', mac=str(max_average_rssi_mac),
                  rssi=float(max_average_rssi),
                  meet=max_average_rssi >= self._average_rssi_threshold)

    self._test.SetStrongestRssiMac(max_average_rssi_mac)

    if self._mac_to_scan and not has_scanned_target_mac():
      found_addresses = [str(k) for k in candidate_rssis]
      self.Fail('Failed to find MAC address %s.'
                'Scanned addresses: %s' % (self._mac_to_scan, found_addresses))
      return

    if self._average_rssi_threshold is None:
      # Test is uninterested in RSSI thresholds
      self.Pass()
    elif self._average_rssi_threshold > max_average_rssi:
      factory.console.error('The largest average RSSI %f does not meet'
                            ' threshold %f. Please ensure that the test BT '
                            'device is \'visible\' and close to the DUT '
                            'antenna.',
                            max_average_rssi, self._average_rssi_threshold)
      self.Fail('ScanDeviceTask: The largest average RSSI %f of device %s does'
                ' not meet threshold %f.' % (
                    max_average_rssi, str(max_average_rssi_mac),
                    self._average_rssi_threshold))
    else:
      factory.console.info('The largest average RSSI %f meets threshold %f.',
                           max_average_rssi, self._average_rssi_threshold)
      self.Pass()


class DetectRSSIofTargetMACTask(factory_task.FactoryTask):
  """The task to detect the RSSI strength at a given target MAC address.

  In this task, a generic test host uses the first adapter from
  BluetoothManager and scans devices around for timeout_secs. The task
  passed if it can detect the RSSI strength at the target MAC.

  Note: this task is intended to be executed on a generic test host to test
  if the RSSI of a target device, e.g., a Ryu base, could be detected.
  """

  def __init__(self, test):
    super(DetectRSSIofTargetMACTask, self).__init__()
    self._test = test
    self._mac_to_scan = test.GetInputDeviceMac()
    self._scan_counts = test.args.scan_counts
    self._timeout_secs = test.args.scan_timeout_secs
    self._input_device_rssi_key = test.args.input_device_rssi_key
    self._progress_thread = None
    self._scan_rssi_event = threading.Event()
    self.fail_msg = ''
    self._average_rssi_lower_threshold = None
    self._average_rssi_upper_threshold = None

  def _DeriveRSSIThreshold(self, threshold, fid):
    if isinstance(threshold, (int, float)):
      return threshold
    elif isinstance(threshold, dict):
      if fid in threshold:
        return threshold.get(fid)
      else:
        self.fail_msg += 'Fixture ID "%s" is not legitimate!\n' % fid
    else:
      self.fail_msg += 'Wrong type of RSSI threshold: %s\n' % str(threshold)

  def Run(self):
    fid = event_log.GetDeviceId()
    self._average_rssi_lower_threshold = self._DeriveRSSIThreshold(
        self._test.args.average_rssi_lower_threshold, fid)
    self._average_rssi_upper_threshold = self._DeriveRSSIThreshold(
        self._test.args.average_rssi_upper_threshold, fid)
    if self.fail_msg:
      factory.console.error(self.fail_msg)
      self.Fail(self.fail_msg)
      return

    bluetooth_manager = self._test.dut.bluetooth
    adapter = bluetooth_manager.GetFirstAdapter(self._test.host_mac)
    logging.info('mac (%s): %s', self._test.host_mac, adapter)

    rssis = []
    for i in xrange(1, 1 + self._scan_counts):
      label = _MSG_DETECT_RSSI(i, self._scan_counts)
      self._test.template.SetState(label)
      self._scan_rssi_event.clear()
      self._progress_thread = SetAndStartScanProgressBar(self._test.template,
                                                         self._timeout_secs,
                                                         self._scan_rssi_event)

      devices = bluetooth_manager.ScanDevices(adapter,
                                              timeout_secs=self._timeout_secs,
                                              match_address=self._mac_to_scan)

      self._scan_rssi_event.set()
      self._progress_thread.join()
      for mac, props in devices.iteritems():
        if mac == self._mac_to_scan and 'RSSI' in props:
          factory.console.info('RSSI of count %d: %.2f', i, props['RSSI'])
          rssis.append(props['RSSI'])

    if len(rssis) == 0:
      self.Fail('DetectRSSIofTargetMACTask: Fail to get RSSI from device %s.' %
                self._mac_to_scan)
    else:
      average_rssi = float(sum(rssis)) / len(rssis)
      factory.set_shared_data(self._input_device_rssi_key, average_rssi)
      logging.info('RSSIs at MAC %s: %s', self._mac_to_scan, rssis)
      factory.console.info('Average RSSI: %.2f', average_rssi)

      fail_msg = ''
      if (self._average_rssi_lower_threshold is not None and
          average_rssi < self._average_rssi_lower_threshold):
        fail_msg += ('Average RSSI %.2f less than the lower threshold %.2f\n' %
                     (average_rssi, self._average_rssi_lower_threshold))
      if (self._average_rssi_upper_threshold is not None and
          average_rssi > self._average_rssi_upper_threshold):
        fail_msg += ('Average RSSI %.2f greater than the upper threshold %.2f' %
                     (average_rssi, self._average_rssi_upper_threshold))

      # Convert dbus.Int16 in rssis below to regular integers.
      status = (('pass' if fail_msg == '' else 'fail') +
                ' exp: [%.2f, %.2f]' % (self._average_rssi_lower_threshold,
                                        self._average_rssi_upper_threshold))
      data = ('Average RSSI: %.2f %s  (%s)\n' %
              (average_rssi, map(int, rssis), status))
      _SaveLogs(self._test.log_file, self._test.aux_log_file, data)

      if fail_msg:
        factory.console.error(fail_msg)
        self.Fail(fail_msg)
      else:
        self.Pass()


class UnpairTask(factory_task.FactoryTask):
  """A task to unpair from bluetooth devices.

  Args:
    device_mac: None, or the MAC address of the device to unpair
    name_fragment: None, or a substring of the name of the device(s) to unpair
  """

  def __init__(self, test, device_mac, name_fragment):
    super(UnpairTask, self).__init__()
    self._test = test
    self._device_mac = device_mac
    self._name_fragment = name_fragment

  def _ShouldUnpairDevice(self, device_props):
    """Indicate if a device matches the filter, and so should be unpaired

    If a name fragment or MAC address is given, the corresponding property
    must match. If neither is given, all devices should be unpaired.
    """
    if self._device_mac and device_props["Address"] != self._device_mac:
      return False
    if (self._name_fragment and
        self._name_fragment not in device_props.get('Name', '')):
      return False
    return device_props["Paired"]

  def Run(self):
    self._test.template.SetState(_MSG_UNPAIRING)
    self._test.ui.AppendCSS('.start-font-size {font-size: 2em;}')

    input_count_before_unpair = CheckInputCount()
    bluetooth_manager = self._test.dut.bluetooth
    adapter = bluetooth_manager.GetFirstAdapter(self._test.host_mac)
    devices = bluetooth_manager.GetAllDevices(adapter).values()
    devices_to_unpair = filter(self._ShouldUnpairDevice, devices)
    logging.info('Unpairing %d device(s)', len(devices_to_unpair))
    for device_to_unpair in devices_to_unpair:
      address = device_to_unpair["Address"]
      bluetooth_manager.DisconnectAndUnpairDevice(adapter, address)
      bluetooth_manager.RemovePairedDevice(adapter, address)

    # Check that we unpaired what we thought we did
    expected_input_count = input_count_before_unpair - len(devices_to_unpair)
    if WaitForInputCount(self, expected_input_count):
      self.Pass()


class CheckDisconnectionOfPairedDeviceTask(factory_task.FactoryTask):
  """A task to check whether a paired device has disconnected.

  Args:
    device_mac: None, or the MAC address of the device to unpair
    name_fragment: None, or a substring of the name of the device(s) to unpair
  """

  def __init__(self, test, device_mac):
    super(CheckDisconnectionOfPairedDeviceTask, self).__init__()
    self._test = test
    self._device_mac = device_mac

  def _ConnectedDevice(self, device_props):
    """Indicates if a device matches the filter, and so should be unpaired

    If a name fragment or MAC address is given, the corresponding property
    must match. If neither is given, all devices should be unpaired.
    """
    return (device_props["Address"] == self._device_mac and
            int(device_props["Connected"]) >= 1)

  def _CheckDisconnection(self):
    bluetooth_manager = self._test.dut.bluetooth
    adapter = bluetooth_manager.GetFirstAdapter(self._test.host_mac)
    devices = bluetooth_manager.GetAllDevices(adapter).values()
    connected_devices = filter(self._ConnectedDevice, devices)
    logging.info('Connected and paired %d device(s)', len(connected_devices))
    return len(connected_devices) == 0

  def Run(self):
    flag_disconnection = RetryWithProgress(
        self._test.template, _MSG_UNPAIR,
        'Check disconnection of the paired base',
        INPUT_MAX_RETRY_TIMES, INPUT_RETRY_INTERVAL,
        self._CheckDisconnection)
    if flag_disconnection:
      msg = 'Shift-P-A-I-R: done'
      self.Pass()
    else:
      msg = 'Shift-P-A-I-R: not done'
      self.Fail(msg)
    factory.console.info(msg)
    _SaveLogs(self._test.log_file, self._test.aux_log_file, msg)


def _ExecuteFixtureMethod(fixture, operation, post_sleep=0):
  """Execute a method of the charge test fixture."""
  # An operation is mapped to its corresponding fixture method defined in
  # base_charge_fixture.BaseChargeFixture class.
  FIXTURE_METHOD_DICT = {'START_CHARGING': 'StartCharging',
                         'STOP_CHARGING': 'StopCharging',
                         'ENABLE_MAGNET': 'EnableMagnet',
                         'DISABLE_MAGNET': 'DisableMagnet'}
  fixture_method = getattr(fixture, FIXTURE_METHOD_DICT.get(operation))
  factory.console.info('Executing fixture method: %s', fixture_method.__name__)
  fixture_method()
  time.sleep(post_sleep)


class FixtureControlTask(factory_task.FactoryTask):
  """The task to control the charge test fixture.

  Args:
    operation: the operation to be performed by the test fixture.
  """

  def __init__(self, test, operation, post_sleep=0):
    # pylint: disable=super-init-not-called
    self._fixture = test.fixture
    self._operation = operation
    self._post_sleep = post_sleep

  def Run(self):
    try:
      _ExecuteFixtureMethod(self._fixture, self._operation,
                            post_sleep=self._post_sleep)
      self.Pass()
    except Exception as e:
      self.Fail('error in executing %s (%s)' % (self._operation, e))


def _SaveLocalBatteryLog(base_enclosure_serial_number, mac, step,
                         battery_level, log_filename):
  """Save the battery log on the local test host."""
  with open(log_filename, 'a') as f:
    f.write('%s %s %s [%s]: %s\n' %
            (GetCurrentTime(), base_enclosure_serial_number, mac, step,
             battery_level))


class ReadBatteryLevelTask(factory_task.FactoryTask):
  """A class to read battery level."""

  MSG_DICT = {READ_BATTERY_STEP_1: _MSG_READ_BATTERY_1,
              READ_BATTERY_STEP_2: _MSG_READ_BATTERY_2}

  def __init__(self, test, mac, step):
    super(ReadBatteryLevelTask, self).__init__()
    self._test = test
    self._mac = mac
    self._step = step

  def Run(self):
    self._test.template.SetState(self.MSG_DICT.get(self._step))

    factory.console.info('%s via %s ...', self._step, self._test.hci_device)
    try:
      battery_level = int(RetryWithProgress(
          self._test.template, self.MSG_DICT.get(self._step), self._step,
          INPUT_MAX_RETRY_TIMES, INPUT_RETRY_INTERVAL,
          bluetooth_utils.GattTool.GetDeviceInfo,
          self._mac, 'battery level', hci_device=self._test.hci_device,
          timeout=self._test.args.read_bluetooth_uuid_timeout_secs))
      factory.console.info('%s: %d', self._step, battery_level)
    except bluetooth_utils.BluetoothUtilsError as e:
      self.Fail('%s failed to get battery level: %s' % (self._step, e))
      return

    old_battery_level = factory.get_shared_data(self._step)
    if (self._step == READ_BATTERY_STEP_1 and
        (old_battery_level is None or battery_level < old_battery_level)):
      # If the battery level at step 1 becomes higher over different rounds
      # (when the operator keeps retesting it for any reasons),
      # we only keep the lowest one. This is because we want to test if the
      # battery could charge to a higher level at step 2 than step 1.
      factory.set_shared_data(self._step, battery_level)
    elif self._step == READ_BATTERY_STEP_2:
      # We keep the latest battery level read at step 2.
      factory.set_shared_data(self._step, battery_level)

    if self._step == READ_BATTERY_STEP_1:
      data = ('\nSN: %s\nMAC: %s\n' %
              (self._test.args.base_enclosure_serial_number, self._mac))
    else:
      data = ''
    data += '%s: %s\n' % (self._step, battery_level)
    _SaveLogs(self._test.log_file, self._test.aux_log_file, data)

    if self._test.args.battery_log:
      _SaveLocalBatteryLog(self._test.args.base_enclosure_serial_number,
                           self._mac, self._step, battery_level,
                           self._test.args.battery_log)

    self.Pass()


class CheckBatteryLevelTask(factory_task.FactoryTask):
  """This battery level test checks whether the following condistions
  are satisfied:

  1. The battery levels are read twice.
  2. battery_level_1 < battery_level_2
  3. battery_level_1 >= expected_battery_level
  """

  def __init__(self, test):
    super(CheckBatteryLevelTask, self).__init__()
    self._test = test

  def Run(self):
    self._test.template.SetState(_MSG_BATTERY_CHARGE_TEST)

    battery_level_1 = factory.get_shared_data(READ_BATTERY_STEP_1)
    battery_level_2 = factory.get_shared_data(READ_BATTERY_STEP_2)
    factory.console.info('%s: %s', READ_BATTERY_STEP_1, str(battery_level_1))
    factory.console.info('%s: %s', READ_BATTERY_STEP_2, str(battery_level_2))

    if not battery_level_1 or not battery_level_2:
      fail_msg = 'Battery levels should be read twice. read_1: %s, read_2: %s'
    elif (battery_level_1 > battery_level_2 or
          (battery_level_1 == battery_level_2 and battery_level_1 < 100)):
      fail_msg = 'Base battery is not charged up. read_1: %s, read_2: %s'
    elif battery_level_1 < self._test.args.expected_battery_level:
      # Note: battery_level_1 instead of battery_level_2 should be larger than
      #       the expected_battery_level since battery_level_2 is read while
      #       charging and its value is usually larger than its actual value.
      fail_msg = 'Measured battery level %s is less than the expected level %s.'
    else:
      self.Pass()
      return
    self.Fail(fail_msg % (battery_level_1, battery_level_2))


class ChargeTestTask(factory_task.FactoryTask):

  def __init__(self, test, mac, step):
    super(ChargeTestTask, self).__init__()
    self._test = test
    self._mac = mac
    self._step = step

  def ReadBatteryLevel(self, step):
    _ResetAdapter()
    if self._test.args.use_charge_fixture:
      _ExecuteFixtureMethod(self._test.fixture, 'ENABLE_MAGNET')
    factory.console.info('Begin reading battery level...')
    value = bluetooth_utils.GattTool.GetDeviceInfo(
        self._mac, 'battery level', hci_device=self._test.hci_device,
        timeout=self._test.args.read_bluetooth_uuid_timeout_secs)
    if self._test.args.use_charge_fixture:
      _ExecuteFixtureMethod(self._test.fixture, 'DISABLE_MAGNET')
    factory.console.info('%s: %s', step, value)
    return int(value)

  def Run(self):
    if self._step == READ_BATTERY_STEP_1:
      self._test.template.SetState(_MSG_READ_BATTERY_1)
      battery_level = self.ReadBatteryLevel(self._step)
      factory.set_shared_data(BATTERY_LEVEL_KEY, battery_level)
      self.Pass()

    elif self._step == READ_BATTERY_STEP_2:
      def _ReadAndCheckBatteryLevel():
        battery_level2 = self.ReadBatteryLevel(self._step)
        result = battery_level2 > battery_level1
        if result:
          factory.set_shared_data(BATTERY_LEVEL_KEY, battery_level2)
        return result

      self._test.template.SetState(_MSG_READ_BATTERY_2)
      battery_level1 = factory.get_shared_data(BATTERY_LEVEL_KEY)

      # Check if the battery is charging for up to READ_BATTERY_MAX_RETRY_TIMES.
      # Note: the magnet needs to be taken away and re-applied each time.
      #       This operation could be performed automatically with a charging
      #       test fixture; otherwise, it must be performed manually. Also
      #       note that there is a 5-second delay at reading the battery level.
      count = 0
      success_increased_level = False
      while (not success_increased_level and
             count < READ_BATTERY_MAX_RETRY_TIMES):
        success_increased_level = _ReadAndCheckBatteryLevel()
        count += 1

      if success_increased_level:
        self.Pass()
      else:
        self.Fail('ChargeTestTask: the battery is not charging!')


class CheckFirmwareRevisionTestTask(factory_task.FactoryTask):
  """A factory task class to read firmware revision string."""

  def __init__(self, test, mac):
    super(CheckFirmwareRevisionTestTask, self).__init__()
    self._test = test
    self._mac = mac

  def Run(self):
    self._test.template.SetState(_MSG_READ_FIRMWARE_REVISION_STRING)

    factory.console.info('Begin reading firmware revision string via %s...',
                         self._test.hci_device)
    try:
      fw = RetryWithProgress(
          self._test.template, _MSG_READ_FIRMWARE_REVISION_STRING,
          'reading firmware', INPUT_MAX_RETRY_TIMES, INPUT_RETRY_INTERVAL,
          bluetooth_utils.GattTool.GetDeviceInfo, self._mac,
          'firmware revision string', hci_device=self._test.hci_device,
          timeout=self._test.args.read_bluetooth_uuid_timeout_secs)
    except bluetooth_utils.BluetoothUtilsError as e:
      self.Fail('Failed to get firmware revision string: %s' % e)
      return

    factory.console.info('Expected firmware: %s',
                         self._test.args.firmware_revision_string)
    factory.console.info('Actual firmware: %s', fw)
    factory.set_shared_data(self._test.args.firmware_revision_string_key, fw)

    data = 'FW: %s\n' % fw
    _SaveLogs(self._test.log_file, self._test.aux_log_file, data)

    if fw == self._test.args.firmware_revision_string:
      self.Pass()
    else:
      self.Fail('Expected firmware: %s, actual firmware: %s' %
                (self._test.args.firmware_revision_string, fw))


class InputTestTask(factory_task.FactoryTask):
  """The task to test bluetooth input device functionality.

  The task will try to pair with the device given by the test,
  and make the connection.
  After the connection, the number of input event should plus one.
  If it does not plus one, the task fails.
  After connection, operator can try to use the input device and press Enter
  to pass checking or Esc to fail the task.
  In the end of test, the task will try to disconnect the device and remove the
  device. If these procedures fail, the task fails.

  Args:
    finish_after_pair: Whether to end the test after pairing. If false,
                       the operator is prompted to test input, and then
                       the device is unpaired

  """

  def __init__(self, test, finish_after_pair):
    super(InputTestTask, self).__init__()
    self._test = test
    self._target_mac = None
    self._bt_manager = None
    self._adapter = None
    self._need_to_cleanup = True
    self._finish_after_pair = finish_after_pair

  def Cleanup(self):
    """Cleans up input device if it was not cleaned"""
    if self._need_to_cleanup:
      success_to_remove = self.RemoveInput()
      if not success_to_remove:
        logging.error('Fail to remove input in Cleanup')

  def RemoveInput(self):
    """Disconnects the input device and removes it.

    Returns:
      If disconnection and removal are both succeeded, return True, return False
      otherwise.
    """
    return_value = True
    try:
      self._bt_manager.SetDeviceConnected(self._adapter, self._target_mac,
                                          False)
      logging.info('Turned off the connection')
    except self._bt_manager.Error:
      logging.exception('Fail to turn off the connection.')
      return_value = False
    try:
      self._bt_manager.RemovePairedDevice(self._adapter, self._target_mac)
      logging.info('Remove the device')
    except self._bt_manager.Error:
      logging.exception('Fail to remove the device.')
      return_value = False
    return return_value

  def RemoveInputAndQuit(self, success):
    """Removes the input device and quits the task.

    Args:
      success: The task is passed by operator or not.

    Returns:
      If the task is passed by operator and input has been removed successfully,
      pass the task, fail the task otherwise.
    """
    success_to_remove = self.RemoveInput()
    # No need to cleanup again after the task does Pass() or Fail() if removal
    # succeeds here.
    self._need_to_cleanup = not success_to_remove
    if success:
      if success_to_remove:
        self.Pass()
      else:
        self.Fail('InputTestTask: Fail to remove input')
    else:
      self.Fail('Failed by operator')

  def OperatorTestInput(self):
    """Lets operator test the input and press key to pass/fail the task."""
    logging.info('InputTestTask: Test the input by operator now')
    self._test.template.SetState(_MSG_TEST_INPUT)
    self._test.ui.BindKey(test_ui.ENTER_KEY,
                          lambda _: self.RemoveInputAndQuit(True))
    self._test.ui.BindKey(test_ui.ESCAPE_KEY,
                          lambda _: self.RemoveInputAndQuit(False))

  def Run(self):
    def SaveLogAndFail(fail_reason):
      """Save the fail log and invoke Fail()."""
      data = 'Pairing fail: %s\n' % fail_reason
      _SaveLogs(self._test.log_file, self._test.aux_log_file, data)
      self.Fail(fail_reason)

    input_count_before_connection = CheckInputCount()
    self._bt_manager = self._test.dut.bluetooth
    self._adapter = self._bt_manager.GetFirstAdapter(self._test.host_mac)
    self._target_mac = self._test.GetInputDeviceMac()
    if not self._target_mac:
      SaveLogAndFail('InputTestTask: No MAC with which to pair')
    logging.info('Attempting pair with %s', self._target_mac)

    self._bt_manager.DisconnectAndUnpairDevice(self._adapter, self._target_mac)

    success_create_device = RetryWithProgress(
        self._test.template, _MSG_PAIR_INPUT_DEVICE, 'create paired device',
        INPUT_MAX_RETRY_TIMES, INPUT_RETRY_INTERVAL,
        self._bt_manager.CreatePairedDevice, self._adapter,
        self._target_mac, self._DisplayPasskey, self._AuthenticationCancelled)
    if not success_create_device:
      SaveLogAndFail('InputTestTask: Fail to create paired device.')
      return

    success_connect_device = RetryWithProgress(
        self._test.template, _MSG_CONNECT_INPUT_DEVICE, 'connect input device',
        INPUT_MAX_RETRY_TIMES, INPUT_RETRY_INTERVAL,
        self._bt_manager.SetDeviceConnected, self._adapter,
        self._target_mac, True)
    if not success_connect_device:
      SaveLogAndFail('InputTestTask: Fail to connect device.')
      return

    if not WaitForInputCount(self, input_count_before_connection + 1):
      return

    if self._finish_after_pair:
      # We leave the device paired
      self._need_to_cleanup = False
      data = 'Pairing finished\n'
      _SaveLogs(self._test.log_file, self._test.aux_log_file, data)
      self.Pass()
      return

    self.OperatorTestInput()

  def _DisplayPasskey(self, passkey):
    logging.info("Displaying passkey %s", passkey)
    label = MakePasskeyLabelPrompt(passkey)
    self._test.template.SetState(label)

  def _AuthenticationCancelled(self):
    self._test.template.SetState(_MSG_AUTH_FAILED)


class BluetoothTest(unittest.TestCase):
  ARGS = [
      Arg('expected_adapter_count', int, 'Number of bluetooth adapters'
          ' on the machine.', default=0),
      Arg('manufacturer_id', int, 'manufacturer id', optional=True),
      Arg('detect_adapters_retry_times', int, 'Maximum retry time to'
          ' detect adapters', default=10),
      Arg('detect_adapters_interval_secs', int, 'Interval in seconds between'
          ' each retry to detect adapters', default=2),
      Arg('read_bluetooth_uuid_timeout_secs', int,
          'Timeout to read bluetooth characteristics via uuid', default=None,
          optional=True),
      Arg('scan_devices', bool, 'Scan bluetooth device.',
          default=False),
      Arg('prompt_scan_message', bool, 'Prompts a message to tell user to'
          ' enable remote devices discovery mode', default=True),
      Arg('keyword', str, 'Only cares remote devices whose "Name" contains'
          ' keyword.', default=None, optional=True),
      Arg('average_rssi_threshold', float, 'Checks the largest average RSSI'
          ' among scanned device is equal to or greater than '
          ' average_rssi_threshold.',
          default=None, optional=True),
      Arg('scan_counts', int, 'Number of scans to calculate average RSSI',
          default=3),
      Arg('scan_timeout_secs', int, 'Timeout to do one scan', default=5),
      Arg('input_device_mac', str, 'The mac address of bluetooth input device',
          default=None, optional=True),
      Arg('input_device_mac_key', str, 'A key for factory shared data '
          'containing the mac address', default=None, optional=True),
      Arg('input_device_rssi_key', str, 'A key for factory shared data '
          'containing the rssi value', default=None, optional=True),
      Arg('firmware_revision_string_key', str,
          'A key of factory shared data containing firmware revision string',
          optional=True),
      Arg('firmware_revision_string', str,
          'the firmware revision string', optional=True),
      Arg('average_rssi_lower_threshold', (float, dict), 'Checks the average'
          ' RSSI of the target mac is equal to or greater than this threshold.',
          default=None, optional=True),
      Arg('average_rssi_upper_threshold', (float, dict), 'Checks the average'
          ' RSSI of the target mac is equal to or less than this threshold.',
          default=None, optional=True),
      Arg('pair_with_match', bool, 'Whether to pair with the strongest match.',
          default=False, optional=True),
      Arg('finish_after_pair', bool, 'Whether the test should end immediately '
          'after pairing completes', default=False),
      Arg('unpair', bool, 'Whether to unpair matching devices instead of pair',
          default=False, optional=True),
      Arg('check_shift_pair_keys', bool,
          'check if shift-p-a-i-r keys are pressed.',
          default=False, optional=True),
      Arg('check_battery_charging', bool,
          'Whether to check if the battery is charging',
          default=False, optional=True),
      Arg('read_battery_level', int, 'read the battery level',
          default=None, optional=True),
      Arg('check_battery_level', bool, 'Whether to check the battery level',
          default=False, optional=True),
      Arg('prompt_into_fixture', bool, 'Prompt the user to place the base into '
          'the test fixture', default=False, optional=True),
      Arg('use_charge_fixture', bool, 'whether a charge fixture is employed',
          default=False, optional=True),
      Arg('reset_fixture', bool, 'whether to reset the fixture',
          default=False, optional=True),
      Arg('start_charging', bool, 'Prompt the user to start charging the base',
          default=False, optional=True),
      Arg('enable_magnet', bool, 'enable the base',
          default=False, optional=True),
      Arg('reset_magnet', bool, 'reset the base',
          default=False, optional=True),
      Arg('stop_charging', bool, 'Prompt the user to stop charging the base',
          default=False, optional=True),
      Arg('base_enclosure_serial_number', unicode,
          'the base enclosure serial number', default=None, optional=True),
      Arg('battery_log', str,
          'the battery log file', default=None, optional=True),
      Arg('expected_battery_level', int,
          'the expected battery level', default=100, optional=True),
      Arg('log_path', str, 'the directory of the log on the local test host',
          optional=True),
      Arg('aux_log_path', str, 'the path of the aux log on shopfloor',
          optional=True),
      Arg('test_host_id_file', str, 'the file storing the id of the test host',
          optional=True),
  ]

  def SetStrongestRssiMac(self, mac_addr):
    self._strongest_rssi_mac = mac_addr

  def GetInputDeviceMac(self):
    """Gets the input device MAC to pair with, or None if None

    This may be specified in the arguments, or computed at scan time.
    """
    if self._input_device_mac:
      return self._input_device_mac
    else:
      return self._strongest_rssi_mac

  def setUp(self):
    self.dut = device_utils.CreateDUTInterface()
    self.ui = test_ui.UI()
    self.ui.AppendCSS('.start-font-size {font-size: 2em;}')
    self.template = ui_templates.TwoSections(self.ui)
    self.template.SetTitle(_TEST_TITLE)
    self._task_list = []
    self._strongest_rssi_mac = None
    self.fixture = None
    if self.args.input_device_mac_key:
      self._input_device_mac = (
          ColonizeMac(factory.get_shared_data(self.args.input_device_mac_key)))
    else:
      self._input_device_mac = self.args.input_device_mac

    self.btmgmt = bluetooth_utils.BtMgmt(self.args.manufacturer_id)
    self.hci_device = self.btmgmt.GetHciDevice()
    self.host_mac = self.btmgmt.GetMac()
    logging.info('manufacturer_id %s: %s %s',
                 self.args.manufacturer_id, self.hci_device, self.host_mac)

    if self.args.base_enclosure_serial_number:
      if (self.args.test_host_id_file and
          os.path.isfile(self.args.test_host_id_file)):
        with open(self.args.test_host_id_file) as f:
          test_host_id = f.read().strip()
      else:
        test_host_id = None

      filename = '.'.join([self.args.base_enclosure_serial_number,
                           str(test_host_id)])
      self.log_file = None
      self.aux_log_file = None
      if self.args.log_path:
        self.log_file = os.path.join(self.args.log_path, filename)
        # Note: aux_log_file is generated from log_file
        #       Not all projects would generate aux_log_file.
        if self.args.aux_log_path:
          self.aux_log_file = os.path.join(self.args.aux_log_path, filename)

  def tearDown(self):
    """Close the charge test fixture."""
    if self.args.use_charge_fixture:
      self.fixture.Close()

  def runTest(self):
    if self.args.use_charge_fixture:
      # Import this module only when a test station needs it.
      # A base SMT test station does not need to use the charge fixture.
      # pylint: disable=E0611
      from cros.factory.test.fixture import base_charge_fixture
      # Note: only reset the fixture in InitializeFixture test.
      #       This will stop charging and disable the magnet initially.
      #       For the following tests, do not reset the fixture so that
      #       the charging could be continued across tests in the test list
      #       defined in the base_host. The purpose is to keep charing the
      #       battery while executing other tests.
      self.fixture = base_charge_fixture.BaseChargeFixture(
          reset=self.args.reset_fixture)

    if self.args.expected_adapter_count:
      self._task_list.append(DetectAdapterTask(
          self, self.args.expected_adapter_count))

    if self.args.scan_devices:
      if self.args.prompt_scan_message:
        self._task_list.append(TurnOnTask(self, _MSG_TURN_ON_DEVICE))
      self._task_list.append(ScanDevicesTask(self))

    if self.args.input_device_rssi_key:
      self._task_list.append(DetectRSSIofTargetMACTask(self))

    if self.args.prompt_into_fixture:
      self._task_list.append(
          TurnOnTask(self, _MSG_INTO_FIXTURE, test_ui.SPACE_KEY))

    if self.args.read_battery_level == 1:
      self._task_list.append(
          ReadBatteryLevelTask(self, self._input_device_mac,
                               READ_BATTERY_STEP_1))

    if self.args.enable_magnet and self.args.use_charge_fixture:
      self._task_list.append(FixtureControlTask(self, 'ENABLE_MAGNET'))

    if self.args.reset_magnet:
      if self.args.use_charge_fixture:
        self._task_list.append(
            FixtureControlTask(self, 'DISABLE_MAGNET', post_sleep=1))
        self._task_list.append(FixtureControlTask(self, 'ENABLE_MAGNET'))
      else:
        self._task_list.append(
            TurnOnTask(self, _MSG_RESET_MAGNET, test_ui.SPACE_KEY))

    if self.args.start_charging:
      if self.args.use_charge_fixture:
        self._task_list.append(
            # Let it charge for a little while.
            FixtureControlTask(self, 'START_CHARGING'))
      else:
        self._task_list.append(
            TurnOnTask(self, _MSG_START_CHARGE, test_ui.SPACE_KEY))

    if self.args.check_shift_pair_keys:
      self._task_list.append(
          CheckDisconnectionOfPairedDeviceTask(self, self._input_device_mac))

    if self.args.unpair:
      self._task_list.append(
          UnpairTask(self, self._input_device_mac, self.args.keyword))

    if self.args.firmware_revision_string:
      self._task_list.append(
          CheckFirmwareRevisionTestTask(self, self._input_device_mac))

    if self.args.pair_with_match:
      self._task_list.append(InputTestTask(self, self.args.finish_after_pair))

    if self.args.read_battery_level == 2:
      self._task_list.append(
          ReadBatteryLevelTask(self, self._input_device_mac,
                               READ_BATTERY_STEP_2))

    if self.args.check_battery_level:
      self._task_list.append(CheckBatteryLevelTask(self))

    if self.args.stop_charging:
      if self.args.use_charge_fixture:
        self._task_list.append(FixtureControlTask(self, 'STOP_CHARGING'))
      else:
        self._task_list.append(
            TurnOnTask(self, _MSG_STOP_CHARGE, test_ui.SPACE_KEY))

    factory_task.FactoryTaskManager(self.ui, self._task_list).Run()