sascha/devlib
1
0
Fork 0
mirror of https://github.com/ARM-software/devlib.git synced 2024-10-06 10:50:51 +01:00
Code Issues Releases Wiki Activity

pylint fixes

Browse Source
This commit is contained in:
Marc Bonnici 2018-07-11 17:30:45 +01:00 committed by setrofim
parent 5cb551b315
commit 454b94501c
44 changed files with 330 additions and 275 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
devlib/derived/__init__.py
View File

@ -36,25 +36,28 @@ class DerivedMetric(object):
msg = 'Unknown measurement type: {}'
raise ValueError(msg.format(measurement_type))
def __cmp__(self, other):
if hasattr(other, 'value'):
return cmp(self.value, other.value)
else:
return cmp(self.value, other)
def __str__(self):
if self.units:
return '{}: {} {}'.format(self.name, self.value, self.units)
else:
return '{}: {}'.format(self.name, self.value)
# pylint: disable=undefined-variable
def __cmp__(self, other):
if hasattr(other, 'value'):
return cmp(self.value, other.value)
else:
return cmp(self.value, other)
__repr__ = __str__
class DerivedMeasurements(object):
# pylint: disable=no-self-use,unused-argument
def process(self, measurements_csv):
return []
# pylint: disable=no-self-use
def process_raw(self, *args):
return []

5
devlib/derived/energy.py
View File

@ -15,12 +15,13 @@
from __future__ import division
from collections import defaultdict
from devlib import DerivedMeasurements, DerivedMetric
from devlib.instrument import MEASUREMENT_TYPES, InstrumentChannel
from devlib.derived import DerivedMeasurements, DerivedMetric
from devlib.instrument import MEASUREMENT_TYPES
class DerivedEnergyMeasurements(DerivedMeasurements):
# pylint: disable=too-many-locals,too-many-branches
@staticmethod
def process(measurements_csv):

10
devlib/derived/fps.py
View File

@ -15,7 +15,6 @@
from __future__ import division
import os
import re
try:
import pandas as pd
@ -24,8 +23,9 @@ except ImportError:
from past.builtins import basestring
from devlib import DerivedMeasurements, DerivedMetric, MeasurementsCsv, InstrumentChannel
from devlib.derived import DerivedMeasurements, DerivedMetric
from devlib.exception import HostError
from devlib.instrument import MeasurementsCsv
from devlib.utils.csvutil import csvwriter
from devlib.utils.rendering import gfxinfo_get_last_dump, VSYNC_INTERVAL
from devlib.utils.types import numeric
@ -45,6 +45,7 @@ class DerivedFpsStats(DerivedMeasurements):
if filename is not None and os.sep in filename:
raise ValueError('filename cannot be a path (cannot countain "{}"'.format(os.sep))
# pylint: disable=no-member
def process(self, measurements_csv):
if isinstance(measurements_csv, basestring):
measurements_csv = MeasurementsCsv(measurements_csv)
@ -65,6 +66,7 @@ class DerivedFpsStats(DerivedMeasurements):
class DerivedGfxInfoStats(DerivedFpsStats):
#pylint: disable=arguments-differ
@staticmethod
def process_raw(filepath, *args):
metrics = []
@ -155,6 +157,7 @@ class DerivedGfxInfoStats(DerivedFpsStats):
class DerivedSurfaceFlingerStats(DerivedFpsStats):
# pylint: disable=too-many-locals
def _process_with_pandas(self, measurements_csv):
data = pd.read_csv(measurements_csv.path)
@ -193,7 +196,7 @@ class DerivedSurfaceFlingerStats(DerivedFpsStats):
janks = 0
not_at_vsync = 0
janks_pc = 0 if frame_count == 0 else janks * 100 / frame_count
janks_pc = 0 if frame_count == 0 else janks * 100 / frame_count
return [DerivedMetric('fps', fps, 'fps'),
DerivedMetric('total_frames', frame_count, 'frames'),
@ -202,6 +205,7 @@ class DerivedSurfaceFlingerStats(DerivedFpsStats):
DerivedMetric('janks_pc', janks_pc, 'percent'),
DerivedMetric('missed_vsync', not_at_vsync, 'count')]
# pylint: disable=unused-argument,no-self-use
def _process_without_pandas(self, measurements_csv):
# Given that SurfaceFlinger has been deprecated in favor of GfxInfo,
# it does not seem worth it implementing this.

3
devlib/exception.py
View File

@ -37,6 +37,7 @@ class HostError(DevlibError):
pass
# pylint: disable=redefined-builtin
class TimeoutError(DevlibError):
"""Raised when a subprocess command times out. This is basically a ``DevlibError``-derived version
of ``subprocess.CalledProcessError``, the thinking being that while a timeout could be due to
@ -79,7 +80,7 @@ def get_traceback(exc=None):
object, or for the current exception exc is not specified.
"""
import io, traceback, sys
import io, traceback, sys # pylint: disable=multiple-imports
if exc is None:
exc = sys.exc_info()
if not exc:

3
devlib/host.py
View File

@ -25,6 +25,7 @@ from devlib.utils.misc import check_output
PACKAGE_BIN_DIRECTORY = os.path.join(os.path.dirname(__file__), 'bin')
# pylint: disable=redefined-outer-name
def kill_children(pid, signal=signal.SIGKILL):
with open('/proc/{0}/task/{0}/children'.format(pid), 'r') as fd:
for cpid in map(int, fd.read().strip().split()):
@ -35,6 +36,7 @@ class LocalConnection(object):
name = 'local'
# pylint: disable=unused-argument
def __init__(self, platform=None, keep_password=True, unrooted=False,
password=None, timeout=None):
self.logger = logging.getLogger('local_connection')
@ -55,6 +57,7 @@ class LocalConnection(object):
else:
shutil.copy(source, dest)
# pylint: disable=unused-argument
def execute(self, command, timeout=None, check_exit_code=True,
as_root=False, strip_colors=True):
self.logger.debug(command)

12
devlib/instrument/__init__.py
View File

@ -58,6 +58,7 @@ class MeasurementType(object):
raise ValueError(msg.format(self.name, to.name))
return self.conversions[to.name](value)
# pylint: disable=undefined-variable
def __cmp__(self, other):
if isinstance(other, MeasurementType):
other = other.name
@ -151,6 +152,7 @@ class Measurement(object):
self.value = value
self.channel = channel
# pylint: disable=undefined-variable
def __cmp__(self, other):
if hasattr(other, 'value'):
return cmp(self.value, other.value)
@ -204,7 +206,7 @@ class MeasurementsCsv(object):
for mt in MEASUREMENT_TYPES:
suffix = '_{}'.format(mt)
if entry.endswith(suffix):
site = entry[:-len(suffix)]
site = entry[:-len(suffix)]
measure = mt
break
else:
@ -218,6 +220,7 @@ class MeasurementsCsv(object):
chan = InstrumentChannel(site, measure)
self.channels.append(chan)
# pylint: disable=stop-iteration-return
def _iter_rows(self):
with csvreader(self.path) as reader:
next(reader) # headings
@ -308,16 +311,16 @@ class Instrument(object):
msg = 'Unexpected channel "{}"; must be in {}'
raise ValueError(msg.format(e, self.channels.keys()))
elif sites is None and kinds is None:
self.active_channels = sorted(self.channels.itervalues(), key=lambda x: x.label)
self.active_channels = sorted(self.channels.values(), key=lambda x: x.label)
else:
if isinstance(sites, basestring):
sites = [sites]
if isinstance(kinds, basestring):
kinds = [kinds]
wanted = lambda ch : ((kinds is None or ch.kind in kinds) and
wanted = lambda ch: ((kinds is None or ch.kind in kinds) and
(sites is None or ch.site in sites))
self.active_channels = filter(wanted, self.channels.itervalues())
self.active_channels = filter(wanted, self.channels.values())
# instantaneous
@ -332,6 +335,7 @@ class Instrument(object):
def stop(self):
pass
# pylint: disable=no-self-use
def get_data(self, outfile):
pass

8
devlib/instrument/arm_energy_probe.py
View File

@ -34,8 +34,6 @@ from __future__ import division
import os
import subprocess
import signal
import struct
import sys
import tempfile
import shutil
@ -109,8 +107,8 @@ class ArmEnergyProbeInstrument(Instrument):
shell=True)
def stop(self):
self.logger.debug("kill running arm-probe")
os.killpg(self.armprobe.pid, signal.SIGTERM)
self.logger.debug("kill running arm-probe")
os.killpg(self.armprobe.pid, signal.SIGTERM)
def get_data(self, outfile): # pylint: disable=R0914
self.logger.debug("Parse data and compute consumed energy")
@ -133,7 +131,7 @@ class ArmEnergyProbeInstrument(Instrument):
if len(row) < len(active_channels):
continue
# all data are in micro (seconds/watt)
new = [ float(row[i])/1000000 for i in active_indexes ]
new = [float(row[i])/1000000 for i in active_indexes]
writer.writerow(new)
self.output_fd_error.close()

