Source code for wlauto.core.device

#    Copyright 2013-2015 ARM Limited
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# See the License for the specific language governing permissions and
# limitations under the License.

Base classes for device interfaces.

    :Device: The base class for all devices. This defines the interface that must be
             implemented by all devices and therefore any workload and instrumentation
             can always rely on.
    :AndroidDevice: Implements most of the :class:`Device` interface, and extends it
                    with a number of Android-specific methods.
    :BigLittleDevice: Subclasses :class:`AndroidDevice` to implement big.LITTLE-specific
                      runtime parameters.
    :SimpleMulticoreDevice: Subclasses :class:`AndroidDevice` to implement homogeneous cores
                          device runtime parameters.


import os
import imp
import string
from collections import OrderedDict
from contextlib import contextmanager

from wlauto.core.extension import Extension, ExtensionMeta, AttributeCollection, Parameter
from wlauto.core.extension_loader import ExtensionLoader
from wlauto.exceptions import DeviceError, ConfigError
from wlauto.utils.types import list_of_integers, list_of, caseless_string

__all__ = ['RuntimeParameter', 'CoreParameter', 'Device', 'DeviceMeta']

[docs]class RuntimeParameter(object): """ A runtime parameter which has its getter and setter methods associated it with it. """ def __init__(self, name, getter, setter, getter_args=None, setter_args=None, value_name='value', override=False): """ :param name: the name of the parameter. :param getter: the getter method which returns the value of this parameter. :param setter: the setter method which sets the value of this parameter. The setter always expects to be passed one argument when it is called. :param getter_args: keyword arguments to be used when invoking the getter. :param setter_args: keyword arguments to be used when invoking the setter. :param override: A ``bool`` that specifies whether a parameter of the same name further up the hierarchy should be overridden. If this is ``False`` (the default), an exception will be raised by the ``AttributeCollection`` instead. """ = name self.getter = getter self.setter = setter self.getter_args = getter_args or {} self.setter_args = setter_args or {} self.value_name = value_name self.override = override def __str__(self): return __repr__ = __str__
[docs]class CoreParameter(RuntimeParameter): """A runtime parameter that will get expanded into a RuntimeParameter for each core type."""
[docs] def get_runtime_parameters(self, core_names): params = [] for core in set(core_names): name = string.Template( getter = string.Template(self.getter).substitute(core=core) setter = string.Template(self.setter).substitute(core=core) getargs = dict(self.getter_args.items() + [('core', core)]) setargs = dict(self.setter_args.items() + [('core', core)]) params.append(RuntimeParameter(name, getter, setter, getargs, setargs, self.value_name, self.override)) return params
class DynamicModuleSpec(dict): @property def name(self): return self.keys()[0] def __init__(self, *args, **kwargs): dict.__init__(self) if args: if len(args) > 1: raise ValueError(args) value = args[0] else: value = kwargs if isinstance(value, basestring): self[value] = {} elif isinstance(value, dict) and len(value) == 1: for k, v in value.iteritems(): self[k] = v else: raise ValueError(value)
[docs]class DeviceMeta(ExtensionMeta): to_propagate = ExtensionMeta.to_propagate + [ ('runtime_parameters', RuntimeParameter, AttributeCollection), ('dynamic_modules', DynamicModuleSpec, AttributeCollection), ]
[docs]class Device(Extension): """ Base class for all devices supported by Workload Automation. Defines the interface the rest of WA uses to interact with devices. :name: Unique name used to identify the device. :platform: The name of the device's platform (e.g. ``Android``) this may be used by workloads and instrumentation to assess whether they can run on the device. :working_directory: a string of the directory which is going to be used by the workloads on the device. :binaries_directory: a string of the binary directory for the device. :has_gpu: Should be ``True`` if the device as a separate GPU, and ``False`` if graphics processing is done on a CPU. .. note:: Pretty much all devices currently on the market have GPUs, however this may not be the case for some development boards. :path_module: The name of one of the modules implementing the os.path interface, e.g. ``posixpath`` or ``ntpath``. You can provide your own implementation rather than relying on one of the standard library modules, in which case you need to specify the *full* path to you module. e.g. '/home/joebloggs/' :parameters: A list of RuntimeParameter objects. The order of the objects is very important as the setters and getters will be called in the order the RuntimeParameter objects inserted. :active_cores: This should be a list of all the currently active cpus in the device in ``'/sys/devices/system/cpu/online'``. The returned list should be read from the device at the time of read request. """ __metaclass__ = DeviceMeta parameters = [ Parameter('core_names', kind=list_of(caseless_string), mandatory=True, default=None, description=""" This is a list of all cpu cores on the device with each element being the core type, e.g. ``['a7', 'a7', 'a15']``. The order of the cores must match the order they are listed in ``'/sys/devices/system/cpu'``. So in this case, ``'cpu0'`` must be an A7 core, and ``'cpu2'`` an A15.' """), Parameter('core_clusters', kind=list_of_integers, mandatory=True, default=None, description=""" This is a list indicating the cluster affinity of the CPU cores, each element correponding to the cluster ID of the core coresponding to its index. E.g. ``[0, 0, 1]`` indicates that cpu0 and cpu1 are on cluster 0, while cpu2 is on cluster 1. If this is not specified, this will be inferred from ``core_names`` if possible (assuming all cores with the same name are on the same cluster). """), ] runtime_parameters = [] # dynamic modules are loaded or not based on whether the device supports # them (established at runtime by module probling the device). dynamic_modules = [] # These must be overwritten by subclasses. name = None platform = None default_working_directory = None has_gpu = None path_module = None active_cores = None def __init__(self, **kwargs): # pylint: disable=W0613 super(Device, self).__init__(**kwargs) if not self.path_module: raise NotImplementedError('path_module must be specified by the deriving classes.') libpath = os.path.dirname(os.__file__) modpath = os.path.join(libpath, self.path_module) if not modpath.lower().endswith('.py'): modpath += '.py' try: self.path = imp.load_source('device_path', modpath) except IOError: raise DeviceError('Unsupported path module: {}'.format(self.path_module)) def validate(self): # pylint: disable=access-member-before-definition,attribute-defined-outside-init if self.core_names and not self.core_clusters: self.core_clusters = [] clusters = [] for cn in self.core_names: if cn not in clusters: clusters.append(cn) self.core_clusters.append(clusters.index(cn)) if len(self.core_names) != len(self.core_clusters): raise ConfigError('core_names and core_clusters are of different lengths.') def initialize(self, context): """ Initialization that is performed at the begining of the run (after the device has been connecte). """ loader = ExtensionLoader() for module_spec in self.dynamic_modules: module = self._load_module(loader, module_spec) if not hasattr(module, 'probe'): message = 'Module {} does not have "probe" attribute; cannot be loaded dynamically' raise ValueError(message.format( if module.probe(self): self.logger.debug('Installing module "{}"'.format( self._install_module(module) else: self.logger.debug('Module "{}" is not supported by the device'.format(
[docs] def reset(self): """ Initiate rebooting of the device. Added in version 2.1.3. """ raise NotImplementedError()
[docs] def boot(self, *args, **kwargs): """ Perform the seteps necessary to boot the device to the point where it is ready to accept other commands. Changed in version 2.1.3: no longer expected to wait until boot completes. """ raise NotImplementedError()
[docs] def connect(self, *args, **kwargs): """ Establish a connection to the device that will be used for subsequent commands. Added in version 2.1.3. """ raise NotImplementedError()
[docs] def disconnect(self): """ Close the established connection to the device. """ raise NotImplementedError()
[docs] def ping(self): """ This must return successfully if the device is able to receive commands, or must raise :class:`wlauto.exceptions.DeviceUnresponsiveError` if the device cannot respond. """ raise NotImplementedError()
[docs] def get_runtime_parameter_names(self): return [ for p in self._expand_runtime_parameters()]
[docs] def get_runtime_parameters(self): """ returns the runtime parameters that have been set. """ # pylint: disable=cell-var-from-loop runtime_parameters = OrderedDict() for rtp in self._expand_runtime_parameters(): if not rtp.getter: continue getter = getattr(self, rtp.getter) rtp_value = getter(**rtp.getter_args) runtime_parameters[] = rtp_value return runtime_parameters
[docs] def set_runtime_parameters(self, params): """ The parameters are taken from the keyword arguments and are specific to a particular device. See the device documentation. """ runtime_parameters = self._expand_runtime_parameters() rtp_map = { rtp for rtp in runtime_parameters} params = OrderedDict((k.lower(), v) for k, v in params.iteritems() if v is not None) expected_keys = rtp_map.keys() if not set(params.keys()).issubset(set(expected_keys)): unknown_params = list(set(params.keys()).difference(set(expected_keys))) raise ConfigError('Unknown runtime parameter(s): {}'.format(unknown_params)) for param in params: self.logger.debug('Setting runtime parameter "{}"'.format(param)) rtp = rtp_map[param] setter = getattr(self, rtp.setter) args = dict(rtp.setter_args.items() + [(rtp.value_name, params[])]) setter(**args)
[docs] def capture_screen(self, filepath): """Captures the current device screen into the specified file in a PNG format.""" raise NotImplementedError()
[docs] def get_properties(self, output_path): """Captures and saves the device configuration properties version and any other relevant information. Return them in a dict""" raise NotImplementedError()
[docs] def listdir(self, path, **kwargs): """ List the contents of the specified directory. """ raise NotImplementedError()
[docs] def push_file(self, source, dest): """ Push a file from the host file system onto the device. """ raise NotImplementedError()
[docs] def pull_file(self, source, dest): """ Pull a file from device system onto the host file system. """ raise NotImplementedError()
[docs] def delete_file(self, filepath): """ Delete the specified file on the device. """ raise NotImplementedError()
[docs] def file_exists(self, filepath): """ Check if the specified file or directory exist on the device. """ raise NotImplementedError()
[docs] def get_pids_of(self, process_name): """ Returns a list of PIDs of the specified process name. """ raise NotImplementedError()
[docs] def kill(self, pid, as_root=False): """ Kill the process with the specified PID. """ raise NotImplementedError()
[docs] def killall(self, process_name, as_root=False): """ Kill all running processes with the specified name. """ raise NotImplementedError()
[docs] def install(self, filepath, **kwargs): """ Install the specified file on the device. What "install" means is device-specific and may possibly also depend on the type of file.""" raise NotImplementedError()
[docs] def uninstall(self, filepath): """ Uninstall the specified file on the device. What "uninstall" means is device-specific and may possibly also depend on the type of file.""" raise NotImplementedError()
[docs] def execute(self, command, timeout=None, **kwargs): """ Execute the specified command command on the device and return the output. :param command: Command to be executed on the device. :param timeout: If the command does not return after the specified time, execute() will abort with an error. If there is no timeout for the command, this should be set to 0 or None. Other device-specific keyword arguments may also be specified. :returns: The stdout output from the command. """ raise NotImplementedError()
[docs] def sleep(self, seconds): """Sleep for the specified time on the target device. :param seconds: Time in seconds to sleep on the device The sleep is executed on the device using self.execute(). We set the timeout for this command to be 10 seconds longer than the sleep itself to make sure the command has time to complete before we timeout. """ self.execute("sleep {}".format(seconds), timeout=seconds + 10)
[docs] def set_sysfile_value(self, filepath, value, verify=True, binary=False): """ Write the specified value to the specified file on the device and verify that the value has actually been written. :param file: The file to be modified. :param value: The value to be written to the file. Must be an int or a string convertable to an int. :param verify: Specifies whether the value should be verified, once written. :param binary: Specifies whether the value should be written as binary data. Should raise DeviceError if could write value. """ raise NotImplementedError()
[docs] def get_sysfile_value(self, sysfile, kind=None, binary=False): """ Get the contents of the specified sysfile. :param sysfile: The file who's contents will be returned. :param kind: The type of value to be expected in the sysfile. This can be any Python callable that takes a single str argument. If not specified or is None, the contents will be returned as a string. :param binary: Whether the value should be encoded into base64 for reading to deal with binary format. """ raise NotImplementedError()
[docs] def start(self): """ This gets invoked before an iteration is started and is endented to help the device manange any internal supporting functions. """ pass
[docs] def stop(self): """ This gets invoked after iteration execution has completed and is endented to help the device manange any internal supporting functions. """ pass
[docs] def is_network_connected(self): """ Checks if the device is connected to the internet """ raise NotImplementedError()
def __str__(self): return 'Device<{}>'.format( __repr__ = __str__ def _expand_runtime_parameters(self): expanded_params = [] for param in self.runtime_parameters: if isinstance(param, CoreParameter): expanded_params.extend(param.get_runtime_parameters(self.core_names)) # pylint: disable=no-member else: expanded_params.append(param) return expanded_params @contextmanager def _check_alive(self): try: yield except Exception as e: raise e