blob: 5a26c7bc5dd9369079030c86d94a232f72eec51e [file] [log] [blame]
# Copyright 2012 The Chromium Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
from __future__ import division
import logging as real_logging
import os
import subprocess
import sys
import time
import six
from telemetry.core import local_server
from telemetry.core import memory_cache_http_server
from telemetry.core import network_controller
from telemetry.core import tracing_controller
from telemetry.core import util
from telemetry.internal.platform import (platform_backend as
from py_utils import discover
# Remote platform is a dictionary from device ids to remote platform instances.
def _InitHostPlatformIfNeeded():
global _HOST_PLATFORM # pylint: disable=global-statement
backend = None
backends = _IterAllPlatformBackendClasses()
for platform_backend_class in backends:
if platform_backend_class.IsPlatformBackendForHost():
backend = platform_backend_class()
if not backend:
raise NotImplementedError()
_HOST_PLATFORM = Platform(backend)
def GetHostPlatform():
def _IterAllPlatformBackendClasses():
platform_dir = os.path.dirname(os.path.realpath(
return six.itervalues(discover.DiscoverClasses(
platform_dir, util.GetTelemetryDir(),
def GetPlatformForDevice(device, finder_options, logging=real_logging):
""" Returns a platform instance for the device.
device: a device.Device instance.
if device.guid in _REMOTE_PLATFORMS:
return _REMOTE_PLATFORMS[device.guid]
for platform_backend_class in _IterAllPlatformBackendClasses():
if platform_backend_class.SupportsDevice(device):
_REMOTE_PLATFORMS[device.guid] = (
return _REMOTE_PLATFORMS[device.guid]
return None
except Exception:
current_exception = sys.exc_info()
logging.error('Fail to create platform instance for %s.',
raise current_exception[0], current_exception[1], current_exception[2]
class Platform(object):
"""The platform that the target browser is running on.
Provides a limited interface to interact with the platform itself, where
possible. It's important to note that platforms may not provide a specific
API, so check with IsFooBar() for availability.
def __init__(self, platform_backend):
self._platform_backend = platform_backend
self._network_controller = network_controller.NetworkController(
self._tracing_controller = tracing_controller.TracingController(
self._local_server_controller = local_server.LocalServerController(
self._forwarder = None
def is_host_platform(self):
return self == GetHostPlatform()
def network_controller(self):
"""Control network settings and servers to simulate the Web."""
return self._network_controller
def tracing_controller(self):
return self._tracing_controller
def Initialize(self):
def CanMonitorThermalThrottling(self):
"""Platforms may be able to detect thermal throttling.
Some fan-less computers go into a reduced performance mode when their heat
exceeds a certain threshold. Performance tests in particular should use this
API to detect if this has happened and interpret results accordingly.
return self._platform_backend.CanMonitorThermalThrottling()
def GetSystemLog(self):
return self._platform_backend.GetSystemLog()
def IsThermallyThrottled(self):
"""Returns True if the device is currently thermally throttled."""
return self._platform_backend.IsThermallyThrottled()
def HasBeenThermallyThrottled(self):
"""Returns True if the device has been thermally throttled."""
return self._platform_backend.HasBeenThermallyThrottled()
def GetDeviceTypeName(self):
"""Returns a string description of the Platform device, or None.
Examples: Nexus 7, Nexus 6, Desktop"""
return self._platform_backend.GetDeviceTypeName()
def GetArchName(self):
"""Returns a string description of the Platform architecture.
Examples: x86_64 (posix), AMD64 (win), armeabi-v7a, x86"""
return self._platform_backend.GetArchName()
def GetOSName(self):
"""Returns a string description of the Platform OS.
return self._platform_backend.GetOSName()
def GetDeviceId(self):
"""Returns a string identifying the device.
Examples: 0123456789abcdef"""
return self._platform_backend.GetDeviceId()
def GetOSVersionName(self):
"""Returns a logically sortable, string-like description of the Platform OS
return self._platform_backend.GetOSVersionName()
def GetOSVersionDetailString(self):
"""Returns more detailed information about the OS version than
GetOSVersionName, if available. Otherwise returns the empty string.
Examples: '10.12.4' on macOS."""
return self._platform_backend.GetOSVersionDetailString()
def GetSystemTotalPhysicalMemory(self):
"""Returns an integer with the total physical memory in bytes."""
return self._platform_backend.GetSystemTotalPhysicalMemory()
def CanFlushIndividualFilesFromSystemCache(self):
"""Returns true if the disk cache can be flushed for individual files."""
return self._platform_backend.CanFlushIndividualFilesFromSystemCache()
def SupportFlushEntireSystemCache(self):
"""Returns true if entire system cache can be flushed.
Also checks that platform has required privilegues to flush system caches.
return self._platform_backend.SupportFlushEntireSystemCache()
def _WaitForPageCacheToBeDropped(self):
# There seems to be no reliable way to wait for all pages to be dropped from
# the OS page cache (also known as 'file cache'). There is no guaranteed
# moment in time when everything is out of page cache. A number of pages
# will likely be reused before other pages are evicted. While individual
# files can be watched in limited ways, we choose not to be clever.
def FlushEntireSystemCache(self):
"""Flushes the OS's file cache completely.
This function may require root or administrator access. Clients should
call SupportFlushEntireSystemCache to check first.
def FlushSystemCacheForDirectories(self, directories):
"""Flushes the OS's file cache for the specified directory.
This function does not require root or administrator access."""
for path in directories:
def FlushDnsCache(self):
"""Flushes the OS's DNS cache completely.
This function may require root or administrator access."""
return self._platform_backend.FlushDnsCache()
def LaunchApplication(self,
""""Launches the given |application| with a list of |parameters| on the OS.
Set |elevate_privilege| to launch the application with root or admin rights.
A popen style process handle for host platforms.
return self._platform_backend.LaunchApplication(
def StartActivity(self, intent, blocking=False):
"""Starts an activity for the given intent on the device."""
return self._platform_backend.StartActivity(intent, blocking)
def CanLaunchApplication(self, application):
"""Returns whether the platform can launch the given application."""
return self._platform_backend.CanLaunchApplication(application)
def InstallApplication(self, application, **kwargs):
"""Installs the given application."""
return self._platform_backend.InstallApplication(application, **kwargs)
def IsCooperativeShutdownSupported(self):
"""Indicates whether CooperativelyShutdown, below, is supported.
It is not necessary to implement it on all platforms."""
return self._platform_backend.IsCooperativeShutdownSupported()
def CooperativelyShutdown(self, proc, app_name):
"""Cooperatively shut down the given process from subprocess.Popen.
Currently this is only implemented on Windows. See for background on why it was added.
proc: a process object returned from subprocess.Popen.
app_name: on Windows, is the prefix of the application's window
class name that should be searched for. This helps ensure
that only the application's windows are closed.
Returns True if it is believed the attempt succeeded.
return self._platform_backend.CooperativelyShutdown(proc, app_name)
def CanTakeScreenshot(self):
return self._platform_backend.CanTakeScreenshot()
# TODO( Implement this on Mac, Linux & Win.
def TakeScreenshot(self, file_path):
""" Takes a screenshot of the platform and save to |file_path|.
Note that this method may not be supported on all platform, so check with
CanTakeScreenshot before calling this.
file_path: Where to save the screenshot to. If the platform is remote,
|file_path| is the path on the host platform.
Returns True if it is believed the attempt succeeded.
return self._platform_backend.TakeScreenshot(file_path)
def CanRecordVideo(self):
return self._platform_backend.CanRecordVideo()
def StartVideoRecording(self):
"""Starts recording a video on the device.
Note that this method may not be supported on all platforms, so the caller
must check with CanRecordVideo before calling this. Once the caller starts
recording a video using this call, the caller must stop recording the video
by calling StopVideoRecording() before attempting to start recording another
def StopVideoRecording(self, video_path):
"""Stops recording a video on the device and saves to |video_path|.
This method must be called only if recording a video had started using a
call to StartVideoRecording(), and it was not already stopped using a call
to StopVideoRecording().
video_path: Where to save the video to. If the platform is remote,
|video_path| is the path on the host platform.
def SetPerformanceMode(self, performance_mode):
""" Set the performance mode on the platform.
Note: this can be no-op on certain platforms.
return self._platform_backend.SetPerformanceMode(performance_mode)
def StartLocalServer(self, server):
"""Starts a LocalServer and associates it with this platform.
|server.Close()| should be called manually to close the started server.
def http_server(self):
# TODO( Ownership of the local server should be moved
# to the network_controller.
server = self._local_server_controller.GetRunningServer(
memory_cache_http_server.MemoryCacheDynamicHTTPServer, None)
if server:
return server
return self._local_server_controller.GetRunningServer(
memory_cache_http_server.MemoryCacheHTTPServer, None)
def SetHTTPServerDirectories(self, paths, handler_class=None):
"""Returns True if the HTTP server was started, False otherwise."""
# pylint: disable=redefined-variable-type
if isinstance(paths, six.string_types):
paths = {paths}
paths = set(os.path.realpath(p) for p in paths)
# If any path is in a subdirectory of another, remove the subdirectory.
duplicates = set()
for parent_path in paths:
for sub_path in paths:
if parent_path == sub_path:
if os.path.commonprefix((parent_path, sub_path)) == parent_path:
paths -= duplicates
if self.http_server:
old_handler_class = getattr(self.http_server,
"dynamic_request_handler_class", None)
if not old_handler_class and not handler_class and \
self.http_server.paths == paths:
return False
if old_handler_class and handler_class \
and old_handler_class.__name__ == handler_class.__name__ \
and self.http_server.paths == paths:
return False
if not paths:
return False
if handler_class:
server = memory_cache_http_server.MemoryCacheDynamicHTTPServer(
paths, handler_class)'MemoryCacheDynamicHTTPServer created')
server = memory_cache_http_server.MemoryCacheHTTPServer(paths)'MemoryCacheHTTPServer created')
# For now, Fuchsia needs to do port forwarding due to --proxy-server
# flag not being supported in its browser.
# TODO( Remove once debug flags supported in
# Fuchsia browsers.
if self._platform_backend.GetOSName() == 'fuchsia':
self._platform_backend.forwarder_factory.Create(server.port, server.port)
return True
def StopAllLocalServers(self):
if self._forwarder:
def local_servers(self):
"""Returns the currently running local servers."""
return self._local_server_controller.local_servers
def WaitForBatteryTemperature(self, temp):
"""Waits for the battery on the device under test to cool down to temp.
temp: temperature target in degrees C.
return self._platform_backend.WaitForBatteryTemperature(temp)
def WaitForCpuTemperature(self, temp):
"""Waits for the CPU temperature to be less than temp.
temp: A float containing the maximum temperature to allow
in degrees c.
return self._platform_backend.WaitForCpuTemperature(temp)
def GetTypExpectationsTags(self):
return self._platform_backend.GetTypExpectationsTags()
def SupportsIntelPowerGadget(self):
"""Check if Intel Power Gadget is supported.
Intel Power Gadget is supported on Intel based desktop platforms. Currently
this function only checks if Intel Power Gadget is installed on the device.
The assumption is if it is installed, it can run successfully.
return self._platform_backend.SupportsIntelPowerGadget()
def RunIntelPowerGadget(self, duration, output_path):
"""Runs Intel Power Gadget and collects power data for a period of time.
Note that Intel Power Gadget will stop after |duration|, output the
collected data, and then exit. However, this function returns right away
after Intel Power Gadget is launched and starts collection power data.
Also, makes sure SupportsIntelPowerGadget() returns True before calling
this function.
duration: Seconds that Intel Power Gadget runs and collects power data.
output_path: Specifies to which file Intel Power Gadget writes the
collected data.
Returns True if Intel Power Gadget starts successfully.
ipg_path = self._platform_backend.GetIntelPowerGadgetPath()
if not ipg_path:
real_logging.error('Fail to locate Intel Power Gadget. Please call '
'SupportsIntelPowerGadget() before calling this.')
return False
command = '"%s" -duration %d -file "%s"' % (ipg_path, duration, output_path)
subprocess.Popen(command, shell=True, stderr=subprocess.STDOUT)
return True
def CollectIntelPowerGadgetResults(self, output_path, skip_duration=0,
"""Processes power data output from Intel Power Gadget.
Note that the file |output_path| is removed when this function returns.
output_path: Specifies to which file Intel Power Gadget wrote the
collected data.
skip_duration: Seconds of data to skip in the beginning of the collected
data, assuming there are extra noises in the beginning.
sample_duration: Seconds of data to process and return after the
|skip_duration|. If this is 0, process data to the end.
A dictionary in the format of {measurement_name: average_value}.
Specifically, 'samples' indicates the number of samples that contributed
to the returned data.
If an error occurs, an empty disctionary is returned.
if not os.path.isfile(output_path):
real_logging.error('Cannot locate output file at ' + output_path)
return {}
first_line = True
samples = 0
cols = 0
indices = []
labels = []
sums = []
col_time = None
for line in open(output_path):
tokens = [token.strip('" ') for token in line.split(',')]
if first_line:
first_line = False
cols = len(tokens)
for ii in range(0, cols):
token = tokens[ii]
if token.startswith('Elapsed Time'):
col_time = ii
elif token.endswith('(Watt)'):
assert col_time
assert cols > 0
assert len(indices) > 0
if len(tokens) != cols:
if skip_duration > 0 and float(tokens[col_time]) < skip_duration:
if (sample_duration > 0 and
float(tokens[col_time]) > skip_duration + sample_duration):
samples += 1
for ii in range(0, len(indices)):
index = indices[ii]
sums[ii] += float(tokens[index])
results = {'samples': samples}
if samples > 0:
for ii in range(0, len(indices)):
#2To3-division: this line is unchanged as sums[] are floats.
results[labels[ii]] = sums[ii] / samples
return results