blob: 1f2d9fa1c07d68fc4693ce6d728cd4e01a43949e [file] [log] [blame]
# Copyright 2015 The Chromium OS Authors. All rights reserved.
# Use of this source code is governed by a BSD-style license that can be
# found in the LICENSE file.
"""Allows testing and verification of GPS chips on Android DUTs.
GPS devices have a standard text-based output format called NMEA sentences.
This test reads those NMEA sentences for certain values, and checks that they
are within configurable ranges.
Here's an example of input arguments::
'station_name': 'gps_fatp',
'fixture_id': session.GetDeviceID(),
'timeout': 30,
'init_timeout': 30,
'warmup_count': 1,
'gps_config_file': 'gpsconfig_jobs.xml',
'gps_config_job': 'Factory_Track_Test',
'nmea_out_path': '/data/gps/nmea_out',
'nmea_prefix': '$PREFIX,1,',
'nmea_fields': {'signal_strength': 4,
'carrier_to_noise': 7,
'ref_clock_offset': 12},
'limits': [('signal_strength', 'count', '>', 25),
('signal_strength', 'mean', '>', -120),
('ref_clock_offset', 'mean', '>', -3000),
('ref_clock_offset', 'mean', '<', 3000)]})}
from __future__ import print_function
import logging
import os
import re
import subprocess
import sys
import threading
import unittest
import numpy
import factory_common # pylint: disable=unused-import
from cros.factory import device
from cros.factory.device import device_utils
from cros.factory.test import event_log # TODO(chuntsen): Deprecate event log.
from cros.factory.test import session
from cros.factory.testlog import testlog
from cros.factory.utils.arg_utils import Arg
from cros.factory.utils import sync_utils
from cros.factory.utils import time_utils
DEVICE_GPS_CONFIG_PATH = '/data/gps/gpsconfig_jobs.xml'
GLGPS_BINARY = 'glgps'
'count': len,
'min': lambda l: float(numpy.min(l)),
'max': lambda l: float(numpy.max(l)),
'median': lambda l: float(numpy.median(l)),
'mean': lambda l: float(numpy.mean(l)),
'std': lambda l: float(numpy.std(l))}
'<': lambda x, y: x < y,
'<=': lambda x, y: x <= y,
'>': lambda x, y: x > y,
'>=': lambda x, y: x >= y}
class GPS(unittest.TestCase):
ARGS = [
Arg('station_name', str,
'Name of the station. We might want to run the GPS test at '
'different points in the factory. This can be used to identify '
'them, and will be saved in event_logs. e.g. "gps_smt"'),
Arg('fixture_id', str,
'Name of the fixture. This will be saved in event_logs.'),
Arg('init_timeout', (int, float),
'How long to poll for good data before giving up. '
'Default %d seconds.' % DEFAULT_INIT_TIMEOUT,
Arg('timeout', (int, float),
'How long to run the test. Default %d seconds.' % DEFAULT_TIMEOUT,
Arg('warmup_count', int,
'How many initial matching NMEA sentences to ignore before starting '
'to record data.',
Arg('gps_config_file', str,
'Relative or absolute path to GPS configuration file to upload '
'to device. If relative, searches both in directory of this test '
'file, and in directory of currently executing Python script.'),
Arg('gps_config_job', str,
'Name of the job within gps_config_file to run. This will be passed '
'as an argument to glgps to start the job during test execution.'),
Arg('nmea_out_path', str,
'Path to the nmea_out file on the device. This should be a named '
'pipe which is specified in gps_config_file in the <hal> element, '
'like so: <hal NmeaOutName="/data/gps/nmea_out">'),
Arg('nmea_prefix', str,
'Prefix of NMEA sentences for which to filter. Only these sentences '
'will be parsed based on nmea_fields.'),
Arg('nmea_fields', dict,
'Dictionary of fields to pull from the NMEA sentence. '
'Key is a string representing the name of the field, and '
'value is its comma-separated index.'),
Arg('limits', list,
'List of limits, in the format ("nmea_field", "fn", "cmp", value), '
'where fn can be any of %s, and cmp can be any of %s.'
% (STAT_FNS.keys(), CMP_FNS.keys()),
def setUp(self):
self.dut = device_utils.CreateDUTInterface()
# Store the serial numbers for later use.
self._mlb_serial_number = ( or
self._serial_number = ( or SERIAL_NOT_AVAILABLE)'Got MLB serial number: %s', self._mlb_serial_number)'Got serial number: %s', self._serial_number)
# Create a list for test-level failures.
self._failures = []
def _PushConfigFile(self):
# Push the jobs config so we can start the job <self.args.gps_config_job>.'Pushing config file...')
config_path = None
if self.args.gps_config_file.startswith('/'):
if os.path.isfile(self.args.gps_config_file):
config_path = self.args.gps_config_file
# May be in the directory of this file, or the executing Python script.
possible_dirs = [os.path.dirname(os.path.abspath(__file__)),
for possible_dir in possible_dirs:
try_config_path = os.path.join(possible_dir, self.args.gps_config_file)
if os.path.isfile(try_config_path):
config_path = try_config_path
if not config_path:'Config file %s could not be found' % self.args.gps_config_file)
with open(config_path) as f:
def _ParseNMEAStream(self, file_stream):
"""Parse NMEA stream and return values.
Dictionary, where keys are from self.args.nmea_fields, and values
are lists of parsed values in NMEA sentences.
all_values = {key: [] for key in self.args.nmea_fields}
start_time = time_utils.MonotonicTime()
timeout = self.args.init_timeout # Initial timeout to get valid data.
warmup_count = self.args.warmup_count # Initial sentences to ignore.
init = False
warmup_init = False
nmea_line = None
while True:
time_left = timeout - (time_utils.MonotonicTime() - start_time)
if time_left <= 0:
# Make sure we have a useful NMEA line to work with.
nmea_line = file_stream.readline().rstrip()
if not nmea_line.startswith(self.args.nmea_prefix):
logging.debug('[%d] nmea_line: %s', time_left, nmea_line)
# Split the comma-separated data. Use the indices sorted in ascending
# order to look at values deterministically from left-to-right.
nmea_values = nmea_line.split(',')
sorted_indices = sorted(self.args.nmea_fields.values())
index_to_key = {v: k for k, v in self.args.nmea_fields.items()}
values = [nmea_values[index] for index in sorted_indices]
# Initialization.
if not init:
if all([value != '' for value in values]):
# This is the first set of valid values we are getting.
init = True
else:'[%d] Waiting for initialization...', time_left)
# Warmup.
if not warmup_init:
if warmup_count == 0:
# This is the first warmed up value we are getting.
warmup_init = True
start_time = time_utils.MonotonicTime()
time_left = timeout = self.args.timeout
else:'[%d] Warming up...', time_left)
warmup_count -= 1
# Check and record each value.
current_values = {}
for index in sorted_indices:
key = index_to_key[index]
if nmea_values[index] == '':
'Empty value for %s encountered after initialization' % key)
return all_values
parsed_value = float(nmea_values[index])
except ValueError:
self._failures.append('Non-numeric value encountered for %s: %s'
% (key, nmea_values[index]))
return all_values
current_values[key] = parsed_value
# Print parsed values.
current_values_str = ' '.join(
'%s=%.1f' % (key, value)
for key, value in current_values.iteritems())'[%d] %s', time_left, current_values_str)
logging.debug('Timeout has been reached (%d secs)', self.args.timeout)
return all_values
def _CheckLimits(self, all_values):
"""Check limits specified by self.args.limits.
The tuple (field_stats, limit_results, limit_failures_str), where:
field_stats: A dictionary mapping each NMEA field. Values are
dictionaries of STAT_FNS.keys() to its corresponding calculated
limit_results: A dictionary mapping each limit (from self.args.limits)
to a boolean representing its success/failure.
limit_failures_str: A list of strings, where each string represents a
human-readable representation of a limit failure.
# Are there values to calculate stats?
if not all(all_values.values()):
self._failures.append('Never initialized')
field_stats = None
field_stats = {key: {} for key in self.args.nmea_fields}
for key, values in all_values.iteritems():
for fn_name, fn in STAT_FNS.iteritems():
field_stats[key][fn_name] = fn(values)
field_stats_str = ' '.join(
'%s=%.1f' % (k, v)
for k, v in field_stats[key].iteritems())'%s: %s', key, field_stats_str)
# Do limit testing.
limit_results = {}
limit_failures_str = []
for nmea_field, stat_fn, cmp_fn_key, limit_value in self.args.limits:
cmp_fn = CMP_FNS[cmp_fn_key]
if not field_stats:
test_value = None
passed = False
test_value = field_stats[nmea_field][stat_fn]
passed = cmp_fn(test_value, limit_value)
passed_str = 'PASS' if passed else 'FAIL'
limit_str = ('%s.%s %s %.1f'
% (nmea_field, stat_fn, cmp_fn_key, limit_value))
result_str = '%s %s' % (passed_str, limit_str)
limit_results[limit_str] = {'test_value': test_value, 'passed': passed}
if not passed:
logging.debug('Results to be logged: %s', limit_results)
return (field_stats, limit_results, limit_failures_str)
def runTest(self):
# Stop glgps if it's already running.
# Stop gpsd if it's already running.'Stopping gpsd...')
self.dut.Call('stop gpsd')
# Run glgps for <args.timeout> seconds with <self.args.gps_config_job>.'Starting %s job...', self.args.gps_config_job)
def StartGLGPS():
glgps_thread = threading.Thread(target=StartGLGPS)
glgps_thread.daemon = True
# Check that glgps is running and is writing to <self.args.nmea_out_path>.
def CheckGLGPSRunning():
self.dut.CheckCall('ps | grep %s' % GLGPS_BINARY)
self.dut.CheckCall('[[ -n `timeout 1 cat %s` ]]'
% self.args.nmea_out_path)
return True
except device.CalledProcessError:
return False
if not sync_utils.PollForCondition(poll_method=CheckGLGPSRunning,
poll_interval_secs=0):'%s was not running' % GLGPS_BINARY)
# Get the latest readings from the NMEA output file.'Reading from NMEA output file...')
# TODO(kitching): Move this into AdbTarget so we can use something like
# self.dut.Popen() instead of calling adb directly.
cat_process = subprocess.Popen(
['adb', 'shell', 'cat %s' % self.args.nmea_out_path],
all_values = self._ParseNMEAStream(cat_process.stdout)
field_stats, limit_results, limit_failures_str = (
# Log directly to event_log.
# The 'results' value format is:
# {'signal_strength.min > -130.0': {'test_value': -120.0,
# 'passed': True},
# ... }
log_dict = {
'station_name': self.args.station_name,
'fixture_id': self.args.fixture_id,
'mlb_serial_number': self._mlb_serial_number,
'serial_number': self._serial_number,
'passed': not self._failures and not limit_failures_str,
'failures': self._failures,
'stats': field_stats,
'results': limit_results}
event_log.Log(EVENT_LOG_NAME, **log_dict)
testlog.LogParam('stats', field_stats)
testlog.LogParam('results', limit_results)
# Check for failures.
if limit_failures_str:'\n'.join(limit_failures_str))
def tearDown(self):
# Kill glgps.
# Restart normal gpsd operation.'Restarting normal gpsd...')
self.dut.Call('start gpsd')
def _KillGLGPS(self):
# Stop the glgps with Factory_Test_Track.
ps_line = self.dut.CheckOutput('ps | grep %s' % GLGPS_BINARY)
except device.CalledProcessError:
# Process is not running. Don't kill it!'%s already stopped', GLGPS_BINARY)
glgps_pid = re.split(r' *', ps_line)[1]
if not glgps_pid:
# Process is not running. Don't kill it!'%s already stopped', GLGPS_BINARY)
else:'Killing %s pid %d...', GLGPS_BINARY, int(glgps_pid))
self.dut.CheckOutput(['kill', glgps_pid])
# TODO(kitching): Join the GLGPS thread before sending a kill signal?