graphyte/device.py

# Copyright 2018 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 graphyte_common  # pylint: disable=unused-import
from graphyte import link
from graphyte.default_setting import logger


class DeviceType(object):
  """Used to select device in action."""
  DUT = 'DUT'
  INST = 'INST'
  ALL = 'ALL'

  @classmethod
  def IsDUT(cls, device):
    return device in [cls.DUT, cls.ALL]

  @classmethod
  def IsInst(cls, device):
    return device in [cls.INST, cls.ALL]


class DeviceBase(object):
  """The abstract device class.

  The class hierarchy of the device is:
    DeviceBase
    - dut.DUTBase
    - inst.InstBase
  """

  name = 'Plugin Name'
  need_link = False

  def __init__(self, **kwargs):
    """Abstract init method. The init method of each child class should be:

    def __init__(self, arg1, **kwargs):
      # Call parent's __init__
      super(ChildClassName, self).__init__(**kwargs)
      # Set the child arguments
      self.arg1 = arg1
      # Set the RF controllers
      self.controllers = {
          'WLAN': ChildWlanController(self),
          'BLUETOOTH': ChildBluetoothController(self),
      }
    """
    self.rf_type = None
    self.controllers = {}
    self.link = link.Create(kwargs['link_options']) if self.need_link else None

  def __str__(self):
    return "Name: %s\nRF: %s\nLink: %s" % (
        self.name, self.rf_type, self.link)

  @property
  def active_controller(self):
    if self.rf_type not in self.controllers:
      raise KeyError('rf_type "%s" is not supported.' % self.rf_type)
    return self.controllers[self.rf_type]

  @active_controller.setter
  def active_controller(self, value):
    del value
    raise ValueError(
        'Update `rf_type` value instead of updating `active_controller`.')

  def Initialize(self):
    """Initializes the device."""
    logger.debug('device.Initialize')
    if self.need_link:
      self.link.Open()
      self.link.CheckReady()
    return_value = self._Initialize()
    logger.info('Device version is %r.', self.GetVersion())
    return return_value

  def _Initialize(self):
    pass

  def Terminate(self):
    """Terminates the device."""
    logger.debug('device.Terminate')
    if self.rf_type is not None:
      self.controllers[self.rf_type].Terminate()
      self.rf_type = None
    self._Terminate()
    if self.link is not None:
      self.link.Close()

  def _Terminate(self):
    pass

  def SetRF(self, rf_type):
    """Set the RF type."""
    if self.rf_type == rf_type:
      return
    if self.rf_type is not None:
      self.active_controller.Terminate()
    self.rf_type = rf_type
    if self.rf_type is not None:
      self.active_controller.Initialize()

  def OnFailure(self):
    """Called when a test is failed."""
    self.active_controller.OnFailure()

  def GetVersion(self):
    """Returns device version."""
    raise NotImplementedError('This function returns an object that can be '
                              'formatted by %r which represents the version '
                              'of the device. For example, returns the output '
                              'from "uname -a".')

class ControllerBase(object):
  """The abstract RF controller class.

  The class hierarchy of the RF controller is:
    device.ControllerBase
    - dut.ControllerBase
      - dut.WlanControllerBase
      - dut.BluetoothControllerBase
      - dut.ZigbeeControllerBase
    - inst.ControllerBase
      - inst.PowerMeterControllerBase
      - inst.WlanControllerBase
      - inst.BluetoothControllerBase
      - inst.ZigbeeControllerBase
  """

  # The RF type of the controller. Child class should override this.
  RF_TYPE = ''

  def Initialize(self):
    return self._Initialize()

  def _Initialize(self):
    """Initializes the controller."""
    pass

  def Terminate(self):
    return self._Terminate()

  def _Terminate(self):
    """Disconnects/unconfigures the interface that was being used."""
    pass

  def OnFailure(self):
    return self._OnFailure()

  def _OnFailure(self):
    """Called when a test is failed."""
    pass

  def _CheckTestArgument(self, test_case, required_test_type):
    """Checks the test case has correct test type and RF type.

    Args:
      test_case: a dict containing the test arguments.
      required_test_type: the required test type. 'TX' or 'RX'.

    Returns:
      the test case.
    """
    assert test_case['test_type'] == required_test_type
    assert self.RF_TYPE == 'ALL' or test_case['rf_type'] == self.RF_TYPE
    return test_case