Source code
Revision control
Copy as Markdown
Other Tools
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
import os
import platform
import re
import subprocess
import six
from .macintelpower import MacIntelPower
from .mozpowerutils import average_summary, frequency_summary, get_logger, sum_summary
OSCPU_COMBOS = {
"darwin-intel": MacIntelPower,
}
SUMMARY_METHODS = {
"utilization": average_summary,
"power-usage": sum_summary,
"power-watts": frequency_summary,
"frequency-cpu": frequency_summary,
"frequency-gpu": frequency_summary,
"default": sum_summary,
}
class OsCpuComboMissingError(Exception):
"""OsCpuComboMissingError is raised when we cannot find
a class responsible for the OS, and CPU combination that
was detected.
"""
pass
class MissingProcessorInfoError(Exception):
"""MissingProcessorInfoError is raised when we cannot find
the processor information on the machine. This is raised when
the file is missing (mentioned in the error message) or if
an exception occurs when we try to gather the information
from the file.
"""
pass
class MozPower(object):
"""MozPower provides an OS and CPU independent interface
for initializing, finalizing, and gathering power measurement
data from OS+CPU combo-dependent measurement classes. The combo
is detected automatically, and the correct class is instantiated
based on the OSCPU_COMBOS list. If it cannot be found, an
OsCpuComboMissingError will be raised.
If a newly added power measurer does not have the required functions
`initialize_power_measurements`, `finalize_power_measurements`,
or `get_perfherder_data`, then a NotImplementedError will be
raised.
Android power measurements are currently not supported by this
module.
::
from mozpower import MozPower
mp = MozPower(output_file_path='dir/power-testing')
mp.initialize_power_measurements()
# Run test...
mp.finalize_power_measurements(test_name='raptor-test-name')
perfherder_data = mp.get_perfherder_data()
"""
def __init__(
self,
android=False,
logger_name="mozpower",
output_file_path="power-testing",
**kwargs
):
"""Initializes the MozPower object, detects OS and CPU (if not android),
and instatiates the appropriate combo-dependent class for measurements.
:param bool android: run android power measurer.
:param str logger_name: logging logger name. Defaults to 'mozpower'.
:param str output_file_path: path to where output files will be stored.
Can include a prefix for the output files, i.e. 'dir/raptor-test'
would output data to 'dir' with the prefix 'raptor-test'.
Defaults to 'power-testing', the current directory using
the prefix 'power-testing'.
:param dict kwargs: optional keyword arguments passed to power measurer.
:raises: * OsCpuComboMissingError
* NotImplementedError
"""
self.measurer = None
self._os = None
self._cpu = None
self._logger = get_logger(logger_name)
if android:
self._logger.error("Android power usage measurer has not been implemented")
raise NotImplementedError
else:
self._os = self._get_os().lower()
cpu = six.text_type(self._get_processor_info().lower())
if "intel" in cpu:
self._cpu = "intel"
else:
self._cpu = "arm64"
# OS+CPU combos are specified through strings such as 'darwin-intel'
# for mac power measurement on intel-based machines. If none exist in
# OSCPU_COMBOS, OsCpuComboMissingError will be raised.
measurer = None
oscpu_combo = "%s-%s" % (self._os, self._cpu)
if oscpu_combo in OSCPU_COMBOS:
measurer = OSCPU_COMBOS[oscpu_combo]
else:
raise OsCpuComboMissingError(
"Cannot find OS+CPU combo for %s" % oscpu_combo
)
if measurer:
self._logger.info(
"Intializing measurer %s for %s power measurements, see below for errors..."
% (measurer.__name__, oscpu_combo)
)
self.measurer = measurer(
logger_name=logger_name, output_file_path=output_file_path, **kwargs
)
def _get_os(self):
"""Returns the operating system of the machine being tested. platform.system()
returns 'darwin' on MacOS, 'windows' on Windows, and 'linux' on Linux systems.
:returns: str
"""
return platform.system()
def _get_processor_info(self):
"""Returns the processor model type of the machine being tested.
Each OS has it's own way of storing this information. Raises
MissingProcessorInfoError if we cannot get the processor info
from the expected locations.
:returns: str
:raises: * MissingProcessorInfoError
"""
model = ""
if self._get_os() == "Windows":
model = platform.processor()
elif self._get_os() == "Darwin":
proc_info_path = "/usr/sbin/sysctl"
command = [proc_info_path, "-n", "machdep.cpu.brand_string"]
if not os.path.exists(proc_info_path):
raise MissingProcessorInfoError(
"Missing processor info file for darwin platform, "
"expecting it here %s" % proc_info_path
)
try:
model = subprocess.check_output(command).strip()
except subprocess.CalledProcessError as e:
error_log = str(e)
if e.output:
error_log = e.output.decode()
raise MissingProcessorInfoError(
"Error while attempting to get darwin processor information "
"from %s (exists) with the command %s: %s"
% (proc_info_path, str(command), error_log)
)
elif self._get_os() == "Linux":
proc_info_path = "/proc/cpuinfo"
model_re = re.compile(r""".*model name\s+[:]\s+(.*)\s+""")
if not os.path.exists(proc_info_path):
raise MissingProcessorInfoError(
"Missing processor info file for linux platform, "
"expecting it here %s" % proc_info_path
)
try:
with open(proc_info_path) as cpuinfo:
for line in cpuinfo:
if not line.strip():
continue
match = model_re.match(line)
if match:
model = match.group(1)
if not model:
raise Exception(
"No 'model name' entries found in the processor info file"
)
except Exception as e:
raise MissingProcessorInfoError(
"Error while attempting to get linux processor information "
"from %s (exists): %s" % (proc_info_path, str(e))
)
return model
def initialize_power_measurements(self, **kwargs):
"""Starts the power measurements by calling the power measurer's
`initialize_power_measurements` function.
:param dict kwargs: keyword arguments for power measurer initialization
function if they are needed.
"""
if self.measurer is None:
return
self.measurer.initialize_power_measurements(**kwargs)
def finalize_power_measurements(self, **kwargs):
"""Stops the power measurements by calling the power measurer's
`finalize_power_measurements` function.
:param dict kwargs: keyword arguments for power measurer finalization
function if they are needed.
"""
if self.measurer is None:
return
self.measurer.finalize_power_measurements(**kwargs)
def get_perfherder_data(self):
"""Returns the partial perfherder data output produced by the measurer.
For a complete perfherder data blob, see get_full_perfherder_data.
:returns: dict
"""
if self.measurer is None:
return
return self.measurer.get_perfherder_data()
def _summarize_values(self, datatype, values):
"""Summarizes the given values based on the type of the
data. See SUMMARY_METHODS for the methods used for each
known data type. Defaults to using the sum of the values
when a data type cannot be found.
Data type entries in SUMMARY_METHODS are case-sensitive.
:param str datastype: the measurement type being summarized.
:param list values: the values to summarize.
:returns: float
"""
if datatype not in SUMMARY_METHODS:
self._logger.warning(
"Missing summary method for data type %s, defaulting to sum" % datatype
)
datatype = "default"
summary_func = SUMMARY_METHODS[datatype]
return summary_func(values)
def get_full_perfherder_data(
self, framework, lowerisbetter=True, alertthreshold=2.0
):
"""Returns a list of complete perfherder data blobs compiled from the
partial perfherder data blob returned from the measurer. Each key entry
(measurement type) in the partial perfherder data is parsed into its
own suite within a single perfherder data blob.
For example, a partial perfherder data blob such as:
::
{
'utilization': {<perfherder_data>},
'power-usage': {<perfherder_data>}
}
would produce two suites within a single perfherder data blobs -
one for utilization, and one for power-usage.
Note that the 'values' entry must exist, otherwise the measurement
type is skipped. Furthermore, if 'name', 'unit', or 'type' is missing
we default to:
::
{
'name': 'mozpower',
'unit': 'mWh',
'type': 'power'
}
Subtests produced for each sub-suite (measurement type), have the naming
pattern: <measurement_type>-<measured_name>
Utilization of cpu would have the following name: 'utilization-cpu'
Power-usage for cpu has the following name: 'power-usage-cpu'
:param str framework: name of the framework being tested, i.e. 'raptor'.
:param bool lowerisbetter: if set to true, low values are better than high ones.
:param float alertthreshold: determines the crossing threshold at
which an alert is generated.
:returns: dict
"""
if self.measurer is None:
return
# Get the partial data, and the measurers name for
# logging purposes.
partial_perfherder_data = self.get_perfherder_data()
measurer_name = self.measurer.__class__.__name__
suites = []
perfherder_data = {"framework": {"name": framework}, "suites": suites}
for measurement_type in partial_perfherder_data:
self._logger.info("Summarizing %s data" % measurement_type)
dataset = partial_perfherder_data[measurement_type]
# Skip this measurement type if the 'values' entry
# doesn't exist, and output a warning.
if "values" not in dataset:
self._logger.warning(
"Missing 'values' entry in partial perfherder data for measurement type %s "
"obtained from %s. This measurement type will not be processed."
% (measurement_type, measurer_name)
)
continue
# Get the settings, if they exist, otherwise output
# a warning and use a default entry.
settings = {"test": "mozpower", "unit": "mWh", "type": "power"}
for setting in settings:
if setting in dataset:
settings[setting] = dataset[setting]
else:
self._logger.warning(
"Missing '%s' entry in partial perfherder data for measurement type %s "
"obtained from %s, using %s as the default"
% (setting, measurement_type, measurer_name, settings[setting])
)
subtests = []
suite = {
"name": "%s-%s" % (settings["test"], measurement_type),
"type": settings["type"],
"value": 0,
"subtests": subtests,
"lowerIsBetter": lowerisbetter,
"unit": settings["unit"],
"alertThreshold": alertthreshold,
}
# Parse the 'values' entries into subtests
values = []
for measure in dataset["values"]:
value = dataset["values"][measure]
subtest = {
"name": "%s-%s" % (measurement_type, measure),
"value": float(value),
"lowerIsBetter": lowerisbetter,
"alertThreshold": alertthreshold,
"unit": settings["unit"],
}
values.append((value, measure))
subtests.append(subtest)
# Summarize the data based on the measurement type
if len(values) > 0:
suite["value"] = self._summarize_values(measurement_type, values)
suites.append(suite)
return perfherder_data