1
devlib/instrument/daq.py
View File

@ -156,4 +156,3 @@ class DaqInstrument(Instrument):
def execute(self, command, **kwargs):
return execute_command(self.server_config, command, **kwargs)

2
devlib/instrument/energy_probe.py
View File

@ -114,7 +114,7 @@ class EnergyProbeInstrument(Instrument):
writer.writerow(row)
except struct.error:
if not_a_full_row_seen:
self.logger.warn('possibly missaligned caiman raw data, row contained {} bytes'.format(len(data)))
self.logger.warning('possibly missaligned caiman raw data, row contained {} bytes'.format(len(data)))
continue
else:
not_a_full_row_seen = True

1
devlib/instrument/frames.py
View File

@ -41,6 +41,7 @@ class FramesInstrument(Instrument):
def reset(self, sites=None, kinds=None, channels=None):
super(FramesInstrument, self).reset(sites, kinds, channels)
# pylint: disable=not-callable
self.collector = self.collector_cls(self.target, self.period,
self.collector_target, self.header)
self._need_reset = False

6
devlib/instrument/gem5power.py
View File

@ -13,11 +13,10 @@
# limitations under the License.
from __future__ import division
import re
from devlib.platform.gem5 import Gem5SimulationPlatform
from devlib.instrument import Instrument, CONTINUOUS, MeasurementsCsv
from devlib.exception import TargetError, HostError
from devlib.exception import TargetError
from devlib.utils.csvutil import csvwriter
@ -69,7 +68,7 @@ class Gem5PowerInstrument(Instrument):
with csvwriter(outfile) as writer:
writer.writerow([c.label for c in self.active_channels]) # headers
sites_to_match = [self.site_mapping.get(s, s) for s in active_sites]
for rec, rois in self.target.gem5stats.match_iter(sites_to_match,
for rec, _ in self.target.gem5stats.match_iter(sites_to_match,
[self.roi_label], self._base_stats_dump):
writer.writerow([rec[s] for s in sites_to_match])
return MeasurementsCsv(outfile, self.active_channels, self.sample_rate_hz)
@ -77,4 +76,3 @@ class Gem5PowerInstrument(Instrument):
def reset(self, sites=None, kinds=None, channels=None):
super(Gem5PowerInstrument, self).reset(sites, kinds, channels)
self._base_stats_dump = self.target.gem5stats.next_dump_no()

8
devlib/instrument/monsoon.py
View File

@ -21,12 +21,11 @@ from tempfile import NamedTemporaryFile
from devlib.instrument import Instrument, CONTINUOUS, MeasurementsCsv
from devlib.exception import HostError
from devlib.host import PACKAGE_BIN_DIRECTORY
from devlib.utils.csvutil import csvwriter
from devlib.utils.misc import which
INSTALL_INSTRUCTIONS="""
INSTALL_INSTRUCTIONS = """
MonsoonInstrument requires the monsoon.py tool, available from AOSP:
https://android.googlesource.com/platform/cts/+/master/tools/utils/monsoon.py
@ -68,6 +67,7 @@ class MonsoonInstrument(Instrument):
self.process = None
self.output = None
self.buffer_file = None
self.sample_rate_hz = 500
self.add_channel('output', 'power')
@ -110,7 +110,7 @@ class MonsoonInstrument(Instrument):
process.send_signal(signal.SIGINT)
stderr = process.stderr.read()
stderr = process.stderr.read()
self.buffer_file.close()
with open(self.buffer_file.name) as f:
@ -124,7 +124,7 @@ class MonsoonInstrument(Instrument):
if self.process:
raise RuntimeError('`get_data` called before `stop`')
stdout, stderr = self.output
stdout, _ = self.output
with csvwriter(outfile) as writer:
active_sites = [c.site for c in self.active_channels]

1
devlib/instrument/netstats/__init__.py
View File

@ -101,6 +101,7 @@ class NetstatsInstrument(Instrument):
self.add_channel(package, 'tx')
self.add_channel(package, 'rx')
# pylint: disable=keyword-arg-before-vararg,arguments-differ
def setup(self, force=False, *args, **kwargs):
if self.target.package_is_installed(self.package):
if force:

4
devlib/module/__init__.py
View File

@ -61,7 +61,7 @@ class Module(object):
self.logger = logging.getLogger(self.name)
class HardRestModule(Module): # pylint: disable=R0921
class HardRestModule(Module):
kind = 'hard_reset'
@ -69,7 +69,7 @@ class HardRestModule(Module): # pylint: disable=R0921
raise NotImplementedError()
class BootModule(Module): # pylint: disable=R0921
class BootModule(Module):
kind = 'boot'

1
devlib/module/android.py
View File

@ -125,4 +125,3 @@ def get_mapping(base_dir, partition_file):
HostError('file {} was not found in the bundle or was misplaced'.format(pair[1]))
mapping[pair[0]] = image_path
return mapping

52
devlib/module/biglittle.py
View File

@ -60,150 +60,150 @@ class BigLittleModule(Module):
def list_bigs_frequencies(self):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
return self.target.cpufreq.list_frequencies(bigs_online[0])
def list_bigs_governors(self):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
return self.target.cpufreq.list_governors(bigs_online[0])
def list_bigs_governor_tunables(self):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
return self.target.cpufreq.list_governor_tunables(bigs_online[0])
def list_littles_frequencies(self):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
return self.target.cpufreq.list_frequencies(littles_online[0])
def list_littles_governors(self):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
return self.target.cpufreq.list_governors(littles_online[0])
def list_littles_governor_tunables(self):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
return self.target.cpufreq.list_governor_tunables(littles_online[0])
def get_bigs_governor(self):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
return self.target.cpufreq.get_governor(bigs_online[0])
def get_bigs_governor_tunables(self):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
return self.target.cpufreq.get_governor_tunables(bigs_online[0])
def get_bigs_frequency(self):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
return self.target.cpufreq.get_frequency(bigs_online[0])
def get_bigs_min_frequency(self):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
return self.target.cpufreq.get_min_frequency(bigs_online[0])
def get_bigs_max_frequency(self):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
return self.target.cpufreq.get_max_frequency(bigs_online[0])
def get_littles_governor(self):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
return self.target.cpufreq.get_governor(littles_online[0])
def get_littles_governor_tunables(self):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
return self.target.cpufreq.get_governor_tunables(littles_online[0])
def get_littles_frequency(self):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
return self.target.cpufreq.get_frequency(littles_online[0])
def get_littles_min_frequency(self):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
return self.target.cpufreq.get_min_frequency(littles_online[0])
def get_littles_max_frequency(self):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
return self.target.cpufreq.get_max_frequency(littles_online[0])
def set_bigs_governor(self, governor, **kwargs):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
self.target.cpufreq.set_governor(bigs_online[0], governor, **kwargs)
else:
raise ValueError("All bigs appear to be offline")
def set_bigs_governor_tunables(self, governor, **kwargs):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
self.target.cpufreq.set_governor_tunables(bigs_online[0], governor, **kwargs)
else:
raise ValueError("All bigs appear to be offline")
def set_bigs_frequency(self, frequency, exact=True):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
self.target.cpufreq.set_frequency(bigs_online[0], frequency, exact)
else:
raise ValueError("All bigs appear to be offline")
def set_bigs_min_frequency(self, frequency, exact=True):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
self.target.cpufreq.set_min_frequency(bigs_online[0], frequency, exact)
else:
raise ValueError("All bigs appear to be offline")
def set_bigs_max_frequency(self, frequency, exact=True):
bigs_online = self.bigs_online
if len(bigs_online) > 0:
if bigs_online:
self.target.cpufreq.set_max_frequency(bigs_online[0], frequency, exact)
else:
raise ValueError("All bigs appear to be offline")
def set_littles_governor(self, governor, **kwargs):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
self.target.cpufreq.set_governor(littles_online[0], governor, **kwargs)
else:
raise ValueError("All littles appear to be offline")
def set_littles_governor_tunables(self, governor, **kwargs):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
self.target.cpufreq.set_governor_tunables(littles_online[0], governor, **kwargs)
else:
raise ValueError("All littles appear to be offline")
def set_littles_frequency(self, frequency, exact=True):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
self.target.cpufreq.set_frequency(littles_online[0], frequency, exact)
else:
raise ValueError("All littles appear to be offline")
def set_littles_min_frequency(self, frequency, exact=True):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
self.target.cpufreq.set_min_frequency(littles_online[0], frequency, exact)
else:
raise ValueError("All littles appear to be offline")
def set_littles_max_frequency(self, frequency, exact=True):
littles_online = self.littles_online
if len(littles_online) > 0:
if littles_online:
self.target.cpufreq.set_max_frequency(littles_online[0], frequency, exact)
else:
raise ValueError("All littles appear to be offline")

43
devlib/module/cgroups.py
View File

@ -121,18 +121,20 @@ class Controller(object):
cgroups.append(cg)
return cgroups
def move_tasks(self, source, dest, exclude=[]):
def move_tasks(self, source, dest, exclude=None):
if exclude is None:
exclude = []
try:
srcg = self._cgroups[source]
dstg = self._cgroups[dest]
except KeyError as e:
raise ValueError('Unkown group: {}'.format(e))
output = self.target._execute_util(
raise ValueError('Unknown group: {}'.format(e))
self.target._execute_util( # pylint: disable=protected-access
'cgroups_tasks_move {} {} \'{}\''.format(
srcg.directory, dstg.directory, exclude),
as_root=True)
def move_all_tasks_to(self, dest, exclude=[]):
def move_all_tasks_to(self, dest, exclude=None):
"""
Move all the tasks to the specified CGroup
@ -145,8 +147,10 @@ class Controller(object):
tasks.
:param exclude: list of commands to keep in the root CGroup
:type exlude: list(str)
:type exclude: list(str)
"""
if exclude is None:
exclude = []
if isinstance(exclude, str):
exclude = [exclude]
@ -169,6 +173,7 @@ class Controller(object):
if cgroup != dest:
self.move_tasks(cgroup, dest, grep_filters)
# pylint: disable=too-many-locals
def tasks(self, cgroup,
filter_tid='',
filter_tname='',
@ -203,8 +208,8 @@ class Controller(object):
try:
cg = self._cgroups[cgroup]
except KeyError as e:
raise ValueError('Unkown group: {}'.format(e))
output = self.target._execute_util(
raise ValueError('Unknown group: {}'.format(e))
output = self.target._execute_util( # pylint: disable=protected-access
'cgroups_tasks_in {}'.format(cg.directory),
as_root=True)
entries = output.splitlines()
@ -234,7 +239,7 @@ class Controller(object):
try:
cg = self._cgroups[cgroup]
except KeyError as e:
raise ValueError('Unkown group: {}'.format(e))
raise ValueError('Unknown group: {}'.format(e))
output = self.target.execute(
'{} wc -l {}/tasks'.format(
self.target.busybox, cg.directory),
@ -286,7 +291,7 @@ class CGroup(object):
self.controller.kind)
logging.debug(' %s',
self.directory)
output = self.target._execute_util(
output = self.target._execute_util( # pylint: disable=protected-access
'cgroups_get_attributes {} {}'.format(
self.directory, self.controller.kind),
as_root=True)
@ -302,7 +307,7 @@ class CGroup(object):
if isiterable(attrs[idx]):
attrs[idx] = list_to_ranges(attrs[idx])
# Build attribute path
if self.controller._noprefix:
if self.controller._noprefix: # pylint: disable=protected-access
attr_name = '{}'.format(idx)
else:
attr_name = '{}.{}'.format(self.controller.kind, idx)
@ -363,7 +368,7 @@ class CgroupsModule(Module):
# Get the list of the available controllers
subsys = self.list_subsystems()
if len(subsys) == 0:
if subsys:
self.logger.warning('No CGroups controller available')
return
@ -444,11 +449,11 @@ class CgroupsModule(Module):
A regexps of tasks names can be used to defined tasks which should not
be moved.
"""
return self.target._execute_util(
return self.target._execute_util( # pylint: disable=protected-access
'cgroups_tasks_move {} {} {}'.format(srcg, dstg, exclude),
as_root=True)
def isolate(self, cpus, exclude=[]):
def isolate(self, cpus, exclude=None):
"""
Remove all userspace tasks from specified CPUs.
@ -465,6 +470,8 @@ class CgroupsModule(Module):
sandbox is the CGroup of sandboxed CPUs
isolated is the CGroup of isolated CPUs
"""
if exclude is None:
exclude = []
all_cpus = set(range(self.target.number_of_cpus))
sbox_cpus = list(all_cpus - set(cpus))
isol_cpus = list(all_cpus - set(sbox_cpus))
@ -483,7 +490,7 @@ class CgroupsModule(Module):
return sbox_cg, isol_cg
def freeze(self, exclude=[], thaw=False):
def freeze(self, exclude=None, thaw=False):
"""
Freeze all user-space tasks but the specified ones
@ -501,6 +508,9 @@ class CgroupsModule(Module):
:type thaw: bool
"""
if exclude is None:
exclude = []
# Create Freezer CGroup
freezer = self.controller('freezer')
if freezer is None:
@ -509,7 +519,8 @@ class CgroupsModule(Module):
cmd = 'cgroups_freezer_set_state {{}} {}'.format(freezer_cg.directory)
if thaw:
# Restart froozen tasks
# Restart frozen tasks
# pylint: disable=protected-access
freezer.target._execute_util(cmd.format('THAWED'), as_root=True)
# Remove all tasks from freezer
freezer.move_all_tasks_to('/')
@ -522,7 +533,7 @@ class CgroupsModule(Module):
tasks = freezer.tasks('/')
# Freeze all tasks
# pylint: disable=protected-access
freezer.target._execute_util(cmd.format('FROZEN'), as_root=True)
return tasks

2
devlib/module/cooling.py
View File

@ -37,12 +37,14 @@ class MbedFanActiveCoolingModule(Module):
with open_serial_connection(timeout=self.timeout,
port=self.port,
baudrate=self.baud) as target:
# pylint: disable=no-member
target.sendline('motor_{}_1'.format(self.fan_pin))
def stop(self):
with open_serial_connection(timeout=self.timeout,
port=self.port,
baudrate=self.baud) as target:
# pylint: disable=no-member
target.sendline('motor_{}_0'.format(self.fan_pin))

11
devlib/module/cpufreq.py
View File

@ -200,7 +200,7 @@ class CpufreqModule(Module):
could not be found.
"""
freqs = self.list_frequencies(cpu)
return freqs and max(freqs) or None
return max(freqs) if freqs else None
@memoized
def get_min_available_frequency(self, cpu):
@ -209,7 +209,7 @@ class CpufreqModule(Module):
could not be found.
"""
freqs = self.list_frequencies(cpu)
return freqs and min(freqs) or None
return min(freqs) if freqs else None
def get_min_frequency(self, cpu):
"""
@ -380,6 +380,7 @@ class CpufreqModule(Module):
"""
Set the specified (minimum) frequency for all the (online) CPUs
"""
# pylint: disable=protected-access
return self.target._execute_util(
'cpufreq_set_all_frequencies {}'.format(freq),
as_root=True)
@ -388,6 +389,7 @@ class CpufreqModule(Module):
"""
Get the current frequency for all the (online) CPUs
"""
# pylint: disable=protected-access
output = self.target._execute_util(
'cpufreq_get_all_frequencies', as_root=True)
frequencies = {}
@ -403,6 +405,7 @@ class CpufreqModule(Module):
Set the specified governor for all the (online) CPUs
"""
try:
# pylint: disable=protected-access
return self.target._execute_util(
'cpufreq_set_all_governors {}'.format(governor),
as_root=True)
@ -421,6 +424,7 @@ class CpufreqModule(Module):
"""
Get the current governor for all the (online) CPUs
"""
# pylint: disable=protected-access
output = self.target._execute_util(
'cpufreq_get_all_governors', as_root=True)
governors = {}
@ -435,6 +439,7 @@ class CpufreqModule(Module):
"""
Report current frequencies on trace file
"""
# pylint: disable=protected-access
return self.target._execute_util('cpufreq_trace_all_frequencies', as_root=True)
def get_affected_cpus(self, cpu):
@ -478,7 +483,7 @@ class CpufreqModule(Module):
"""
cpus = set(range(self.target.number_of_cpus))
while cpus:
cpu = next(iter(cpus))
cpu = next(iter(cpus)) # pylint: disable=stop-iteration-return
domain = self.target.cpufreq.get_related_cpus(cpu)
yield domain
cpus = cpus.difference(domain)

5
devlib/module/cpuidle.py
View File

@ -16,7 +16,6 @@
from past.builtins import basestring
from devlib.module import Module
from devlib.utils.misc import memoized
from devlib.utils.types import integer, boolean
@ -51,6 +50,7 @@ class CpuidleState(object):
self.desc = desc
self.power = power
self.latency = latency
self.residency = residency
self.id = self.target.path.basename(self.path)
self.cpu = self.target.path.basename(self.target.path.dirname(path))
@ -166,7 +166,8 @@ class Cpuidle(Module):
"""
Momentarily wake each CPU. Ensures cpu_idle events in trace file.
"""
output = self.target._execute_util('cpuidle_wake_all_cpus')
# pylint: disable=protected-access
self.target._execute_util('cpuidle_wake_all_cpus')
def get_driver(self):
return self.target.read_value(self.target.path.join(self.root_path, 'current_driver'))

9
devlib/module/devfreq.py
View File

@ -200,7 +200,7 @@ class DevfreqModule(Module):
Set the specified governor for all the (available) devices
"""
try:
return self.target._execute_util(
return self.target._execute_util( # pylint: disable=protected-access
'devfreq_set_all_governors {}'.format(governor), as_root=True)
except TargetError as e:
if ("echo: I/O error" in str(e) or
@ -217,7 +217,7 @@ class DevfreqModule(Module):
"""
Get the current governor for all the (online) CPUs
"""
output = self.target._execute_util(
output = self.target._execute_util( # pylint: disable=protected-access
'devfreq_get_all_governors', as_root=True)
governors = {}
for x in output.splitlines():
@ -241,7 +241,7 @@ class DevfreqModule(Module):
"""
Set the specified (minimum) frequency for all the (available) devices
"""
return self.target._execute_util(
return self.target._execute_util( # pylint: disable=protected-access
'devfreq_set_all_frequencies {}'.format(freq),
as_root=True)
@ -249,7 +249,7 @@ class DevfreqModule(Module):
"""
Get the current frequency for all the (available) devices
"""
output = self.target._execute_util(
output = self.target._execute_util( # pylint: disable=protected-access
'devfreq_get_all_frequencies', as_root=True)
frequencies = {}
for x in output.splitlines():
@ -258,4 +258,3 @@ class DevfreqModule(Module):
break
frequencies[kv[0]] = kv[1]
return frequencies

12
devlib/module/gem5stats.py
View File

@ -14,16 +14,13 @@
import re
import sys
import logging
import os.path
from collections import defaultdict
import devlib
from devlib.exception import TargetError
from devlib.exception import TargetError, HostError
from devlib.module import Module
from devlib.platform import Platform
from devlib.platform.gem5 import Gem5SimulationPlatform
from devlib.utils.gem5 import iter_statistics_dump, GEM5STATS_ROI_NUMBER, GEM5STATS_DUMP_TAIL
from devlib.utils.gem5 import iter_statistics_dump, GEM5STATS_ROI_NUMBER
class Gem5ROI:
@ -158,7 +155,7 @@ class Gem5StatsModule(Module):
}
}
'''
records = defaultdict(lambda : defaultdict(list))
records = defaultdict(lambda: defaultdict(list))
for record, active_rois in self.match_iter(keys, rois_labels, base_dump):
for key in record:
for roi_label in active_rois:
@ -243,7 +240,7 @@ class Gem5StatsModule(Module):
dump_iterator = iter_statistics_dump(stats_file)
while curr_dump < target_dump:
try:
dump = next(dump_iterator)
next(dump_iterator)
except StopIteration:
break
# End of passed dump is beginning og next one
@ -251,4 +248,3 @@ class Gem5StatsModule(Module):
curr_dump += 1
self._dump_pos_cache[curr_dump] = curr_pos
return curr_dump

9
devlib/module/gpufreq.py
View File

@ -28,7 +28,6 @@
# limitations under the License.
import re
import json
from devlib.module import Module
from devlib.exception import TargetError
from devlib.utils.misc import memoized
@ -57,7 +56,7 @@ class GpufreqModule(Module):
def set_governor(self, governor):
if governor not in self.governors:
raise TargetError('Governor {} not supported for gpu {}'.format(governor, cpu))
raise TargetError('Governor {} not supported for gpu'.format(governor))
self.target.write_value("/sys/kernel/gpu/gpu_governor", governor)
def get_frequencies(self):
@ -85,6 +84,6 @@ class GpufreqModule(Module):
Returns the model name reported by the GPU.
"""
try:
return self.target.read_value("/sys/kernel/gpu/gpu_model")
except:
return "unknown"
return self.target.read_value("/sys/kernel/gpu/gpu_model")
except: # pylint: disable=bare-except
return "unknown"

3
devlib/module/hotplug.py
View File

@ -36,7 +36,7 @@ class HotplugModule(Module):
return target.path.join(cls.base_path, cpu, 'online')
def online_all(self):
self.target._execute_util('hotplug_online_all',
self.target._execute_util('hotplug_online_all', # pylint: disable=protected-access
as_root=self.target.is_rooted)
def online(self, *args):
@ -53,4 +53,3 @@ class HotplugModule(Module):
return
value = 1 if online else 0
self.target.write_value(path, value)

2
devlib/module/hwmon.py
View File

@ -12,7 +12,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#
import os
import re
from collections import defaultdict
@ -147,4 +146,3 @@ class HwmonModule(Module):
self.logger.debug('Adding device {}'.format(name))
device = HwmonDevice(self.target, path, name, fields)
self.devices.append(device)

5
devlib/module/sched.py
View File

@ -30,11 +30,11 @@
import logging
import re
from past.builtins import basestring
from devlib.module import Module
from devlib.utils.misc import memoized
from past.builtins import basestring
class SchedProcFSNode(object):
"""
@ -156,6 +156,7 @@ class SchedDomain(SchedProcFSNode):
"""
Represents a sched domain as seen through procfs
"""
# pylint: disable=bad-whitespace
# Domain flags obtained from include/linux/sched/topology.h on v4.17
# https://kernel.googlesource.com/pub/scm/linux/kernel/git/torvalds/linux/+/v4.17/include/linux/sched/topology.h#20
SD_LOAD_BALANCE = 0x0001 # Do load balancing on this domain.

17
devlib/module/vexpress.py
View File

@ -209,6 +209,7 @@ class VexpressUefiShellBoot(VexpressBootModule):
name = 'vexpress-uefi-shell'
# pylint: disable=keyword-arg-before-vararg
def __init__(self, target, uefi_entry='^Shell$',
efi_shell_prompt='Shell>',
image='kernel', bootargs=None,
@ -239,6 +240,7 @@ class VexpressUBoot(VexpressBootModule):
name = 'vexpress-u-boot'
# pylint: disable=keyword-arg-before-vararg
def __init__(self, target, env=None,
*args, **kwargs):
super(VexpressUBoot, self).__init__(target, *args, **kwargs)
@ -260,6 +262,7 @@ class VexpressBootmon(VexpressBootModule):
name = 'vexpress-bootmon'
# pylint: disable=keyword-arg-before-vararg
def __init__(self, target,
image, fdt, initrd, bootargs,
uses_bootscript=False,
@ -282,11 +285,11 @@ class VexpressBootmon(VexpressBootModule):
with open_serial_connection(port=self.port,
baudrate=self.baudrate,
timeout=self.timeout,
init_dtr=0) as tty:
write_characters(tty, 'fl linux fdt {}'.format(self.fdt))
write_characters(tty, 'fl linux initrd {}'.format(self.initrd))
write_characters(tty, 'fl linux boot {} {}'.format(self.image,
self.bootargs))
init_dtr=0) as tty_conn:
write_characters(tty_conn, 'fl linux fdt {}'.format(self.fdt))
write_characters(tty_conn, 'fl linux initrd {}'.format(self.initrd))
write_characters(tty_conn, 'fl linux boot {} {}'.format(self.image,
self.bootargs))
class VersatileExpressFlashModule(FlashModule):
@ -328,9 +331,10 @@ class VersatileExpressFlashModule(FlashModule):
baudrate=self.target.platform.baudrate,
timeout=self.timeout,
init_dtr=0) as tty:
# pylint: disable=no-member
i = tty.expect([self.mcc_prompt, AUTOSTART_MESSAGE, OLD_AUTOSTART_MESSAGE])
if i:
tty.sendline('')
tty.sendline('') # pylint: disable=no-member
wait_for_vemsd(self.vemsd_mount, tty, self.mcc_prompt, self.short_delay)
try:
if image_bundle:
@ -387,4 +391,3 @@ def wait_for_vemsd(vemsd_mount, tty, mcc_prompt=DEFAULT_MCC_PROMPT, short_delay=
if os.path.exists(path):
return
raise TargetError('Could not mount {}'.format(vemsd_mount))

12
devlib/platform/arm.py
View File

@ -19,10 +19,11 @@ import tempfile
import time
import pexpect
from devlib.platform import Platform
from devlib.instrument import Instrument, InstrumentChannel, MeasurementsCsv, Measurement, CONTINUOUS, INSTANTANEOUS
from devlib.exception import TargetError, HostError
from devlib.host import PACKAGE_BIN_DIRECTORY
from devlib.instrument import (Instrument, InstrumentChannel, MeasurementsCsv,
Measurement, CONTINUOUS, INSTANTANEOUS)
from devlib.platform import Platform
from devlib.utils.csvutil import csvreader, csvwriter
from devlib.utils.serial_port import open_serial_connection
@ -99,6 +100,7 @@ class VersatileExpressPlatform(Platform):
addr = self._get_target_ip_address(target)
target.connection_settings['host'] = addr
# pylint: disable=no-member
def _get_target_ip_address(self, target):
with open_serial_connection(port=self.serial_port,
baudrate=self.baudrate,
@ -250,7 +252,7 @@ class JunoEnergyInstrument(Instrument):
self.command = '{} -o {}'.format(self.binary, self.on_target_file)
self.command2 = '{}'.format(self.binary)
def setup(self):
def setup(self): # pylint: disable=arguments-differ
self.binary = self.target.install(os.path.join(PACKAGE_BIN_DIRECTORY,
self.target.abi, self.binname))
self.command = '{} -o {}'.format(self.binary, self.on_target_file)
@ -266,6 +268,7 @@ class JunoEnergyInstrument(Instrument):
def stop(self):
self.target.killall(self.binname, signal='TERM', as_root=True)
# pylint: disable=arguments-differ
def get_data(self, output_file):
temp_file = tempfile.mktemp()
self.target.pull(self.on_target_file, temp_file)
@ -296,10 +299,9 @@ class JunoEnergyInstrument(Instrument):
result = []
output = self.target.execute(self.command2).split()
with csvreader(output) as reader:
headings=next(reader)
headings = next(reader)
values = next(reader)
for chan in self.active_channels:
value = values[headings.index(chan.name)]
result.append(Measurement(value, chan))
return result

16
devlib/platform/gem5.py
View File

@ -15,7 +15,6 @@
import os
import re
import subprocess
import sys
import shutil
import time
import types
@ -55,7 +54,7 @@ class Gem5SimulationPlatform(Platform):
self.stdout_file = None
self.stderr_file = None
self.stderr_filename = None
if self.gem5_port is None:
if self.gem5_port is None: # pylint: disable=simplifiable-if-statement
# Allows devlib to pick up already running simulations
self.start_gem5_simulation = True
else:
@ -234,6 +233,7 @@ class Gem5SimulationPlatform(Platform):
# Call the general update_from_target implementation
super(Gem5SimulationPlatform, self).update_from_target(target)
def gem5_capture_screen(self, filepath):
file_list = os.listdir(self.gem5_out_dir)
screen_caps = []
@ -243,6 +243,7 @@ class Gem5SimulationPlatform(Platform):
if '{ts}' in filepath:
cmd = '{} date -u -Iseconds'
# pylint: disable=no-member
ts = self.target.execute(cmd.format(self.target.busybox)).strip()
filepath = filepath.format(ts=ts)
@ -258,6 +259,7 @@ class Gem5SimulationPlatform(Platform):
im.save(temp_image, "PNG")
shutil.copy(temp_image, filepath)
os.remove(temp_image)
# pylint: disable=undefined-variable
gem5_logger.info("capture_screen: using gem5 screencap")
successful_capture = True
@ -266,12 +268,14 @@ class Gem5SimulationPlatform(Platform):
return successful_capture
# pylint: disable=no-self-use
def _deploy_m5(self, target):
# m5 is not yet installed so install it
host_executable = os.path.join(PACKAGE_BIN_DIRECTORY,
target.abi, 'm5')
return target.install(host_executable)
# pylint: disable=no-self-use
def _resize_shell(self, target):
"""
Resize the shell to avoid line wrapping issues.
@ -282,18 +286,16 @@ class Gem5SimulationPlatform(Platform):
target.execute('reset', check_exit_code=False)
# Methods that will be monkey-patched onto the target
def _overwritten_reset(self):
def _overwritten_reset(self): # pylint: disable=unused-argument
raise TargetError('Resetting is not allowed on gem5 platforms!')
def _overwritten_reboot(self):
def _overwritten_reboot(self): # pylint: disable=unused-argument
raise TargetError('Rebooting is not allowed on gem5 platforms!')
def _overwritten_capture_screen(self, filepath):
connection_screencapped = self.platform.gem5_capture_screen(filepath)
if connection_screencapped == False:
if not connection_screencapped:
# The connection was not able to capture the screen so use the target
# implementation
self.logger.debug('{} was not able to screen cap, using the original target implementation'.format(self.platform.__class__.__name__))
self.target_impl_capture_screen(filepath)

20
devlib/target.py
View File

@ -29,7 +29,7 @@ from collections import namedtuple
from devlib.host import LocalConnection, PACKAGE_BIN_DIRECTORY
from devlib.module import get_module
from devlib.platform import Platform
from devlib.exception import TargetError, TargetNotRespondingError, TimeoutError
from devlib.exception import TargetError, TargetNotRespondingError, TimeoutError # pylint: disable=redefined-builtin
from devlib.utils.ssh import SshConnection
from devlib.utils.android import AdbConnection, AndroidProperties, LogcatMonitor, adb_command, adb_disconnect, INTENT_FLAGS
from devlib.utils.misc import memoized, isiterable, convert_new_lines
@ -45,7 +45,7 @@ ANDROID_SCREEN_RESOLUTION_REGEX = re.compile(r'mUnrestrictedScreen=\(\d+,\d+\)'
r'\s+(?P<width>\d+)x(?P<height>\d+)')
DEFAULT_SHELL_PROMPT = re.compile(r'^.*(shell|root|juno)@?.*:[/~]\S* *[#$] ',
re.MULTILINE)
KVERSION_REGEX =re.compile(
KVERSION_REGEX = re.compile(
r'(?P<version>\d+)(\.(?P<major>\d+)(\.(?P<minor>\d+)(-rc(?P<rc>\d+))?)?)?(.*-g(?P<sha1>[0-9a-fA-F]{7,}))?'
)
@ -222,6 +222,7 @@ class Target(object):
self._cache = {}
self._connections = {}
self._shutils = None
self._file_transfer_cache = None
self.busybox = None
if load_default_modules:
@ -257,7 +258,7 @@ class Target(object):
self._connections = {}
def get_connection(self, timeout=None):
if self.conn_cls == None:
if self.conn_cls is None:
raise ValueError('Connection class not specified on Target creation.')
return self.conn_cls(timeout=timeout, **self.connection_settings) # pylint: disable=not-callable
@ -332,8 +333,8 @@ class Target(object):
# Host location of dir
outdir = os.path.join(dest, tar_file_name)
# Host location of archive
tar_file_name = '{}.tar'.format(tar_file_name)
tempfile = os.path.join(dest, tar_file_name)
tar_file_name = '{}.tar'.format(tar_file_name)
tmpfile = os.path.join(dest, tar_file_name)
# If root is required, use tmp location for tar creation.
if as_root:
@ -351,11 +352,11 @@ class Target(object):
# Pull the file
if not os.path.exists(dest):
os.mkdir(dest)
self.pull(tar_file_name, tempfile)
self.pull(tar_file_name, tmpfile)
# Decompress
f = tarfile.open(tempfile, 'r')
f = tarfile.open(tmpfile, 'r')
f.extractall(outdir)
os.remove(tempfile)
os.remove(tmpfile)
# execution
@ -1113,7 +1114,7 @@ class AndroidTarget(Target):
as_root = self.needs_su
try:
command = 'cd {} && {} nohup {} &'.format(self.working_directory, self.busybox, command)
output = self.execute(command, timeout=1, as_root=as_root)
self.execute(command, timeout=1, as_root=as_root)
except TimeoutError:
pass
@ -1847,6 +1848,7 @@ class ChromeOsTarget(LinuxTarget):
os = 'chromeos'
# pylint: disable=too-many-locals
def __init__(self,
connection_settings=None,
platform=None,

9
devlib/trace/ftrace.py
View File

@ -50,6 +50,7 @@ STATS_RE = re.compile(r'([^ ]*) +([0-9]+) +([0-9.]+) us +([0-9.]+) us +([0-9.]+)
class FtraceCollector(TraceCollector):
# pylint: disable=too-many-locals,too-many-branches,too-many-statements
def __init__(self, target,
events=None,
functions=None,
@ -84,6 +85,7 @@ class FtraceCollector(TraceCollector):
self.function_string = None
self._reset_needed = True
# pylint: disable=bad-whitespace
# Setup tracing paths
self.available_events_file = self.target.path.join(self.tracing_path, 'available_events')
self.available_functions_file = self.target.path.join(self.tracing_path, 'available_filter_functions')
@ -122,7 +124,7 @@ class FtraceCollector(TraceCollector):
_event = '*' + event
event_re = re.compile(_event.replace('*', '.*'))
# Select events matching the required ones
if len(list(filter(event_re.match, available_events))) == 0:
if not list(filter(event_re.match, available_events)):
message = 'Event [{}] not available for tracing'.format(event)
if strict:
raise TargetError(message)
@ -133,7 +135,7 @@ class FtraceCollector(TraceCollector):
# Thus, if not other events have been specified, try to add at least
# a tracepoint which is always available and possibly triggered few
# times.
if self.functions and len(selected_events) == 0:
if self.functions and not selected_events:
selected_events = ['sched_wakeup_new']
self.event_string = _build_trace_events(selected_events)
@ -237,6 +239,7 @@ class FtraceCollector(TraceCollector):
if os.path.isdir(outfile):
outfile = os.path.join(outfile, OUTPUT_PROFILE_FILE)
# pylint: disable=protected-access
output = self.target._execute_util('ftrace_get_function_stats',
as_root=True)
@ -264,7 +267,7 @@ class FtraceCollector(TraceCollector):
self.logger.debug("FTrace stats output [%s]...", outfile)
with open(outfile, 'w') as fh:
json.dump(function_stats, fh, indent=4)
json.dump(function_stats, fh, indent=4)
self.logger.debug("FTrace function stats save in [%s]", outfile)
return function_stats

2
devlib/trace/logcat.py
View File

@ -14,7 +14,6 @@
#
import os
import re
import shutil
from devlib.trace import TraceCollector
@ -27,6 +26,7 @@ class LogcatCollector(TraceCollector):
self.regexps = regexps
self._collecting = False
self._prev_log = None
self._monitor = None
def reset(self):
"""

2
devlib/trace/serial_trace.py
View File

@ -13,9 +13,9 @@
# limitations under the License.
#
from pexpect.exceptions import TIMEOUT
import shutil
from tempfile import NamedTemporaryFile
from pexpect.exceptions import TIMEOUT
from devlib.trace import TraceCollector
from devlib.utils.serial_port import get_connection

1
devlib/trace/systrace.py
View File

@ -122,6 +122,7 @@ class SystraceCollector(TraceCollector):
def _build_cmd(self):
self._tmpfile = NamedTemporaryFile()
# pylint: disable=attribute-defined-outside-init
self.systrace_cmd = '{} -o {} -e {}'.format(
self.systrace_binary,
self._tmpfile.name,

2
devlib/utils/__init__.py
View File

@ -12,5 +12,3 @@
# See the License for the specific language governing permissions and
# limitations under the License.
#

32
devlib/utils/android.py
View File

@ -20,21 +20,18 @@ Utility functions for working with Android devices through adb.
"""
# pylint: disable=E1103
import os
import pexpect
import time
import subprocess
import logging
import re
import threading
import tempfile
import queue
import sys
import time
import logging
import tempfile
import subprocess
from collections import defaultdict
import pexpect
from devlib.exception import TargetError, HostError, DevlibError
from devlib.utils.misc import check_output, which, memoized, ABI_MAP
from devlib.exception import TargetError, HostError
from devlib.utils.misc import check_output, which, ABI_MAP
from devlib.utils.misc import escape_single_quotes, escape_double_quotes
from devlib import host
logger = logging.getLogger('android')
@ -117,6 +114,7 @@ class AdbDevice(object):
self.name = name
self.status = status
# pylint: disable=undefined-variable
def __cmp__(self, other):
if isinstance(other, AdbDevice):
return cmp(self.name, other.name)
@ -146,6 +144,7 @@ class ApkInfo(object):
self.permissions = []
self.parse(path)
# pylint: disable=too-many-branches
def parse(self, apk_path):
_check_env()
command = [aapt, 'dump', 'badging', apk_path]
@ -222,6 +221,7 @@ class AdbConnection(object):
self.ls_command = 'ls'
logger.debug("ls command is set to {}".format(self.ls_command))
# pylint: disable=unused-argument
def __init__(self, device=None, timeout=None, platform=None, adb_server=None):
self.timeout = timeout if timeout is not None else self.default_timeout
if device is None:
@ -255,6 +255,7 @@ class AdbConnection(object):
command = "pull '{}' '{}'".format(source, dest)
return adb_command(self.device, command, timeout=timeout, adb_server=self.adb_server)
# pylint: disable=unused-argument
def execute(self, command, timeout=None, check_exit_code=False,
as_root=False, strip_colors=True):
return adb_shell(self.device, command, timeout, check_exit_code,
@ -365,12 +366,13 @@ def _ping(device):
command = "adb{} shell \"ls /data/local/tmp > /dev/null\"".format(device_string)
logger.debug(command)
result = subprocess.call(command, stderr=subprocess.PIPE, shell=True)
if not result:
if not result: # pylint: disable=simplifiable-if-statement
return True
else:
return False
# pylint: disable=too-many-locals
def adb_shell(device, command, timeout=None, check_exit_code=False,
as_root=False, adb_server=None): # NOQA
_check_env()
@ -443,7 +445,7 @@ def adb_background_shell(device, command,
def adb_list_devices(adb_server=None):
output = adb_command(None, 'devices',adb_server=adb_server)
output = adb_command(None, 'devices', adb_server=adb_server)
devices = []
for line in output.splitlines():
parts = [p.strip() for p in line.split()]
@ -575,6 +577,8 @@ class LogcatMonitor(object):
self.target = target
self._regexps = regexps
self._logcat = None
self._logfile = None
def start(self, outfile=None):
"""
@ -651,7 +655,7 @@ class LogcatMonitor(object):
return [line for line in fh]
def clear_log(self):
with open(self._logfile.name, 'w') as fh:
with open(self._logfile.name, 'w') as _:
pass
def search(self, regexp):
@ -679,7 +683,7 @@ class LogcatMonitor(object):
res = [line for line in log if re.match(regexp, line)]
# Found some matches, return them
if len(res) > 0:
if res:
return res
# Store the number of lines we've searched already, so we don't have to

3
devlib/utils/gem5.py
View File

@ -46,8 +46,7 @@ def iter_statistics_dump(stats_file):
vtext = res.group("value")
try:
v = list(map(numeric, vtext.split()))
cur_dump[k] = v[0] if len(v)==1 else set(v)
cur_dump[k] = v[0] if len(v) == 1 else set(v)
except ValueError:
msg = 'Found non-numeric entry in gem5 stats ({}: {})'
logger.warning(msg.format(k, vtext))

43
devlib/utils/misc.py
View File

@ -19,27 +19,28 @@ Miscellaneous functions that don't fit anywhere else.
"""
from __future__ import division
import os
import sys
import re
import string
import threading
import signal
import subprocess
import pkgutil
import logging
import random
import ctypes
import threading
from operator import itemgetter
from functools import partial, reduce
from itertools import groupby
from functools import partial
from operator import itemgetter
import ctypes
import logging
import os
import pkgutil
import random
import re
import signal
import string
import subprocess
import sys
import threading
import wrapt
from past.builtins import basestring
# pylint: disable=redefined-builtin
from devlib.exception import HostError, TimeoutError
from functools import reduce
# ABI --> architectures list
@ -184,7 +185,7 @@ def check_output(command, timeout=None, ignore=None, inputtext=None,
# Currently errors=replace is needed as 0x8c throws an error
output = output.decode(sys.stdout.encoding, "replace")
if error:
error = error.decode(sys.stderr.encoding, "replace")
error = error.decode(sys.stderr.encoding, "replace")
finally:
if timeout:
timer.cancel()
@ -523,6 +524,12 @@ def get_random_string(length):
class LoadSyntaxError(Exception):
@property
def message(self):
if self.args:
return self.args[0]
return str(self)
def __init__(self, message, filepath, lineno):
super(LoadSyntaxError, self).__init__(message)
self.filepath = filepath
@ -535,6 +542,7 @@ class LoadSyntaxError(Exception):
RAND_MOD_NAME_LEN = 30
BAD_CHARS = string.punctuation + string.whitespace
# pylint: disable=no-member
if sys.version_info[0] == 3:
TRANS_TABLE = str.maketrans(BAD_CHARS, '_' * len(BAD_CHARS))
else:
@ -639,7 +647,7 @@ def __get_memo_id(obj):
@wrapt.decorator
def memoized(wrapped, instance, args, kwargs):
def memoized(wrapped, instance, args, kwargs): # pylint: disable=unused-argument
"""A decorator for memoizing functions and methods."""
func_id = repr(wrapped)
@ -652,4 +660,3 @@ def memoized(wrapped, instance, args, kwargs):
return __memo_cache[id_string]
return memoize_wrapper(*args, **kwargs)

118
devlib/utils/parse_aep.py
View File

@ -28,18 +28,14 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import os
import sys
import getopt
import subprocess
import logging
import signal
import serial
import time
import math
import sys
logger = logging.getLogger('aep-parser')
# pylint: disable=attribute-defined-outside-init
class AepParser(object):
prepared = False
@ -94,7 +90,7 @@ class AepParser(object):
continue
if parent not in virtual:
virtual[parent] = { supply : index }
virtual[parent] = {supply : index}
virtual[parent][supply] = index
@ -102,7 +98,7 @@ class AepParser(object):
# child
for supply in list(virtual.keys()):
if len(virtual[supply]) == 1:
del virtual[supply];
del virtual[supply]
for supply in list(virtual.keys()):
# Add label, hide and duplicate columns for virtual domains
@ -121,7 +117,7 @@ class AepParser(object):
label[0] = array[0]
unit[0] = "(S)"
for i in range(1,len(array)):
for i in range(1, len(array)):
label[i] = array[i][:-3]
unit[i] = array[i][-3:]
@ -138,7 +134,7 @@ class AepParser(object):
# By default we assume that there is no child
duplicate = [0] * len(label)
for i in range(len(label)):
for i in range(len(label)): # pylint: disable=consider-using-enumerate
# We only care about time and Watt
if label[i] == 'time':
hide[i] = 0
@ -167,7 +163,7 @@ class AepParser(object):
@staticmethod
def parse_text(array, hide):
data = [0]*len(array)
for i in range(len(array)):
for i in range(len(array)): # pylint: disable=consider-using-enumerate
if hide[i]:
continue
@ -193,18 +189,18 @@ class AepParser(object):
return data
@staticmethod
def delta_nrj(array, delta, min, max, hide):
def delta_nrj(array, delta, minimu, maximum, hide):
# Compute the energy consumed in this time slice and add it
# delta[0] is used to save the last time stamp
if (delta[0] < 0):
if delta[0] < 0:
delta[0] = array[0]
time = array[0] - delta[0]
if (time <= 0):
if time <= 0:
return delta
for i in range(len(array)):
for i in range(len(array)): # pylint: disable=consider-using-enumerate
if hide[i]:
continue
@ -213,10 +209,10 @@ class AepParser(object):
except ValueError:
continue
if (data < min[i]):
min[i] = data
if (data > max[i]):
max[i] = data
if data < minimu[i]:
minimu[i] = data
if data > maximum[i]:
maximum[i] = data
delta[i] += time * data
# save last time stamp
@ -225,11 +221,11 @@ class AepParser(object):
return delta
def output_label(self, label, hide):
self.fo.write(label[0]+"(uS)")
self.fo.write(label[0] + "(uS)")
for i in range(1, len(label)):
if hide[i]:
continue
self.fo.write(" "+label[i]+"(uW)")
self.fo.write(" " + label[i] + "(uW)")
self.fo.write("\n")
@ -248,34 +244,34 @@ class AepParser(object):
self.fo.write("\n")
def prepare(self, infile, outfile, summaryfile):
# pylint: disable-redefined-outer-name,
def prepare(self, input_file, outfile, summaryfile):
try:
self.fi = open(infile, "r")
self.fi = open(input_file, "r")
except IOError:
logger.warn('Unable to open input file {}'.format(infile))
logger.warn('Usage: parse_arp.py -i <inputfile> [-o <outputfile>]')
logger.warning('Unable to open input file {}'.format(input_file))
logger.warning('Usage: parse_arp.py -i <inputfile> [-o <outputfile>]')
sys.exit(2)
self.parse = True
if len(outfile) > 0:
if outfile:
try:
self.fo = open(outfile, "w")
except IOError:
logger.warn('Unable to create {}'.format(outfile))
logger.warning('Unable to create {}'.format(outfile))
self.parse = False
else:
self.parse = False
self.parse = False
self.summary = True
if len(summaryfile) > 0:
if summaryfile:
try:
self.fs = open(summaryfile, "w")
except IOError:
logger.warn('Unable to create {}'.format(summaryfile))
logger.warning('Unable to create {}'.format(summaryfile))
self.fs = sys.stdout
else:
self.fs = sys.stdout
self.fs = sys.stdout
self.prepared = True
@ -291,7 +287,7 @@ class AepParser(object):
self.prepared = False
# pylint: disable=too-many-branches,too-many-statements
# pylint: disable=too-many-branches,too-many-statements,redefined-outer-name,too-many-locals
def parse_aep(self, start=0, length=-1):
# Parse aep data and calculate the energy consumed
begin = 0
@ -303,7 +299,7 @@ class AepParser(object):
lines = self.fi.readlines()
for myline in lines:
array = myline.split()
array = myline.split()
if "#" in myline:
# update power topology
@ -332,8 +328,8 @@ class AepParser(object):
# Init arrays
nrj = [0]*len(label)
min = [100000000]*len(label)
max = [0]*len(label)
minimum = [100000000]*len(label)
maximum = [0]*len(label)
offset = [0]*len(label)
continue
@ -357,7 +353,7 @@ class AepParser(object):
data = self.add_virtual_data(data, virtual)
# extract power figures
self.delta_nrj(data, nrj, min, max, hide)
self.delta_nrj(data, nrj, minimum, maximum, hide)
# write data into new file
if self.parse:
@ -366,7 +362,6 @@ class AepParser(object):
# if there is no data just return
if label_line or len(nrj) == 1:
raise ValueError('No data found in the data file. Please check the Arm Energy Probe')
return
# display energy consumption of each channel and total energy consumption
total = 0
@ -378,27 +373,33 @@ class AepParser(object):
nrj[i] -= offset[i] * nrj[0]
total_nrj = nrj[i]/1000000000000.0
duration = (max[0]-min[0])/1000000.0
duration = (maximum[0]-minimum[0])/1000000.0
channel_name = label[i]
average_power = total_nrj/duration
self.fs.write("Total nrj: %8.3f J for %s -- duration %8.3f sec -- min %8.3f W -- max %8.3f W\n" % (nrj[i]/1000000000000.0, label[i], (max[0]-min[0])/1000000.0, min[i]/1000000.0, max[i]/1000000.0))
total = nrj[i]/1000000000000.0
duration = (maximum[0]-minimum[0])/1000000.0
min_power = minimum[i]/1000000.0
max_power = maximum[i]/1000000.0
output = "Total nrj: %8.3f J for %s -- duration %8.3f sec -- min %8.3f W -- max %8.3f W\n"
self.fs.write(output.format(total, label[i], duration, min_power, max_power))
# store each AEP channel info except Platform in the results table
results_table[channel_name] = total_nrj, average_power
if (min[i] < offset[i]):
self.fs.write ("!!! Min below offset\n")
if minimum[i] < offset[i]:
self.fs.write("!!! Min below offset\n")
if duplicate[i]:
continue
total += nrj[i]
self.fs.write ("Total nrj: %8.3f J for %s -- duration %8.3f sec\n" % (total/1000000000000.0, "Platform ", (max[0]-min[0])/1000000.0))
output = "Total nrj: %8.3f J for Platform -- duration %8.3f sec\n"
self.fs.write(output.format(total/1000000000000.0, (maximum[0]-minimum[0])/1000000.0))
total_nrj = total/1000000000000.0
duration = (max[0]-min[0])/1000000.0
duration = (maximum[0]-minimum[0])/1000000.0
average_power = total_nrj/duration
# store AEP Platform channel info in the results table
@ -406,11 +407,12 @@ class AepParser(object):
return results_table
# pylint: disable=too-many-branches,no-self-use,too-many-locals
def topology_from_config(self, topofile):
try:
ft = open(topofile, "r")
except IOError:
logger.warn('Unable to open config file {}'.format(topofile))
logger.warning('Unable to open config file {}'.format(topofile))
return
lines = ft.readlines()
@ -452,10 +454,11 @@ class AepParser(object):
topo[items[0]] = info
# Increase index
index +=1
index += 1
# Create an entry for each virtual parent
# pylint: disable=consider-iterating-dictionary
for supply in topo.keys():
# Parent is in the topology
parent = topo[supply]['parent']
@ -463,23 +466,25 @@ class AepParser(object):
continue
if parent not in virtual:
virtual[parent] = { supply : topo[supply]['index'] }
virtual[parent] = {supply : topo[supply]['index']}
virtual[parent][supply] = topo[supply]['index']
# Remove parent with 1 child as they don't give more information than their
# child
# pylint: disable=consider-iterating-dictionary
for supply in list(virtual.keys()):
if len(virtual[supply]) == 1:
del virtual[supply];
del virtual[supply]
topo_list = ['']*(1+len(topo)+len(virtual))
topo_list[0] = 'time'
# pylint: disable=consider-iterating-dictionary
for chnl in topo.keys():
topo_list[topo[chnl]['index']] = chnl
for chnl in virtual.keys():
index +=1
index += 1
topo_list[index] = chnl
ft.close()
@ -491,6 +496,7 @@ class AepParser(object):
if __name__ == '__main__':
# pylint: disable=unused-argument
def handleSigTERM(signum, frame):
sys.exit(2)
@ -502,8 +508,8 @@ if __name__ == '__main__':
ch.setLevel(logging.DEBUG)
logger.addHandler(ch)
infile = ""
outfile = ""
in_file = ""
out_file = ""
figurefile = ""
start = 0
length = -1
@ -516,22 +522,22 @@ if __name__ == '__main__':
for o, a in opts:
if o == "-i":
infile = a
in_file = a
if o == "-v":
logger.setLevel(logging.DEBUG)
if o == "-o":
parse = True
outfile = a
out_file = a
if o == "-s":
start = int(float(a)*1000000)
if o == "-l":
length = int(float(a)*1000000)
if o == "-t":
topofile = a
topfile = a
parser = AepParser()
print(parser.topology_from_config(topofile))
print(parser.topology_from_config(topfile))
exit(0)
parser = AepParser()
parser.prepare(infile, outfile, figurefile)
parser.prepare(in_file, out_file, figurefile)
parser.parse_aep(start, length)

9
devlib/utils/rendering.py
View File

@ -15,15 +15,14 @@
import logging
import os
import re
import shutil
import sys
import tempfile
import threading
import time
from collections import namedtuple, OrderedDict
from distutils.version import LooseVersion
from collections import namedtuple
# pylint: disable=redefined-builtin
from devlib.exception import WorkerThreadError, TargetNotRespondingError, TimeoutError
from devlib.utils.csvutil import csvwriter
@ -190,7 +189,7 @@ class GfxinfoFrameCollector(FrameCollector):
def __init__(self, target, period, package, header=None):
super(GfxinfoFrameCollector, self).__init__(target, period)
self.package = package
self.header = None
self.header = None
self._init_header(header)
def collect_frames(self, wfh):
@ -261,7 +260,7 @@ def gfxinfo_get_last_dump(filepath):
ix = buf.find(' **\n')
if ix >= 0:
buf = next(fh_iter) + buf
buf = next(fh_iter) + buf
ix = buf.find('** Graphics')
if ix < 0:
msg = '"{}" appears to be corrupted'

4
devlib/utils/serial_port.py
View File

@ -20,6 +20,7 @@ from logging import Logger
import serial
# pylint: disable=import-error,wrong-import-position,ungrouped-imports,wrong-import-order
import pexpect
from distutils.version import StrictVersion as V
if V(pexpect.__version__) < V('4.0.0'):
@ -48,6 +49,7 @@ def pulse_dtr(conn, state=True, duration=0.1):
conn.setDTR(not state)
# pylint: disable=keyword-arg-before-vararg
def get_connection(timeout, init_dtr=None, logcls=SerialLogger,
logfile=None, *args, **kwargs):
if init_dtr is not None:
@ -89,6 +91,7 @@ def write_characters(conn, line, delay=0.05):
conn.sendline('')
# pylint: disable=keyword-arg-before-vararg
@contextmanager
def open_serial_connection(timeout, get_conn=False, init_dtr=None,
logcls=SerialLogger, *args, **kwargs):
@ -118,4 +121,3 @@ def open_serial_connection(timeout, get_conn=False, init_dtr=None,
target.close() # Closes the file descriptor used by the conn.
del conn

25
devlib/utils/ssh.py
View File

@ -26,6 +26,7 @@ import socket
import sys
import time
# pylint: disable=import-error,wrong-import-position,ungrouped-imports,wrong-import-order
import pexpect
from distutils.version import StrictVersion as V
if V(pexpect.__version__) < V('4.0.0'):
@ -34,6 +35,7 @@ else:
from pexpect import pxssh
from pexpect import EOF, TIMEOUT, spawn
# pylint: disable=redefined-builtin,wrong-import-position
from devlib.exception import HostError, TargetError, TimeoutError
from devlib.utils.misc import which, strip_bash_colors, check_output
from devlib.utils.misc import (escape_single_quotes, escape_double_quotes,
@ -73,7 +75,7 @@ def ssh_get_shell(host, username, password=None, keyfile=None, port=None, timeou
raise TargetError(message.format(host))
time.sleep(5)
conn.setwinsize(500,200)
conn.setwinsize(500, 200)
conn.sendline('')
conn.prompt()
conn.setecho(False)
@ -147,12 +149,13 @@ class SshConnection(object):
default_password_prompt = '[sudo] password'
max_cancel_attempts = 5
default_timeout=10
default_timeout = 10
@property
def name(self):
return self.host
# pylint: disable=unused-argument,super-init-not-called
def __init__(self,
host,
username,
@ -310,6 +313,7 @@ class SshConnection(object):
class TelnetConnection(SshConnection):
# pylint: disable=super-init-not-called
def __init__(self,
host,
username,
@ -333,6 +337,7 @@ class TelnetConnection(SshConnection):
class Gem5Connection(TelnetConnection):
# pylint: disable=super-init-not-called
def __init__(self,
platform,
host=None,
@ -348,9 +353,9 @@ class Gem5Connection(TelnetConnection):
host_system = socket.gethostname()
if host_system != host:
raise TargetError("Gem5Connection can only connect to gem5 "
"simulations on your current host, which "
"differs from the one given {}!"
.format(host_system, host))
"simulations on your current host {}, which "
"differs from the one given {}!"
.format(host_system, host))
if username is not None and username != 'root':
raise ValueError('User should be root in gem5!')
if password is not None and password != '':
@ -517,8 +522,8 @@ class Gem5Connection(TelnetConnection):
trial = 0
while os.path.isfile(redirection_file):
# Log file already exists so add to name
redirection_file = 'BACKGROUND_{}{}.log'.format(command_name, trial)
trial += 1
redirection_file = 'BACKGROUND_{}{}.log'.format(command_name, trial)
trial += 1
# Create the command to pass on to gem5 shell
complete_command = '{} >> {} 2>&1 &'.format(command, redirection_file)
@ -548,7 +553,7 @@ class Gem5Connection(TelnetConnection):
try:
shutil.rmtree(self.gem5_interact_dir)
except OSError:
gem5_logger.warn("Failed to remove the temporary directory!")
gem5_logger.warning("Failed to remove the temporary directory!")
# Delete the lock file
os.remove(self.lock_file_name)
@ -563,6 +568,7 @@ class Gem5Connection(TelnetConnection):
self.connect_gem5(port, gem5_simulation, gem5_interact_dir, gem5_out_dir)
# Handle the EOF exception raised by pexpect
# pylint: disable=no-self-use
def _gem5_EOF_handler(self, gem5_simulation, gem5_out_dir, err):
# If we have reached the "EOF", it typically means
# that gem5 crashed and closed the connection. Let's
@ -576,6 +582,7 @@ class Gem5Connection(TelnetConnection):
raise err
# This function connects to the gem5 simulation
# pylint: disable=too-many-statements
def connect_gem5(self, port, gem5_simulation, gem5_interact_dir,
gem5_out_dir):
"""
@ -754,7 +761,7 @@ class Gem5Connection(TelnetConnection):
# prompt has returned. Hence, we have a bit of an issue. We
# warn, and return the whole output.
if command_index == -1:
gem5_logger.warn("gem5_shell: Unable to match command in "
gem5_logger.warning("gem5_shell: Unable to match command in "
"command output. Expect parsing errors!")
command_index = 0

1
devlib/utils/uboot.py
View File

@ -113,4 +113,3 @@ class UbootMenu(object):
except TIMEOUT:
pass
self.conn.buffer = ''

2
devlib/utils/uefi.py
View File

@ -237,5 +237,3 @@ class UefiMenu(object):
self.options = {}
self.prompt = None
self.empty_buffer()