Source code for wlauto.instrumentation.netstats

import os
import re
import csv
import tempfile
import logging
from datetime import datetime
from collections import defaultdict
from itertools import izip_longest

from wlauto import Instrument, Parameter
from wlauto import ApkFile
from wlauto.exceptions import InstrumentError, HostError
from import ApkInfo
from wlauto.utils.types import list_of_strings

THIS_DIR = os.path.dirname(__file__)

NETSTAT_REGEX = re.compile(r'I[\/ ](?P<tag>netstats-\d+).*?: (?P<ts>\d+) '
                           r'"(?P<package>[^"]+)" TX: (?P<tx>\S+) RX: (?P<rx>\S+)')

[docs]def extract_netstats(filepath, tag=None): netstats = [] with open(filepath) as fh: for line in fh: match = if not match: continue if tag and'tag') != tag: continue netstats.append(('tag'),'ts'),'package'),'tx'),'rx'))) return netstats
[docs]def netstats_to_measurements(netstats): measurements = defaultdict(list) for row in netstats: tag, ts, package, tx, rx = row # pylint: disable=unused-variable measurements[package + '_tx'].append(tx) measurements[package + '_rx'].append(rx) return measurements
[docs]def write_measurements_csv(measurements, filepath): headers = sorted(measurements.keys()) columns = [measurements[h] for h in headers] with open(filepath, 'wb') as wfh: writer = csv.writer(wfh) writer.writerow(headers) writer.writerows(izip_longest(*columns))
[docs]class NetstatsCollector(object): def __init__(self, target, apk, service='.TrafficMetricsService'): """ Additional paramerter: :apk: Path to the APK file that contains ``com.arm.devlab.netstats`` package. If not specified, it will be assumed that an APK with name "netstats.apk" is located in the same directory as the Python module for the instrument. :service: Name of the service to be launched. This service must be present in the APK. """ = target self.apk = apk self.logger = logging.getLogger('netstat') self.package = ApkInfo(self.apk).package self.service = service self.tag = None self.command = None self.stop_command = 'am kill {}'.format(self.package)
[docs] def setup(self, force=False): if if force: self.logger.debug('Re-installing {} (forced)'.format(self.package)), timeout=300) else: self.logger.debug('{} already present on target'.format(self.package)) else: self.logger.debug('Deploying {} to target'.format(self.package))
[docs] def reset(self, sites=None, period=None): period_arg, packages_arg = '', '' self.tag = 'netstats-{}'.format('%Y%m%d%H%M%s')) tag_arg = ' --es tag {}'.format(self.tag) if sites: packages_arg = ' --es packages {}'.format(','.join(sites)) if period: period_arg = ' --ei period {}'.format(period) self.command = 'am startservice{}{}{} {}/{}'.format(tag_arg, period_arg, packages_arg, self.package, self.service) # ensure the service is not running.
[docs] def start(self): if self.command is None: raise RuntimeError('reset() must be called before start()')
[docs] def stop(self):
[docs] def get_data(self, outfile): raw_log_file = tempfile.mktemp() data = extract_netstats(raw_log_file) measurements = netstats_to_measurements(data) write_measurements_csv(measurements, outfile) os.remove(raw_log_file)
[docs] def teardown(self):
[docs]class NetstatsInstrument(Instrument): # pylint: disable=unused-argument name = 'netstats' description = """ Measures transmit/receive network traffic on an Android divice on per-package basis. """ parameters = [ Parameter('packages', kind=list_of_strings, description=""" List of Android packages who's traffic will be monitored. If unspecified, all packages in the device will be monitorred. """), Parameter('period', kind=int, default=5, description=""" Polling period for instrumentation on the device. Traffic statistics will be updated every ``period`` seconds. """), Parameter('force_reinstall', kind=bool, default=False, description=""" If ``True``, instrumentation APK will always be re-installed even if it already installed on the device. """), Parameter('uninstall_on_completion', kind=bool, default=False, global_alias='cleanup', description=""" If ``True``, instrumentation will be uninstalled upon run completion. """), ] def initialize(self, context): if self.device.platform != 'android': raise InstrumentError('nestats instrument is only supported on Android devices.') apk = context.resolver.get(ApkFile(self)) self.collector = NetstatsCollector(self.device, apk) # pylint: disable=attribute-defined-outside-init self.collector.setup(force=self.force_reinstall)
[docs] def setup(self, context): self.collector.reset(sites=self.packages, period=self.period)
[docs] def start(self, context): self.collector.start()
[docs] def stop(self, context): self.collector.stop()
[docs] def update_result(self, context): outfile = os.path.join(context.output_directory, 'netstats.csv') self.collector.get_data(outfile) context.add_artifact('netstats', outfile, kind='data') with open(outfile, 'rb') as fh: reader = csv.reader(fh) metrics = data = [c for c in izip_longest(*list(reader))] for name, values in zip(metrics, data): value = sum(map(int, [v for v in values if v])) context.add_metric(name, value, units='bytes')
def finalize(self, context): if self.uninstall_on_completion: self.collector.teardown()