graphyte/graphyte.py

#!/usr/bin/python
# -*- coding: utf-8 -*-
# Copyright 2015 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.

"""The Google RF Test Framework module.

This is the main application module that will drive the other modules.
This module can also be imported and an instance of graphyte created from there.

Typical Usage (Import to other modules):
  from graphyte.graphyte import Graphyte
  app = Graphyte()
Please refer to graphyte_main.py.
"""

import json
import logging
import os

from . import utils
from .default_setting import DEFAULT_CONFIG_FILE
from .default_setting import DEFAULT_LOG_FILE
from .default_setting import DEFAULT_RESULT_FILE
from .default_setting import GRAPHYTE_DIR
from .default_setting import logger
from .default_setting import GRAPHYTE_OUT
from .result_writer import ResultWriter
from .port_config import PortConfig
from .testplan import TestPlan


class Graphyte(object):
  """The graphyte module.

  Instantiating the class does everything now. Refer to the constructor below.
  """
  def __init__(self, config_file=None, log_file=None, result_file=None,
               verbose=False, console_output=False, append_mode=False):
    """The constructor for this module does most of it right now.

    - Load the graphyte.settings file to find the modules to load.
    - Load the plugins
    - Load test plans
    - Execute all the tests in the test plan.

    The configuration file graphyte_config.sample.json follows the json format.
    The file has the dut and inst module name and config file.
    These entries are initialized to sample_dut.py and sample_inst.py. The dut
    pluings are searched under ./dut/ and the instrument plugins are searched
    under ./inst/. Once the plugins are found, they are loaded.

    The testplan module is responsible for reading the CSV file which defines
    the testplan and generate a list of the test case, that can be passed to
    inst and dut plugins.

    The interaction with the dut and the instrument plugin is dipicted in
    the flow diagram.
    """
    config_file = config_file or DEFAULT_CONFIG_FILE
    log_file = log_file or DEFAULT_LOG_FILE
    result_file = result_file or DEFAULT_RESULT_FILE

    self.SetLogging(log_file, verbose, append_mode, console_output)
    logger.info('Starting Graphyte....')

    config_path = utils.SearchConfig(config_file, GRAPHYTE_DIR)
    config_dir = os.path.dirname(config_path)
    with open(config_path, 'r') as f:
      graphyte_config = json.load(f)

    # Load DUT and Inst class
    self.dut = self._LoadDevice(graphyte_config, 'dut', config_dir)
    self.inst = self._LoadDevice(graphyte_config, 'inst', config_dir)

    # Load test plan
    self.testplan = TestPlan(graphyte_config['test_plan'], config_dir)
    if not self.testplan:
      raise ValueError('Load test plan failed.')
    self.rf_type = None

    # Load port config, including port mapping and path loss
    self.port_config = PortConfig(graphyte_config['port_config'], config_dir)

    # Load the test argument
    self.retries = graphyte_config.get('retries', 1)

    # Open a result writer
    self.result_writer = ResultWriter(result_file)

    # Record all tests are pass or not
    self.all_tests_pass = None

  def __del__(self):
    self.TerminateDevices()

  def _LoadDevice(self, graphyte_config, device_type, config_dir):
    """Dynamically load the device instance by the global config.

    Args:
      graphyte_config: the global config.
      device_type: 'dut' or 'inst'
      config_dir: the directory of the graphyte config file.
    Returns:
      The device instance.
    """
    class_name_mapping = {'dut': 'DUT',
                          'inst': 'Inst'}
    module_name = graphyte_config[device_type]
    class_name = class_name_mapping[device_type]
    module = __import__('%s.%s' % (device_type, module_name),
                        globals(), level=1, fromlist=[class_name])
    # Load config file.
    default_config_file = module_name.replace('.', os.path.sep) + '.json'
    default_config_path = os.path.join(GRAPHYTE_DIR, device_type,
                                       default_config_file)
    device_config = utils.LoadConfig(default_config_path)

    config_key = '%s_config_file' % device_type
    if config_key in graphyte_config:
      config_file = utils.SearchConfig(graphyte_config[config_key],
                                       [config_dir])
      device_config = utils.OverrideConfig(device_config,
                                           utils.LoadConfig(config_file))
    config_key = '%s_config' % device_type
    if config_key in graphyte_config:
      device_config = utils.OverrideConfig(device_config,
                                           graphyte_config[config_key])

    logger.info('Loading %s: %s, device config: %s',
                device_type, module_name, device_config)
    return module.__dict__[class_name](**device_config)

  def SetLogging(self, log_file, verbose, append_mode, console_output):
    level = logging.DEBUG if verbose else logging.INFO
    mode = 'a' if append_mode else 'w'
    log_file = utils.PrepareOutputFile(log_file)
    fmt = '[%(levelname)s] %(asctime)s %(filename)s:%(lineno)d %(message)s'
    date_fmt = '%H:%M:%S'
    formatter = logging.Formatter(fmt, date_fmt)

    logger.setLevel(level)
    file_handler = logging.FileHandler(log_file, mode)
    file_handler.setFormatter(formatter)
    logger.addHandler(file_handler)
    if console_output:
      console_handler = logging.StreamHandler(GRAPHYTE_OUT)
      console_handler.setFormatter(formatter)
      logger.addHandler(console_handler)

  def Run(self):
    """The main method of the class.

    It returns whether all tests are passed or not.
    """
    self.all_tests_pass = True
    self.InitialzeDevices()
    for test_case in self.testplan:
      self.RunTest(test_case)
    self.TerminateDevices()
    logger.info('All tests are pass? %s', self.all_tests_pass)
    self.result_writer.WriteTotalResult(self.all_tests_pass)
    return self.all_tests_pass

  def InitialzeDevices(self):
    try:
      self.inst.Initialize()
      self.inst.LockInstrument()
      self.inst.SelfCalibrate()
      self.inst.InitPortConfig(self.port_config)
      self.dut.Initialize()
    except Exception:
      logger.exception('Initialize devices failed')

  def TerminateDevices(self):
    try:
      self.inst.UnlockInstrument()
      self.inst.Terminate()
      self.dut.Terminate()
    except Exception:
      logger.exception('Terminate devices failed')

  def SetRF(self, rf_type):
    logger.info('Set RF type: %s', rf_type)
    if rf_type != self.rf_type:
      self.rf_type = rf_type
      self.dut.SetRF(rf_type)
      self.inst.SetRF(rf_type)

  def RunTest(self, test_case):
    logger.info('Running test: \n%s', test_case)
    self.SetRF(test_case.rf_type)

    self.inst.SetPortConfig(test_case)
    function_map = {
        'TX': self.RunTxTest,
        'RX': self.RunRxTest}

    def _RunTest():
      try:
        return function_map[test_case.test_type](test_case)
      except Exception:
        logger.exception('RunTest error. test_case: %s', test_case)
        return False, {}
    def _Condition(ret):
      # ret is (is_pass, result)
      return ret[0]

    # pylint: disable=W0633
    is_pass, result = utils.Retry(self.retries, 0, _RunTest, _Condition)
    self.all_tests_pass &= is_pass
    self.result_writer.WriteResult(test_case, result)

  def RunTxTest(self, test_case):
    self.dut.TxConfig(test_case)
    self.inst.TxConfig(test_case)
    self.dut.TxStart(test_case)
    self.inst.TxMeasure(test_case)
    self.dut.TxStop(test_case)
    result = self.inst.TxGetResult(test_case)
    return result

  def RunRxTest(self, test_case):
    self.dut.RxConfig(test_case)
    self.inst.RxConfig(test_case)
    self.dut.RxClearResult(test_case)
    self.inst.RxGenerate(test_case)
    result = self.dut.RxGetResult(test_case)
    self.inst.RxStop(test_case)
    return result