| # -*- coding: utf-8; tab-width: 4; python-indent: 4 -*- |
| # Copyright (c) 2012 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. |
| |
| # Import guard for OpenCV. |
| try: |
| import cv |
| import cv2 |
| except ImportError: |
| pass |
| |
| import base64 |
| import logging |
| import numpy as np |
| import os |
| import pprint |
| import pyudev |
| import re |
| import select |
| import serial |
| import signal |
| import socket |
| import StringIO |
| import subprocess |
| import sys |
| import tempfile |
| import threading |
| import time |
| import xmlrpclib |
| |
| import autotest_lib.client.cros.camera.perf_tester as camperf |
| import autotest_lib.client.cros.camera.renderer as renderer |
| |
| from autotest_lib.client.bin import test |
| from autotest_lib.client.common_lib import error |
| from autotest_lib.client.cros import factory_setup_modules #pylint:disable=W0611 |
| from cros.factory.event_log import Log, GetDeviceId, TimedUuid |
| from cros.factory.test import factory |
| from cros.factory.test import leds |
| from cros.factory.test import shopfloor |
| from cros.factory.test import test_ui |
| from cros.factory.test import utils |
| from cros.factory.test.media_util import MountedMedia |
| from cros.factory.utils import net_utils, file_utils |
| from cros.factory.utils.process_utils import SpawnOutput |
| from autotest_lib.client.cros.rf.config import PluggableConfig |
| from autotest_lib.client.cros import tty |
| from cros.factory.test.test_ui import UI |
| |
| |
| # Test type constants: |
| _TEST_TYPE_MODULE = 'Module' |
| _TEST_TYPE_AB = 'AB' |
| _TEST_TYPE_FULL = 'Full' |
| |
| # Content type constants: |
| _CONTENT_IMG = 'image' |
| _CONTENT_TXT = 'text' |
| |
| _INSERT_ETHERNET_DONGLE_TIMEOUT_SECS = 30 # Timeout for inserting dongle. |
| _IP_SETUP_TIMEOUT_SECS = 10 # Timeout for setting IP address. |
| _SHOPFLOOR_TIMEOUT_SECS = 10 # Timeout for shopfloor connection. |
| _SHOPFLOOR_RETRY_INTERVAL_SECS = 10 # Seconds to wait between retries. |
| |
| |
| class ShopfloorBridgeException(Exception): |
| pass |
| |
| |
| class ShopfloorBridge(object): |
| '''Warps shopfloor bridge connection.''' |
| # Uses very simple protocol in shopfloor bridge. |
| # <4-char ID><space><8-char data length>\r\n<variable length of data> |
| _PROT_PORT = 8000 # Default port for shopfloor bridge. |
| # Hello! |
| _PROT_COMMAND_HELLO = 'HELO' |
| # download test parameters |
| _PROT_COMMAND_PARA = 'PARA' |
| # uploads test results |
| _PROT_COMMAND_TEST = 'TEST' |
| _PROT_COMMAND_SIZE = 4 |
| _PROT_STATUS_OK = 'OKAY' |
| _PROT_STATUS_FAIL = 'FAIL' |
| _PROT_DATALEN_SIZE = 8 |
| _PROT_PREAMBLE_SIZE = _PROT_COMMAND_SIZE + 1 + _PROT_DATALEN_SIZE + 2 |
| # Seconds to wait for request / connection. (The value should be large |
| # enough to avoid unnecessary retrying.) |
| _TIMEOUT_SECS = 20 |
| # Seconds to wait between retries. |
| _RETRY_INTERVAL_SECS = 3 |
| |
| def __init__(self, bridge_ip, setup_network_callback): |
| '''ShopfloorBridge is usually used in module-level testing. |
| |
| Reconfigure the IP with setup_network_callback since the test may run |
| for a very long time. Otherwise, if IP setting is lost, |
| self._reconnect() may get stuck in infinite loop. |
| |
| Args: |
| bridge_ip: IP address of the shopfloor bridge server. |
| setup_network_callback: Callback to setup network IP. |
| ''' |
| assert(bridge_ip) |
| self.bridge_ip = bridge_ip |
| self.setup_network_callback = setup_network_callback |
| self.socket = None |
| self.rfile = None |
| self.wfile = None |
| |
| def __del__(self): |
| self._close() |
| |
| def _close(self): |
| """Closes the connection to shopfloor bridge.""" |
| if self.rfile: |
| self.rfile.close() |
| self.rfile = None |
| if self.wfile: |
| self.wfile.close() |
| self.wfile = None |
| if self.socket: |
| self.socket.close() |
| self.socket = None |
| |
| def _reconnect(self): |
| """Reconnects to shopfloor bridge.""" |
| self._close(); |
| self._connect() |
| |
| def _connect(self, |
| timeout_secs=_TIMEOUT_SECS, |
| retry_interval_secs=_RETRY_INTERVAL_SECS): |
| """Connects to shopfloor bridge. |
| |
| Args: |
| timeout_secs: Seconds to wait for connection. |
| retry_interval_secs: Seconds to wait between retries. |
| """ |
| if self.socket: # already connected |
| return |
| |
| factory.console.info('Connecting to shopfloor bridge...') |
| while True: |
| try: |
| self.socket = socket.socket(socket.AF_INET, |
| socket.SOCK_STREAM) |
| self.socket.settimeout(timeout_secs) |
| self.socket.connect((self.bridge_ip, self._PROT_PORT)) |
| self.socket.settimeout(None) |
| break |
| except: # pylint: disable=W0702 |
| exception_string = utils.FormatExceptionOnly() |
| # Log the exception string only since it may happen repeatedly. |
| factory.console.error('Unable to connect to bridge: %s', |
| exception_string) |
| if self.setup_network_callback: |
| self.setup_network_callback() |
| |
| time.sleep(retry_interval_secs) |
| |
| self.socket.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1) |
| self.rfile = self.socket.makefile('rb', -1) |
| self.wfile = self.socket.makefile('wb', 0) # No buffering |
| |
| def _send_command(self, command, data, |
| timeout_secs=_TIMEOUT_SECS, |
| retry_interval_secs=_RETRY_INTERVAL_SECS): |
| """Sends command to shopfloor bridge. |
| |
| Args: |
| command: Command to bridge. |
| data: Data to bridge. |
| timeout_secs: Seconds to wait for response. |
| retry_interval_secs: Seconds to wait between retries. |
| |
| Returns: |
| A pair of (response okay, received data). |
| """ |
| received_data = '' |
| self._connect() |
| |
| data_len = len(data) |
| assert len(command) == 4 |
| assert len('%d' % data_len) <= 8 |
| preamble = '%s %08d\r\n' % (command, data_len) |
| assert len(preamble) == self._PROT_PREAMBLE_SIZE |
| |
| response_ok = None |
| while response_ok is None: |
| try: |
| with utils.Timeout(timeout_secs): |
| self.wfile.write(preamble) |
| if data: |
| self.wfile.write(data) |
| self.wfile.flush() |
| ack_preamble = self.rfile.read(self._PROT_PREAMBLE_SIZE) |
| if len(ack_preamble) != self._PROT_PREAMBLE_SIZE: |
| raise ShopfloorBridgeException( |
| "Shopfloor Error: can't read %d bytes (preamble)." % |
| self._PROT_PREAMBLE_SIZE) |
| if ack_preamble[-2] != '\r' or ack_preamble[-1] != '\n': |
| raise ShopfloorBridgeException( |
| 'Shopfloor Error: cannot find \\r or \\n in reply.') |
| ack_status = ack_preamble[0:self._PROT_COMMAND_SIZE] # 0:4 |
| ack_datalen_str = ack_preamble[ |
| self._PROT_COMMAND_SIZE + 1: |
| self._PROT_COMMAND_SIZE + 1 + |
| self._PROT_DATALEN_SIZE] # 5:13 |
| factory.console.info( |
| 'Received "%s" "%s" from shopfloor bridge', ack_status, |
| ack_datalen_str) |
| ack_datalen = int(ack_datalen_str) |
| if ack_datalen > 0: |
| received_data = self.rfile.read(ack_datalen) |
| if len(received_data) != ack_datalen: |
| raise ShopfloorBridgeException( |
| "Shopfloor Error: can't read %d bytes (data)." % |
| ack_datalen) |
| |
| if ack_status != self._PROT_STATUS_OK: |
| # On fail, received_data contains error message. End the |
| # loop here because shopfloor may not recover from the error |
| # anytime soon. We should make the test fails instead of |
| # retrying to talk to shopfloor again. |
| factory.console.error( |
| 'Shopfloor Error: bridge returns %s: %s', |
| ack_status, received_data) |
| response_ok = False |
| else: |
| response_ok = True |
| except: # pylint: disable=W0702 |
| exception_string = utils.FormatExceptionOnly() |
| # Log the exception string only since it may happen repeatedly. |
| factory.console.error('Unable to get response from command %s ' |
| 'from shopfloor bridge: %s', |
| command, exception_string) |
| |
| time.sleep(retry_interval_secs) |
| # Assume it's network problem. Simply reconnects the bridge! |
| self._reconnect() |
| |
| return (response_ok, received_data) |
| |
| def download_param(self, param_filename): |
| """Downloads parameter. |
| |
| Returns: |
| Returns downloaded parameters. |
| """ |
| |
| response_ok, param_data = self._send_command(self._PROT_COMMAND_PARA, |
| param_filename) |
| # No way to handle the error case shopfloor returns FAIL in downloading |
| # parameters. |
| assert response_ok, 'Cannot download parameter from shopfloor' |
| return param_data |
| |
| def upload_result(self, results): |
| """Uploads test results. |
| |
| Returns: |
| Success or not. |
| """ |
| response_ok, dummy = self._send_command(self._PROT_COMMAND_TEST, |
| results) |
| return response_ok |
| |
| |
| class _ShopfloorWrapper(object): |
| '''Wraps shopfloor connectivity.''' |
| |
| def __init__(self, ip_addr, bridge_mode, bridge_ip, |
| shopfloor_directory, shopfloor_param_file): |
| ''' |
| Args: |
| ip_addr: IP address of DUT (None for DHCP). |
| bridge_mode: Use bridge mode or native shopfloor. |
| bridge_ip: IP of the bridge server. |
| shopfloor_directory: Relative path to shopfloor parameters folder. |
| shopfloor_param_file: Filename of parameter file on shopfloor. |
| ''' |
| self.ip_addr = ip_addr |
| if bridge_mode: |
| self.bridge = ShopfloorBridge( |
| bridge_ip=bridge_ip, |
| setup_network_callback=lambda: self._prepare_network(False)) |
| else: |
| self.bridge = None |
| self.shopfloor_directory = shopfloor_directory |
| self.shopfloor_param_file = shopfloor_param_file |
| |
| def _prepare_network(self, force_new_ip): |
| '''Blocks forever until network is prepared. |
| |
| Args: |
| force_new_ip: always set new IP regardless of existing IP. |
| ''' |
| def _obtain_IP(): |
| if self.ip_addr is None: |
| net_utils.SendDhcpRequest() |
| else: |
| net_utils.SetEthernetIp(self.ip_addr, force=force_new_ip) |
| return True if net_utils.GetEthernetIp() else False |
| |
| factory.console.info('Detecting Ethernet device...') |
| try: |
| net_utils.PollForCondition(condition=( |
| lambda: True if net_utils.FindUsableEthDevice() else False), |
| timeout=_INSERT_ETHERNET_DONGLE_TIMEOUT_SECS, |
| condition_name='Detect Ethernet device') |
| |
| # Only setup the IP if required so. |
| current_ip = net_utils.GetEthernetIp( |
| net_utils.FindUsableEthDevice()) |
| if not current_ip or force_new_ip: |
| factory.console.info('Setting up IP address...') |
| net_utils.PollForCondition( |
| condition=_obtain_IP, |
| timeout=_IP_SETUP_TIMEOUT_SECS, |
| condition_name='Setup IP address') |
| except: # pylint: disable=W0702 |
| exception_string = utils.FormatExceptionOnly() |
| factory.console.error('Unable to setup network: %s', |
| exception_string) |
| |
| factory.console.info('Network prepared. IP: %r', |
| net_utils.GetEthernetIp()) |
| |
| def _get_shopfloor_connection(self, |
| timeout_secs=_SHOPFLOOR_TIMEOUT_SECS, |
| retry_interval_secs=_SHOPFLOOR_RETRY_INTERVAL_SECS): |
| """Returns a shopfloor client object. |
| |
| Try forever until a connection of shopfloor is established. |
| |
| Args: |
| timeout_secs: Timeout for shopfloor connection. |
| retry_interval_secs: Seconds to wait between retries. |
| """ |
| factory.console.info('Connecting to shopfloor...') |
| while True: |
| try: |
| shopfloor_client = shopfloor.get_instance( |
| detect=True, timeout=timeout_secs) |
| break |
| except: # pylint: disable=W0702 |
| exception_string = utils.FormatExceptionOnly() |
| # Log the exception string only since it may happen repeatedly. |
| factory.console.error('Unable to sync with shopfloor: %s', |
| exception_string) |
| time.sleep(retry_interval_secs) |
| return shopfloor_client |
| |
| def download_param(self, force_new_ip): |
| '''Read parameters from shopfloor server. |
| |
| Args: |
| force_new_ip: Always set new IP regardless of existing IP. |
| ''' |
| self._prepare_network(force_new_ip) |
| param_path = os.path.join(self.shopfloor_directory, |
| self.shopfloor_param_file) |
| factory.console.info('Read %s from shopfloor', param_path) |
| if self.bridge: |
| return self.bridge.download_param(param_path) |
| else: |
| sf = self._get_shopfloor_connection() |
| return sf.GetParameter(param_path).data |
| |
| def upload_log(self, force_new_ip, aux_log_list): |
| '''Uploads logs to shopfloor server. |
| |
| Args: |
| force_new_ip: Always set new IP regardless of existing IP. |
| aux_log_list: List of tuples of (log_name, log_data). |
| |
| Returns: |
| Returns False if there is critical issue from shopfloor. |
| ''' |
| self._prepare_network(force_new_ip) |
| if not self.bridge: |
| sf = self._get_shopfloor_connection() |
| |
| for log_name, log_data in aux_log_list: |
| factory.console.info('Uploading %s', log_name) |
| start_time = time.time() |
| try: |
| if self.bridge: |
| # Shopfloor bridge behaves differently than our built-in |
| # shopfloor server. |
| # |
| # 1. Failing to upload to built-in shopfloor is either a bug |
| # or wrong network setup. It should never happen. Raise |
| # exception. |
| # |
| # 2. Failing to upload to shopfloor bridge (usually connects |
| # to a proprietary shopfloor server of module house) may be |
| # caused by factory flow issue. For example, DUT is tested |
| # at a wrong stage. This is a critical error and we should |
| # mark the test 'failed' for retesting DUT later. |
| assert re.search(r'\.txt$', log_name), ( |
| 'Only supports uploading of .txt files.') |
| if not self.bridge.upload_result(log_data): |
| return False |
| else: |
| sf.SaveAuxLog(log_name, xmlrpclib.Binary(log_data)) |
| except: # pylint: disable=W0702 |
| exception_string = utils.FormatExceptionOnly() |
| factory.console.error('Failed to upload %s: %s', |
| log_name, exception_string) |
| raise |
| factory.console.info('Successfully synced %s in %.03f s', |
| log_name, time.time() - start_time) |
| return True |
| |
| |
| class ALS(): |
| '''Class to interface the ambient light sensor over iio.''' |
| |
| # Default device paths. |
| _VAL_DEV_PATH = '/sys/bus/iio/devices/iio:device0/illuminance0_input' |
| _SCALE_DEV_PATH = '/sys/bus/iio/devices/iio:device0/illuminance0_calibscale' |
| |
| # Default min delay seconds. |
| _DEFAULT_MIN_DELAY = 0.178 |
| |
| def __init__(self, val_path=_VAL_DEV_PATH, |
| scale_path=_SCALE_DEV_PATH): |
| self.detected = True |
| if (not os.path.isfile(val_path) or |
| not os.path.isfile(scale_path)): |
| self.detected = False |
| return |
| self.val_path = val_path |
| self.scale_path = scale_path |
| |
| def _read_core(self): |
| fd = open(self.val_path) |
| val = int(fd.readline().rstrip()) |
| fd.close() |
| return val |
| |
| def _read(self, delay=None, samples=1): |
| '''Read the light sensor value. |
| |
| Args: |
| delay: Delay between samples in seconds. 0 means as fast as |
| possible. |
| samples: Total samples to read. |
| |
| Returns: |
| The light sensor values in a list. |
| ''' |
| if samples < 1: |
| samples = 1 |
| if delay is None: |
| delay = self._DEFAULT_MIN_DELAY |
| |
| buf = [] |
| # The first value might be contaminated by previous settings. |
| # We need to skip it for better accuracy. |
| self._read_core() |
| for dummy in range(samples): |
| time.sleep(delay) |
| val = self._read_core() |
| buf.append(val) |
| |
| return buf |
| |
| def read_mean(self, delay=None, samples=1): |
| if not self.detected: |
| return None |
| |
| buf = self._read(delay, samples) |
| return int(round(float(sum(buf)) / len(buf))) |
| |
| def set_scale_factor(self, scale): |
| if not self.detected: |
| return None |
| |
| fd = open(self.scale_path, 'w') |
| fd.write(str(int(round(scale)))) |
| fd.close() |
| return |
| |
| def get_scale_factor(self): |
| if not self.detected: |
| return None |
| |
| fd = open(self.scale_path) |
| s = int(fd.readline().rstrip()) |
| fd.close() |
| return s |
| |
| |
| class FixtureException(Exception): |
| pass |
| |
| |
| class Fixture(): |
| '''Class for communication with the test fixture.''' |
| |
| def __init__(self, params): |
| # Setup the serial port communication. |
| tty_path = tty.find_tty_by_driver(params['driver']) |
| self.fixture = serial.Serial(port=tty_path, |
| **params['serial_params']) |
| self.fixture.flush() |
| |
| # Load parameters. |
| self.serial_delay = params['serial_delay'] |
| self.light_delay = params['light_delay'] |
| self.light_seq = params['light_seq'] |
| self.fixture_echo = params['echo'] |
| self.light_off = params['off'] |
| |
| def send(self, msg): |
| '''Send control messages to the fixture.''' |
| for c in msg: |
| self.fixture.write(str(c)) |
| self.fixture.flush() |
| # The fixture needs some time to process each incoming character. |
| time.sleep(self.serial_delay) |
| |
| def read(self): |
| return self.fixture.read(self.fixture.inWaiting()) |
| |
| def assert_success(self): |
| '''Check if the returned value from the fixture is OK.''' |
| ret = self.read() |
| if not re.search(self.fixture_echo, ret): |
| raise FixtureException('The communication with fixture was broken') |
| |
| def set_light(self, idx): |
| self.send(self.light_seq[idx]) |
| |
| def turn_off_light(self): |
| self.send(self.light_off) |
| |
| def wait_for_light_switch(self): |
| time.sleep(self.light_delay) |
| |
| |
| class ConnectionMonitor(): |
| """A wrapper to monitor hardware plug/unplug events.""" |
| def __init__(self): |
| self._monitoring = False |
| |
| def start(self, subsystem, device_type=None, on_insert=None, |
| on_remove=None): |
| if self._monitoring: |
| raise Exception("Multiple start() call is not allowed") |
| self.on_insert = on_insert |
| self.on_remove = on_remove |
| |
| # Setup the media monitor, |
| context = pyudev.Context() |
| self.monitor = pyudev.Monitor.from_netlink(context) |
| self.monitor.filter_by(subsystem, device_type) |
| self.monitor.start() |
| self._monitoring = True |
| self._watch_thread = threading.Thread(target=self.watch) |
| self._watch_end = threading.Event() |
| self._watch_thread.start() |
| |
| def watch(self): |
| fd = self.monitor.fileno() |
| while not self._watch_end.isSet(): |
| ret, _, _ = select.select([fd],[],[]) |
| if fd in ret: |
| action, dev = self.monitor.receive_device() |
| if action == 'add' and self.on_insert: |
| self.on_insert(dev.device_node) |
| elif action == 'remove' and self.on_remove: |
| self.on_remove(dev.device_node) |
| |
| def stop(self): |
| self._monitoring = False |
| self._watch_end.set() |
| |
| |
| class factory_CameraPerformanceAls(test.test): |
| version = 3 |
| preserve_srcdir = True |
| |
| _BAD_SERIAL_NUMBER = 'BAD_SN' |
| _NO_SERIAL_NUMBER = 'NO_SN' |
| |
| _PACKET_SIZE = 65000 |
| |
| # Status in the final result tab. |
| _STATUS_NAMES = ['cam_stat', 'cam_vc', 'cam_ls', 'cam_mtf', |
| 'als_stat', 'result'] |
| _STATUS_LABELS = ['Camera Functionality', |
| 'Camera Visual Correctness', |
| 'Camera Lens Shading', |
| 'Camera Image Sharpness', |
| 'ALS Functionality', |
| 'Test Result'] |
| _CAM_TESTS = ['cam_stat', 'cam_vc', 'cam_ls', 'cam_mtf'] |
| _ALS_TESTS = ['als_stat'] |
| |
| # LED patterns. |
| _LED_RUNNING_TEST = ((leds.LED_NUM|leds.LED_CAP, 0.05), (0, 0.05)) |
| |
| # CSS style classes defined in the corresponding HTML file. |
| _STYLE_INFO = "color_idle" |
| _STYLE_PASS = "color_good" |
| _STYLE_FAIL = "color_bad" |
| |
| def t_pass(self, msg): |
| return test_ui.MakeLabel(msg, css_class=self._STYLE_PASS) |
| |
| def t_fail(self, msg): |
| return test_ui.MakeLabel(msg, css_class=self._STYLE_FAIL) |
| |
| def update_status(self, mid=None, msg=None): |
| message = '' |
| if msg: |
| message = msg |
| elif mid: |
| message = test_ui.MakeLabel(self.config['message'][mid + '_en'], |
| self.config['message'][mid + '_zh'], |
| self.config['msg_style'][mid]) |
| self.ui.CallJSFunction("UpdateTestStatus", message) |
| |
| def update_pbar(self, pid=None, value=None, add=True): |
| precent = 0 |
| if value: |
| percent = value |
| elif pid: |
| all_time = self.config['chk_point'][self.type] |
| if add: |
| self.progress += self.config['chk_point'][pid] |
| else: |
| self.progress = self.config['chk_point'][pid] |
| percent = int(round((float(self.progress) / all_time) * 100)) |
| self.ui.CallJSFunction("UpdatePrograssBar", '%d%%' % percent) |
| |
| def register_events(self, events): |
| for event in events: |
| assert hasattr(self, event) |
| self.ui.AddEventHandler(event, getattr(self, event)) |
| |
| def get_test_chart_file(self): |
| return 'test_chart_%s.png' % self.test_chart_version |
| |
| def get_test_sample_file(self): |
| return 'sample_%s.png' % self.test_chart_version |
| |
| def prepare_test(self): |
| self.ref_data = camperf.PrepareTest(self.get_test_chart_file()) |
| |
| def download_param_from_shopfloor(self, event): |
| """Download test parameters from shopfloor server. |
| |
| This is callback from Javascript in order to run download_param() in |
| main thread because utils.Timeout only supports running in main thread. |
| """ |
| assert self.shopfloor |
| param = self.shopfloor.download_param(force_new_ip=True) |
| |
| self.prepare_test() |
| self.config = eval(param) |
| self.reset_data() |
| self.config_loaded = True |
| factory.console.info("Config loaded from shopfloor.") |
| self.ui.CallJSFunction("OnShopfloorInit", self.config['sn_format']) |
| |
| def on_usb_insert(self, dev_path): |
| if not self.config_loaded: |
| # Initialize common test reference data. |
| self.prepare_test() |
| # Load config files and reset test results. |
| self.dev_path = dev_path |
| with MountedMedia(dev_path, 1) as config_dir: |
| config_path = os.path.join(config_dir, 'camera.params') |
| self.config = self.base_config.Read(config_path) |
| self.reset_data() |
| self.config_loaded = True |
| factory.console.info("Config loaded.") |
| self.ui.CallJSFunction("OnUSBInit", self.config['sn_format']) |
| else: |
| self.dev_path = dev_path |
| self.ui.CallJSFunction("OnUSBInsertion") |
| |
| def on_usb_remove(self, dev_path): |
| if self.config_loaded: |
| factory.console.info("USB removal is not allowed during test!") |
| self.ui.CallJSFunction("OnUSBRemoval") |
| |
| def setup_fixture(self): |
| '''Initialize the communication with the fixture.''' |
| try: |
| self.fixture = Fixture(self.config['fixture']) |
| |
| # Go with the default(first) lighting intensity. |
| self.light_state = 0 |
| self.fixture.set_light(self.light_state) |
| if not self.unit_test: |
| self.fixture.assert_success() |
| except Exception as e: |
| self.fixture = None |
| self.log('Failed to initialize the test fixture.\n') |
| return False |
| self.log('Test fixture successfully initialized.\n') |
| return True |
| |
| def sync_fixture(self, event): |
| self.ui.CallJSFunction("OnDetectFixtureConnection") |
| cnt = 0 |
| while not self.setup_fixture(): |
| cnt += 1 |
| if cnt >= self.config['fixture']['n_retry']: |
| self.ui.CallJSFunction("OnRemoveFixtureConnection") |
| return |
| time.sleep(self.config['fixture']['retry_delay']) |
| self.ui.CallJSFunction("OnAddFixtureConnection") |
| |
| def on_u2s_insert(self, dev_path): |
| if self.config_loaded: |
| self.sync_fixture(None) |
| |
| def on_u2s_remove(self, dev_path): |
| if self.config_loaded: |
| self.ui.CallJSFunction("OnRemoveFixtureConnection") |
| |
| def update_result(self, row_name, result): |
| result_map = { |
| True: 'PASSED', |
| False: 'FAILED', |
| None: 'UNTESTED' |
| } |
| self.result_dict[row_name] = result_map[result] |
| |
| def update_fail_cause(self, fail_cause): |
| self.fail_cause = fail_cause |
| |
| def reset_data(self): |
| self.target = None |
| self.target_colorful = None |
| self.analyzed = None |
| self.log_to_file = StringIO.StringIO() |
| self.log = lambda *x: (factory.console.info(*x), |
| self.log_to_file.write(*x)) |
| |
| for var in self.status_names: |
| self.update_result(var, None) |
| self.fail_cause = '' |
| self.progress = 0 |
| self.ui.CallJSFunction("ResetUiData", "") |
| |
| def send_img_to_ui(self, data): |
| self.ui.CallJSFunction("ClearBuffer", "") |
| # Send the data in 64K packets due to the socket packet size limit. |
| data_len = len(data) |
| p = 0 |
| while p < data_len: |
| if p + self._PACKET_SIZE > data_len: |
| self.ui.CallJSFunction("AddBuffer", data[p:data_len-1]) |
| p = data_len |
| else: |
| self.ui.CallJSFunction("AddBuffer", |
| data[p:p+self._PACKET_SIZE]) |
| p += self._PACKET_SIZE |
| |
| def update_preview(self, img, container_id, scale=0.5): |
| # Encode the image in the JPEG format. |
| preview = cv2.resize(img, None, fx=scale, fy=scale, |
| interpolation=cv2.INTER_AREA) |
| cv2.imwrite('temp.jpg', preview) |
| with open('temp.jpg', 'r') as fd: |
| img_data = base64.b64encode(fd.read()) + "=" |
| |
| # Update the preview screen with javascript. |
| self.send_img_to_ui(img_data) |
| self.ui.CallJSFunction("UpdateImage", container_id) |
| return |
| |
| def compile_result(self, test_list, use_untest=True): |
| ret = self.result_dict |
| if all('PASSED' == ret[x] for x in test_list): |
| return True |
| if use_untest and any('UNTESTED' == ret[x] for x in test_list): |
| return None |
| return False |
| |
| def generate_final_result(self): |
| self.update_status(mid='end_test') |
| self.cam_pass = self.compile_result(self._CAM_TESTS) |
| if self.use_als: |
| self.als_pass = self.compile_result(self._ALS_TESTS) |
| result = self.compile_result(self.status_names[:-1], |
| use_untest=False) |
| else: |
| result = self.compile_result(self._CAM_TESTS, use_untest=False) |
| self.update_result('result', result) |
| self.log("Result in summary:\n%s\n" % |
| pprint.pformat(self.result_dict)) |
| Log('cam_performance_test_result', **self.result_dict) |
| Log('cam_performance_fail_cause', cam_fail_cause=self.fail_cause) |
| self.update_pbar(pid='end_test') |
| |
| def write_to_usb(self, filename, content, content_type=_CONTENT_TXT): |
| try: |
| with MountedMedia(self.dev_path, 1) as mount_dir: |
| if content_type == _CONTENT_TXT: |
| # Appends the text log. |
| with open(os.path.join(mount_dir, filename), 'a') as f: |
| f.write(content) |
| elif content_type == _CONTENT_IMG: |
| cv2.imwrite(os.path.join(mount_dir, filename), content) |
| except: |
| self.log("Error when writing data to USB!\n") |
| return False |
| return True |
| |
| def save_log_to_usb(self): |
| # Save an image for further analysis |
| self.update_status(mid='save_to_usb') |
| if ((self.target is not None) and |
| ((self.cam_pass and self.log_good_image) or |
| (not self.cam_pass and self.log_bad_image))): |
| if not self.write_to_usb(self.serial_number + ".bmp", |
| self.target, _CONTENT_IMG): |
| return False |
| if self.analyzed is not None: |
| if not self.write_to_usb(self.serial_number + ".result.jpg", |
| self.analyzed, _CONTENT_IMG): |
| return False |
| return self.write_to_usb( |
| self.serial_number + ".txt", self.log_to_file.getvalue()) |
| |
| def save_log_to_shopfloor(self): |
| ''' Updates image and aux log to shopfloor server.''' |
| self.update_status(mid='save_to_shopfloor') |
| |
| aux_log_list = [] |
| log_prefix = '_'.join([re.sub(r'\W+', '_', x) for x in |
| [os.environ.get('CROS_FACTORY_TEST_PATH'), |
| re.sub(r':', '', GetDeviceId()), |
| self.serial_number]]) |
| |
| aux_log_list.append((log_prefix + '.txt', self.log_to_file.getvalue())) |
| |
| # Adds images to aux logs. |
| if ((self.target is not None) and |
| ((self.cam_pass and self.log_good_image) or |
| (not self.cam_pass and self.log_bad_image))): |
| # TODO (jchuang): newer version of OpenCV has better imencode() |
| # Python method. |
| def get_image_data(cv_image, file_ext): |
| temp_fn = os.path.join(tempfile.gettempdir(), |
| TimedUuid() + file_ext) |
| try: |
| cv2.imwrite(temp_fn, cv_image) |
| with open(temp_fn, 'rb') as f: |
| return f.read() |
| except: # pylint: disable=W0702 |
| factory.console.error('Failed to read image data') |
| finally: |
| file_utils.TryUnlink(temp_fn) |
| |
| aux_log_list.extend([(log_prefix + '.bmp', |
| get_image_data(self.target, '.bmp')), |
| (log_prefix + '.jpg', |
| get_image_data(self.analyzed, '.jpg'))]) |
| |
| # If shopfloor.upload_log() returns FALSE only on critical error and |
| # we need to indicate error for retesting later. |
| if not self.shopfloor.upload_log(False, aux_log_list): |
| self.cam_pass = False |
| self.update_fail_cause("Shopfloor") |
| |
| def finalize_test(self): |
| self.generate_final_result() |
| if self.shopfloor: |
| self.save_log_to_shopfloor() |
| self.update_pbar(pid='save_to_shopfloor') |
| elif self.type in [_TEST_TYPE_AB, _TEST_TYPE_MODULE]: |
| # We block the test flow until we successfully dumped the result. |
| while not self.save_log_to_usb(): |
| time.sleep(0.5) |
| self.update_pbar(pid='save_to_usb') |
| |
| # Display final result. |
| def get_str(ret, prefix, use_untest=True): |
| if ret: |
| return self.t_pass(prefix + 'PASS') |
| if use_untest and (ret is None): |
| return self.t_fail(prefix + 'UNFINISHED') |
| return self.t_fail(prefix + 'FAIL') |
| cam_result = get_str(self.cam_pass, 'Camera: ', False) |
| if self.use_als: |
| als_result = get_str(self.als_pass, 'ALS: ') |
| self.update_status(msg=cam_result + ' ' + self.fail_cause + |
| '<br>' + als_result) |
| else: |
| self.update_status(msg=cam_result + ' ' + self.fail_cause) |
| |
| # Reset serial number if passed. Otherwise, operator may forget to input |
| # serial number again. |
| if self.cam_pass and self.type in [_TEST_TYPE_AB, _TEST_TYPE_MODULE]: |
| self.ui.CallJSFunction("RestartSnInputBox") |
| |
| self.update_pbar(value=100) |
| |
| def exit_test(self, event): |
| factory.log('%s run_once finished' % self.__class__) |
| if self.result_dict['result'] == 'PASSED': |
| self.ui.Pass() |
| else: |
| self.ui.Fail('Camera/ALS test failed.') |
| |
| def run_test(self, event=None): |
| self.reset_data() |
| self.update_status(mid='start_test') |
| |
| if self.talk_to_fixture and not self.setup_fixture(): |
| self.update_status(mid='fixture_fail') |
| self.ui.CallJSFunction("OnRemoveFixtureConnection") |
| return |
| self.update_pbar(pid='start_test') |
| |
| self.log('Parameter version: %s\n' % self.config['version']) |
| |
| if self.auto_serial_number: |
| # If fails to get serial number, it will display failed in |
| # test_camera_functionality() later |
| ret, auto_sn, error_message = self.auto_get_serial_number() |
| if ret: |
| self.serial_number = auto_sn |
| self.log('Read serial number: %s\n' % auto_sn) |
| else: |
| self.serial_number = self._BAD_SERIAL_NUMBER |
| self.log('No serial number detected: %s\n' % error_message) |
| elif self.type in [_TEST_TYPE_AB, _TEST_TYPE_MODULE]: |
| self.serial_number = event.data.get('sn', '') |
| else: |
| self.serial_number = self._NO_SERIAL_NUMBER |
| |
| if self.type == _TEST_TYPE_FULL: |
| with leds.Blinker(self._LED_RUNNING_TEST): |
| self.test_camera_performance() |
| self.update_pbar(pid='cam_finish', add=False) |
| if self.use_als: |
| self.test_als_calibration() |
| self.update_pbar(pid='als_finish' + self.type, add=False) |
| else: |
| self.test_camera_performance() |
| self.update_pbar(pid='cam_finish', add=False) |
| if self.use_als: |
| self.test_als_calibration() |
| self.update_pbar(pid='als_finish' + self.type, add=False) |
| |
| self.finalize_test() |
| |
| def capture_low_noise_image(self, cam, n_samples): |
| '''Capture a sequence of images and average them to reduce noise.''' |
| if n_samples < 1: |
| n_samples = 1 |
| success, img = cam.read() |
| if not success: |
| return None |
| img = img.astype(np.float64) |
| for t in range(n_samples - 1): |
| success, temp_img = cam.read() |
| if not success: |
| return None |
| img += temp_img.astype(np.float64) |
| img /= n_samples |
| return img.round().astype(np.uint8) |
| |
| def read_usb_attribute(self, device_string, pattern): |
| '''Read and matches regexp pattern in 'lsusb -v' output. |
| |
| Args: |
| device_string: keyword to search for the camera device |
| pattern: regexp pattern with one matching group in MULTILINE mode |
| |
| Returns: |
| A tuple of (is successful, matched string, error message when failed). |
| ''' |
| lsusb_output = SpawnOutput(['lsusb', '-v'], log=True) |
| |
| # Split into several blocks of individual USB device |
| splitter = re.compile(r'^Bus\s+\d+\s+Device\s+\d+', |
| flags = re.MULTILINE) |
| blocks = splitter.split(lsusb_output) |
| matched_blocks = [b for b in blocks if device_string in b] |
| |
| err_message = None |
| ret = False |
| matched_string = None |
| if len(matched_blocks) == 0: |
| err_message = 'No matched camera device for "%s"' % device_string |
| elif len(matched_blocks) > 1: |
| err_message = ('Multiple matched devices found for "%s"' % |
| device_string) |
| else: |
| # Camera module should publish firmware version in 'bcdDevice' field |
| search_result = re.search(pattern, |
| matched_blocks[0], flags = re.MULTILINE) |
| if search_result: |
| matched_string = search_result.group(1) |
| ret = True |
| else: |
| err_message = ( |
| 'Regexp "%s" not matched in lsusb output for "%s"' % |
| (pattern, device_string)) |
| |
| return (ret, matched_string, err_message) |
| |
| def verify_firmware(self, device_string, firmware_string): |
| '''Checks the firmware version of camera module. |
| |
| Args: |
| device_string: keyword to search for the camera device |
| firmware_string: expected firmware string in bcdDevice field |
| |
| Returns: |
| A tuple of (is successful, error message when failed). |
| ''' |
| ret, matched_string, err_message = self.read_usb_attribute( |
| device_string, |
| r'^\s*bcdDevice\s+(\S+)') |
| |
| if ret: |
| if firmware_string != matched_string: |
| err_message = ('Wrong firmware version %s, expecting %s' % |
| (matched_string, firmware_string)) |
| ret = False |
| |
| return (ret, err_message) |
| |
| def auto_get_serial_number(self): |
| '''Auto read the firmware version of camera module. |
| |
| Returns: |
| A tuple of (is successful, matched string, error message when failed). |
| ''' |
| if not self.auto_serial_number: |
| return (False, None, 'auto_serial_number is not defined') |
| ret, matched_string, err_message = self.read_usb_attribute( |
| self.auto_serial_number[0], |
| self.auto_serial_number[1]) |
| return (ret, matched_string, err_message) |
| |
| def test_camera_functionality(self): |
| # Check serial number |
| if self.serial_number == self._BAD_SERIAL_NUMBER: |
| self.update_result('cam_stat', False) |
| self.update_fail_cause("NoCamera") |
| return False |
| |
| # Check firmware version |
| if self.config['firmware']['check_firmware']: |
| device_string = self.config['firmware']['device_string'] |
| firmware_string = self.config['firmware']['firmware_string'] |
| ret, error_message = self.verify_firmware(device_string, |
| firmware_string) |
| if not ret: |
| self.update_result('cam_stat', False) |
| self.update_fail_cause("NoCamera") |
| self.log(error_message + '\n') |
| return False |
| |
| # Initialize the camera with OpenCV. |
| self.update_status(mid='init_cam') |
| cam = cv2.VideoCapture(self.device_index) |
| if not cam.isOpened(): |
| cam.release() |
| self.update_result('cam_stat', False) |
| self.update_fail_cause("BadCamera") |
| self.log('Failed to initialize the camera. ' |
| 'Could be bad module, bad connection or ' |
| 'bad USB initialization.\n') |
| return False |
| self.update_pbar(pid='init_cam') |
| |
| # Set resolution. |
| self.update_status(mid='set_cam_res') |
| conf = self.config['cam_stat'] |
| cam.set(cv.CV_CAP_PROP_FRAME_WIDTH, conf['img_width']) |
| cam.set(cv.CV_CAP_PROP_FRAME_HEIGHT, conf['img_height']) |
| if (conf['img_width'] != cam.get(cv.CV_CAP_PROP_FRAME_WIDTH) or |
| conf['img_height'] != cam.get(cv.CV_CAP_PROP_FRAME_HEIGHT)): |
| cam.release() |
| self.update_result('cam_stat', False) |
| self.update_fail_cause("BadCamera") |
| self.log("Can't set the image size. " |
| "Possibly caused by bad USB initialization.\n") |
| return False |
| self.update_pbar(pid='set_cam_res') |
| |
| # Try reading an image from the camera. |
| self.update_status(mid='try_read_cam') |
| success, _ = cam.read() |
| if not success: |
| cam.release() |
| self.update_result('cam_stat', False) |
| self.update_fail_cause("BadCamera") |
| self.log("Failed to capture an image with the camera.\n") |
| return False |
| self.update_pbar(pid='try_read_cam') |
| |
| # Let the camera's auto-exposure algorithm adjust to the fixture |
| # lighting condition. |
| self.update_status(mid='wait_cam_awb') |
| start = time.time() |
| while time.time() - start < conf['buf_time']: |
| _, _ = cam.read() |
| self.update_pbar(pid='wait_cam_awb') |
| |
| # Read the image that we will use. |
| self.update_status(mid='record_img') |
| n_samples = conf['n_samples'] |
| self.target_colorful = self.capture_low_noise_image(cam, n_samples) |
| if self.target_colorful is None: |
| cam.release() |
| self.update_result('cam_stat', False) |
| self.update_fail_cause("BadCamera") |
| self.log("Error reading images from the camera!\n") |
| return False |
| if self.unit_test: |
| self.target_colorful = cv2.imread(self.get_test_sample_file()) |
| |
| self.target = cv2.cvtColor(self.target_colorful, cv.CV_BGR2GRAY) |
| self.update_result('cam_stat', True) |
| self.log('Successfully captured a sample image.\n') |
| self.update_preview(self.target_colorful, "camera_image", |
| scale=self.config['preview']['scale']) |
| cam.release() |
| self.update_pbar(pid='record_img') |
| return True |
| |
| def test_camera_performance(self): |
| if not self.test_camera_functionality(): |
| return |
| |
| # Export log to both cros.factory.event_log and text log |
| visual_data = {} |
| visual_data['camera_sn'] = self.serial_number |
| |
| def log_visual_data(value, event_key, log_text_fmt): |
| self.log((log_text_fmt % value) + '\n') |
| visual_data[event_key] = value |
| |
| def finish_log_visual_data(): |
| Log('cam_performance_visual_analysis', **visual_data) |
| |
| # Check the captured test pattern image validity. |
| self.update_status(mid='check_vc') |
| |
| success, tar_data = camperf.CheckVisualCorrectness( |
| self.target, self.ref_data, **self.config['cam_vc']) |
| self.analyzed = self.target_colorful.copy() |
| renderer.DrawVC(self.analyzed, success, tar_data) |
| self.update_preview(self.analyzed, "analyzed_image", |
| scale=self.config['preview']['scale']) |
| |
| self.update_result('cam_vc', success) |
| if hasattr(tar_data, 'shift'): |
| log_visual_data(float(tar_data.shift), 'image_shift', |
| 'Image shift percentage: %f') |
| log_visual_data(float(tar_data.v_shift[0]), 'image_shift_x', |
| 'Image shift X: %f') |
| log_visual_data(float(tar_data.v_shift[1]), 'image_shift_y', |
| 'Image shift Y: %f') |
| log_visual_data(float(tar_data.tilt), 'image_tilt', |
| 'Image tilt: %f degrees') |
| if not success: |
| if hasattr(tar_data, 'sample_corners'): |
| log_visual_data(int(tar_data.sample_corners.shape[0]), |
| 'corners', 'Found corners count: %d') |
| if hasattr(tar_data, 'edges'): |
| log_visual_data(int(tar_data.edges.shape[0]), 'edges', |
| 'Found square edges count: %d') |
| |
| if 'shift' in tar_data.msg: |
| self.update_fail_cause('Shift') |
| elif 'tilt' in tar_data.msg: |
| self.update_fail_cause('Tilt') |
| else: |
| self.update_fail_cause('WrongImage') |
| |
| self.log('Visual correctness: %s\n' % tar_data.msg) |
| finish_log_visual_data() |
| return |
| self.update_pbar(pid='check_vc') |
| |
| # Check if the lens shading is present. |
| self.update_status(mid='check_ls') |
| success, tar_ls = camperf.CheckLensShading( |
| self.target, **self.config['cam_ls']) |
| |
| self.update_result('cam_ls', success) |
| if tar_ls.check_low_freq: |
| log_visual_data(float(tar_ls.response), 'ls_low_freq', |
| 'Low-frequency response value: %f') |
| if tar_ls.lowest_ratio: |
| log_visual_data(float(tar_ls.lowest_ratio), 'ls_lowest_ratio', |
| 'Len shading ratio: %f') |
| if not success: |
| self.log('Lens shading: %s\n' % tar_ls.msg) |
| self.update_fail_cause('LenShading') |
| finish_log_visual_data() |
| return |
| self.update_pbar(pid='check_ls') |
| |
| # Check the image sharpness. |
| self.update_status(mid='check_mtf') |
| success, tar_mtf = camperf.CheckSharpness( |
| self.target, tar_data.edges, **self.config['cam_mtf']) |
| renderer.DrawMTF(self.analyzed, tar_data.edges, tar_mtf.perm, |
| tar_mtf.mtfs, |
| self.config['cam_mtf']['mtf_crop_ratio'], |
| self.config['preview']['mtf_color_map_range']) |
| self.update_preview(self.analyzed, "analyzed_image", |
| scale=self.config['preview']['scale']) |
| |
| self.update_result('cam_mtf', success) |
| log_visual_data(float(tar_mtf.mtf), 'median_MTF', 'MTF value: %f') |
| if hasattr(tar_mtf, 'min_mtf'): |
| log_visual_data(float(tar_mtf.min_mtf), 'lowest_MTF', |
| 'Lowest MTF value: %f') |
| if not success: |
| self.log('Sharpness: %s\n' % tar_mtf.msg) |
| self.update_fail_cause('MTF') |
| |
| finish_log_visual_data() |
| self.update_pbar(pid='check_mtf') |
| return |
| |
| def test_als_write_vpd(self, calib_result): |
| self.update_status(mid='dump_to_vpd') |
| conf = self.config['als'] |
| if not calib_result: |
| self.update_result('als_stat', False) |
| self.update_fail_cause("ALS") |
| self.log('ALS calibration data is incorrect.\n') |
| return False |
| if subprocess.call(conf['save_vpd'] % calib_result, shell=True): |
| self.update_result('als_stat', False) |
| self.update_fail_cause("ALS") |
| self.log('Writing VPD data failed!\n') |
| return False |
| self.log('Successfully calibrated ALS scales.\n') |
| self.update_pbar(pid='dump_to_vpd') |
| return True |
| |
| def test_als_switch_to_next_light(self): |
| self.update_status(mid='adjust_light') |
| conf = self.config['als'] |
| self.light_state += 1 |
| self.fixture.set_light(self.light_state) |
| self.update_pbar(pid='adjust_light') |
| if not self.unit_test: |
| self.fixture.assert_success() |
| if self.light_state >= len(conf['luxs']): |
| return False |
| self.update_status(mid='wait_fixture') |
| self.fixture.wait_for_light_switch() |
| self.update_pbar(pid='wait_fixture') |
| return True |
| |
| def test_als_calibration(self): |
| # Initialize the ALS. |
| self.update_status(mid='init_als') |
| conf = self.config['als'] |
| self.als = ALS(val_path=conf['val_path'], |
| scale_path=conf['scale_path']) |
| if not self.als.detected: |
| self.update_result('als_stat', False) |
| self.update_fail_cause("ALS") |
| self.log('Failed to initialize the ALS.\n') |
| return |
| self.als.set_scale_factor(conf['calibscale']) |
| self.update_pbar(pid='init_als') |
| |
| # Go through all different lighting settings |
| # and record ALS values. |
| calib_result = 0 |
| try: |
| vals = [] |
| while True: |
| # Get ALS values. |
| self.update_status(mid='read_als%d' % self.light_state) |
| scale = self.als.get_scale_factor() |
| val = self.als.read_mean(samples=conf['n_samples'], |
| delay=conf['read_delay']) |
| vals.append(val) |
| self.log('Lighting preset lux value: %d\n' % |
| conf['luxs'][self.light_state]) |
| self.log('ALS value: %d\n' % val) |
| self.log('ALS calibration scale: %d\n' % scale) |
| # Check if it is a false read. |
| if not val: |
| self.update_result('als_stat', False) |
| self.update_fail_cause("ALS") |
| self.log('The ALS value is stuck at zero.\n') |
| return |
| # Compute calibration data if it is the calibration target. |
| if conf['luxs'][self.light_state] == conf['calib_lux']: |
| calib_result = int(round(float(conf['calib_target']) / |
| val * scale)) |
| self.log('ALS calibration data will be %d\n' % |
| calib_result) |
| self.update_pbar(pid='read_als%d' % self.light_state) |
| |
| # Go to the next lighting preset. |
| if not self.test_als_switch_to_next_light(): |
| break |
| |
| # Check value ordering. |
| for i, li in enumerate(conf['luxs']): |
| for j in range(i): |
| if ((li > conf['luxs'][j] and vals[j] >= vals[i]) or |
| (li < conf['luxs'][j] and vals[j] <= vals[i])): |
| self.update_result('als_stat', False) |
| self.update_fail_cause("ALS") |
| self.log('The ordering of ALS values is wrong.\n') |
| return |
| except (FixtureException, serial.serialutil.SerialException) as e: |
| self.fixture = None |
| self.update_result('als_stat', None) |
| self.log("The test fixture was disconnected!\n") |
| self.ui.CallJSFunction("OnRemoveFixtureConnection") |
| return |
| except: |
| self.update_result('als_stat', False) |
| self.update_fail_cause("ALS") |
| self.log('Failed to read values from ALS or unknown error.\n') |
| return |
| self.log('Successfully recorded ALS values.\n') |
| |
| # Save ALS values to vpd for FATP test. |
| if self.type == _TEST_TYPE_FULL: |
| if not self.test_als_write_vpd(calib_result): |
| return |
| self.update_result('als_stat', True) |
| return |
| |
| def run_once(self, test_type = _TEST_TYPE_FULL, unit_test = False, |
| use_als = True, test_chart_version = 'A', |
| log_good_image = False, log_bad_image = True, |
| data_method = 'usb', ip_addr = None, bridge_ip_addr = None, |
| shopfloor_directory = None, shopfloor_param_file = None, |
| device_index = -1, ignore_enter_key = False, |
| auto_serial_number = None): |
| '''The entry point of the test. |
| |
| Args: |
| test_type: Run the full machine test or the AB panel test. The AB |
| panel will be run on a host that is used to test |
| connected AB panels (possibly many), while the full |
| machine test would test only the machine that runs it |
| and then exit. |
| unit_test: Run the unit-test mode. The unit-test mode is used to |
| test the test integrity when the test fixture is not |
| available. It should be run on a machine that has a |
| working camera and a working ALS. Please place the |
| camera parameter file under the src directory on an USB |
| stick for use and connect the machine with an |
| USB-to-RS232 converter cable with the designated chipset |
| in the parameter file. The test will replace the |
| captured image with the sample test image and run the |
| camera performance test on it. |
| use_als: Whether to use the ambient light sensor. |
| test_chart_version: Version of the test chart. |
| log_good_image: Log images that pass that test. |
| log_bad_image: Log images that fail that test. |
| data_method: How to read parameter and save results. |
| ('usb', 'shopfloor', 'bridge') |
| ip_addr: Network setting when data_method is shopfloor or bridge. |
| If ip_addr is None, use DHCP. |
| shopfloor_directory: Relative path to shopfloor parameters folder. |
| shopfloor_param_file: Filename of camera parameter file on shopfloor |
| server. |
| bridge_ip_addr: IP addr of shopfloor bridge. |
| device_index: video device index (-1 to auto pick device by OpenCV). |
| ignore_enter_key: disable enter key in serial number input. |
| (Some barcode reader automatically input enter |
| key, but we may prefer not to start the test |
| immediately after barcode is scanned.) |
| auto_serial_number: None or (module keyword, regexp pattern with one |
| matching group in MULTILINE mode) |
| It support all Module, AB, and Full test types. |
| Ex: ('VendorName', r'^\s*iSerial\s+\S+\s+(\S+)') |
| ''' |
| factory.log('%s run_once' % self.__class__) |
| |
| # Add signal handler to close opened camera interface when get killed |
| # TODO: this should be done in autotest framework instead |
| signal_handler = lambda signum, frame: sys.exit(1) |
| signal.signal(signal.SIGTERM, signal_handler) |
| signal.signal(signal.SIGINT, signal_handler) |
| |
| # Set logging level. Otherwise, update_preview() will log the whole |
| # image data to factory log under the default log level of autotest. |
| logging.getLogger().setLevel(logging.INFO) |
| |
| # Initialize variables and environment. |
| assert test_type in [_TEST_TYPE_FULL, _TEST_TYPE_AB, _TEST_TYPE_MODULE] |
| assert unit_test in [True, False] |
| assert use_als in [True, False] |
| assert test_chart_version in ['A', 'B'] |
| assert log_good_image in [True, False] |
| assert log_bad_image in [True, False] |
| assert data_method in ['usb', 'shopfloor', 'bridge'] |
| assert shopfloor_directory is None or type(shopfloor_directory) == str |
| assert shopfloor_param_file is None or type(shopfloor_param_file) == str |
| assert bridge_ip_addr is None or type(bridge_ip_addr) == str |
| assert ignore_enter_key in [True, False] |
| assert auto_serial_number is None or type(auto_serial_number) == tuple |
| self.type = test_type |
| self.unit_test = unit_test |
| self.use_als = use_als |
| self.test_chart_version = test_chart_version |
| self.log_good_image = log_good_image |
| self.log_bad_image = log_bad_image |
| if data_method in ('shopfloor', 'bridge'): |
| if data_method == 'bridge': |
| bridge_mode = True |
| else: |
| bridge_mode = False |
| self.shopfloor = _ShopfloorWrapper( |
| ip_addr=ip_addr, |
| bridge_mode=bridge_mode, |
| bridge_ip=bridge_ip_addr, |
| shopfloor_directory=shopfloor_directory, |
| shopfloor_param_file=shopfloor_param_file) |
| else: |
| self.shopfloor = None |
| self.device_index = device_index |
| self.ignore_enter_key = ignore_enter_key |
| self.auto_serial_number = auto_serial_number |
| |
| self.talk_to_fixture = use_als |
| self.config_loaded = False |
| self.status_names = self._STATUS_NAMES |
| self.status_labels = self._STATUS_LABELS |
| self.result_dict = {} |
| self.fail_cause = '' |
| self.base_config = PluggableConfig({}) |
| os.chdir(self.srcdir) |
| |
| if not self.shopfloor: |
| # Setup the usb disk and usb-to-serial adapter monitor. |
| usb_monitor = ConnectionMonitor() |
| usb_monitor.start(subsystem='block', device_type='disk', |
| on_insert=self.on_usb_insert, |
| on_remove=self.on_usb_remove) |
| |
| if self.talk_to_fixture: |
| u2s_monitor = ConnectionMonitor() |
| u2s_monitor.start(subsystem='usb-serial', |
| on_insert=self.on_u2s_insert, |
| on_remove=self.on_u2s_remove) |
| |
| if self.type == _TEST_TYPE_FULL or self.auto_serial_number: |
| input_serial_number = False |
| else: |
| input_serial_number = True |
| |
| # Startup the UI. |
| self.ui = UI() |
| self.register_events(['sync_fixture', 'exit_test', 'run_test', |
| 'download_param_from_shopfloor']) |
| self.ui.CallJSFunction("InitLayout", self.talk_to_fixture, |
| True if self.shopfloor else False, |
| input_serial_number, |
| self.ignore_enter_key) |
| try: |
| self.ui.Run() |
| except factory.FactoryTestFailure as e: |
| raise error.TestError(e.message) |
