Home | History | Annotate | Download | only in video 
      1 # Copyright 2017 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 importlib
      6 import logging
      7 import os
      8 import re
      9 
     10 import yaml
     11 
     12 from autotest_lib.client.common_lib import error
     13 
     14 class DeviceCapability(object):
     15     """
     16     Generate capabilities status on DUT from yaml files in a given path.
     17     Answer from the capabilities whether some capability is satisfied on DUT.
     18     """
     19 
     20     def __init__(self, settings_path='/usr/local/etc/autotest-capability'):
     21         """
     22         @param settings_path: string, the base directory for autotest
     23                               capability. There should be yaml files.
     24         """
     25         self.capabilities = self.__get_autotest_capability(settings_path)
     26         logging.info("Capabilities:\n%r", self.capabilities)
     27 
     28 
     29     def __get_autotest_capability(self, settings_path):
     30         """
     31         Generate and summarize capabilities from yaml files in
     32         settings_path with detectors.
     33 
     34         @param settings_path: string, the base directory for autotest
     35                               capability. There should be yaml files.
     36         @returns dict:
     37             The capabilities on DUT.
     38             Its key is string denoting a capability. Its value is 'yes', 'no' or
     39             'disable.'
     40         """
     41 
     42         def run_detector(name):
     43             """
     44             Run a detector in the detector directory. (i.e.
     45             autotest/files/client/cros/video/detectors)
     46             Return the result of the detector.
     47 
     48             @param name: string, the name of running detector.
     49             @returns string, a result of detect() in the detector script.
     50             """
     51             if name not in detect_results:
     52                 detector = importlib.import_module(
     53                     "autotest_lib.client.cros.video.detectors.%s"
     54                     % name)
     55                 detect_results[name] = detector.detect()
     56                 logging.info("Detector result (%s): %s",
     57                              name, detect_results[name])
     58             return detect_results[name]
     59 
     60         managed_cap_fpath = os.path.join(settings_path,
     61                                          'managed-capabilities.yaml')
     62         if not os.path.exists(managed_cap_fpath):
     63             raise error.TestFail("%s is not installed" % managed_cap_fpath)
     64         managed_caps = yaml.load(file(managed_cap_fpath))
     65 
     66         cap_files = [f for f in os.listdir(settings_path)
     67                      if re.match(r'^[0-9]+-.*\.yaml$', f)]
     68         cap_files.sort(key=lambda f: int(f.split('-', 1)[0]))
     69 
     70         detect_results = {}
     71         autotest_caps = dict.fromkeys(managed_caps, 'no')
     72         for fname in cap_files:
     73             logging.debug('Processing caps: %s', fname)
     74             fname = os.path.join(settings_path, fname)
     75             for rule in yaml.load(file(fname)):
     76                 # The type of rule is string or dict
     77                 # If the type is a string, it is a capability (e.g. webcam).
     78                 # If a specific condition (e.g. kepler, cpu type) is required,
     79                 # rule would be dict, for example,
     80                 # {'detector': 'intel_cpu',
     81                 #  'match': ['intel_celeron_1007U'],
     82                 #  'capabilities': ['no hw_h264_enc_1080_30'] }.
     83                 logging.debug("%r", rule)
     84                 caps = []
     85                 if isinstance(rule, dict):
     86                     if run_detector(rule['detector']) in rule['match']:
     87                         caps = rule['capabilities']
     88                 else:
     89                     caps = [rule]
     90 
     91                 for capability in caps:
     92                     m = re.match(r'(?:(disable|no)\s+)?([\w\-]+)$', capability)
     93                     prefix, capability = m.groups()
     94                     if capability in managed_caps:
     95                         autotest_caps[capability] = prefix or 'yes'
     96                     else:
     97                         raise error.TestFail(
     98                             "Unexpected capability: %s" % capability)
     99 
    100         return autotest_caps
    101 
    102 
    103     def get_managed_caps(self):
    104         return self.capabilities.keys()
    105 
    106 
    107     def get_capability_results(self):
    108         return self.capabilities
    109 
    110 
    111     def get_capability(self, cap):
    112         """
    113         Decide if a device satisfies a required capability for an autotest.
    114 
    115         @param cap: string, denoting one capability. It must be one in
    116                     settings_path + 'managed-capabilities.yaml.'
    117         @returns 'yes', 'no', or 'disable.'
    118         """
    119         try:
    120             return self.capabilities[cap]
    121         except KeyError:
    122             raise error.TestFail("Unexpected capability: %s" % cap)
    123 
    124 
    125     def ensure_capability(self, cap):
    126         """
    127         Raise TestNAError if a device doesn't satisfy cap.
    128         """
    129         if self.get_capability(cap) != 'yes':
    130             raise error.TestNAError("Missing Capability: %s" % cap)
    131 
    132 
    133     def have_capability(self, cap):
    134         """
    135         Return whether cap is available.
    136         """
    137         return self.get_capability(cap) == 'yes'
    138