| # Copyright 2013 The Android Open Source Project |
| # |
| # Licensed under the Apache License, Version 2.0 (the "License"); |
| # you may not use this file except in compliance with the License. |
| # You may obtain a copy of the License at |
| # |
| # http://www.apache.org/licenses/LICENSE-2.0 |
| # |
| # Unless required by applicable law or agreed to in writing, software |
| # distributed under the License is distributed on an "AS IS" BASIS, |
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| # See the License for the specific language governing permissions and |
| # limitations under the License. |
| |
| import json |
| import os |
| import socket |
| import string |
| import subprocess |
| import sys |
| import time |
| import unicodedata |
| import unittest |
| |
| import its.error |
| import numpy |
| |
| from collections import namedtuple |
| |
| |
| class ItsSession(object): |
| """Controls a device over adb to run ITS scripts. |
| |
| The script importing this module (on the host machine) prepares JSON |
| objects encoding CaptureRequests, specifying sets of parameters to use |
| when capturing an image using the Camera2 APIs. This class encapsulates |
| sending the requests to the device, monitoring the device's progress, and |
| copying the resultant captures back to the host machine when done. TCP |
| forwarded over adb is the transport mechanism used. |
| |
| The device must have CtsVerifier.apk installed. |
| |
| Attributes: |
| sock: The open socket. |
| """ |
| |
| # Open a connection to localhost:<host_port>, forwarded to port 6000 on the |
| # device. <host_port> is determined at run-time to support multiple |
| # connected devices. |
| IPADDR = '127.0.0.1' |
| REMOTE_PORT = 6000 |
| BUFFER_SIZE = 4096 |
| |
| # LOCK_PORT is used as a mutex lock to protect the list of forwarded ports |
| # among all processes. The script assumes LOCK_PORT is available and will |
| # try to use ports between CLIENT_PORT_START and |
| # CLIENT_PORT_START+MAX_NUM_PORTS-1 on host for ITS sessions. |
| CLIENT_PORT_START = 6000 |
| MAX_NUM_PORTS = 100 |
| LOCK_PORT = CLIENT_PORT_START + MAX_NUM_PORTS |
| |
| # Seconds timeout on each socket operation. |
| SOCK_TIMEOUT = 20.0 |
| # Additional timeout in seconds when ITS service is doing more complicated |
| # operations, for example: issuing warmup requests before actual capture. |
| EXTRA_SOCK_TIMEOUT = 5.0 |
| |
| SEC_TO_NSEC = 1000*1000*1000.0 |
| |
| PACKAGE = 'com.android.cts.verifier.camera.its' |
| INTENT_START = 'com.android.cts.verifier.camera.its.START' |
| ACTION_ITS_RESULT = 'com.android.cts.verifier.camera.its.ACTION_ITS_RESULT' |
| EXTRA_VERSION = 'camera.its.extra.VERSION' |
| CURRENT_ITS_VERSION = '1.0' # version number to sync with CtsVerifier |
| EXTRA_CAMERA_ID = 'camera.its.extra.CAMERA_ID' |
| EXTRA_RESULTS = 'camera.its.extra.RESULTS' |
| ITS_TEST_ACTIVITY = 'com.android.cts.verifier/.camera.its.ItsTestActivity' |
| |
| # This string must be in sync with ItsService. Updated when interface |
| # between script and ItsService is changed. |
| ITS_SERVICE_VERSION = "1.0" |
| |
| RESULT_PASS = 'PASS' |
| RESULT_FAIL = 'FAIL' |
| RESULT_NOT_EXECUTED = 'NOT_EXECUTED' |
| RESULT_VALUES = {RESULT_PASS, RESULT_FAIL, RESULT_NOT_EXECUTED} |
| RESULT_KEY = 'result' |
| SUMMARY_KEY = 'summary' |
| START_TIME_KEY = 'start' |
| END_TIME_KEY = 'end' |
| |
| adb = "adb -d" |
| device_id = "" |
| |
| CAMERA_ID_TOKENIZER = '.' |
| |
| # Definitions for some of the common output format options for do_capture(). |
| # Each gets images of full resolution for each requested format. |
| CAP_RAW = {"format":"raw"} |
| CAP_DNG = {"format":"dng"} |
| CAP_YUV = {"format":"yuv"} |
| CAP_JPEG = {"format":"jpeg"} |
| CAP_RAW_YUV = [{"format":"raw"}, {"format":"yuv"}] |
| CAP_DNG_YUV = [{"format":"dng"}, {"format":"yuv"}] |
| CAP_RAW_JPEG = [{"format":"raw"}, {"format":"jpeg"}] |
| CAP_DNG_JPEG = [{"format":"dng"}, {"format":"jpeg"}] |
| CAP_YUV_JPEG = [{"format":"yuv"}, {"format":"jpeg"}] |
| CAP_RAW_YUV_JPEG = [{"format":"raw"}, {"format":"yuv"}, {"format":"jpeg"}] |
| CAP_DNG_YUV_JPEG = [{"format":"dng"}, {"format":"yuv"}, {"format":"jpeg"}] |
| |
| # Predefine camera props. Save props extracted from the function, |
| # "get_camera_properties". |
| props = None |
| |
| # Initialize the socket port for the host to forward requests to the device. |
| # This method assumes localhost's LOCK_PORT is available and will try to |
| # use ports between CLIENT_PORT_START and CLIENT_PORT_START+MAX_NUM_PORTS-1 |
| def __init_socket_port(self): |
| NUM_RETRIES = 100 |
| RETRY_WAIT_TIME_SEC = 0.05 |
| |
| # Bind a socket to use as mutex lock |
| socket_lock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| for i in range(NUM_RETRIES): |
| try: |
| socket_lock.bind((ItsSession.IPADDR, ItsSession.LOCK_PORT)) |
| break |
| except socket.error or socket.timeout: |
| if i == NUM_RETRIES - 1: |
| raise its.error.Error(self.device_id, |
| "socket lock returns error") |
| else: |
| time.sleep(RETRY_WAIT_TIME_SEC) |
| |
| # Check if a port is already assigned to the device. |
| command = "adb forward --list" |
| proc = subprocess.Popen(command.split(), stdout=subprocess.PIPE) |
| output, error = proc.communicate() |
| |
| port = None |
| used_ports = [] |
| for line in output.split(os.linesep): |
| # each line should be formatted as: |
| # "<device_id> tcp:<host_port> tcp:<remote_port>" |
| forward_info = line.split() |
| if len(forward_info) >= 3 and \ |
| len(forward_info[1]) > 4 and forward_info[1][:4] == "tcp:" and \ |
| len(forward_info[2]) > 4 and forward_info[2][:4] == "tcp:": |
| local_p = int(forward_info[1][4:]) |
| remote_p = int(forward_info[2][4:]) |
| if forward_info[0] == self.device_id and \ |
| remote_p == ItsSession.REMOTE_PORT: |
| port = local_p |
| break |
| else: |
| used_ports.append(local_p) |
| |
| # Find the first available port if no port is assigned to the device. |
| if port is None: |
| for p in range(ItsSession.CLIENT_PORT_START, |
| ItsSession.CLIENT_PORT_START + |
| ItsSession.MAX_NUM_PORTS): |
| if p not in used_ports: |
| # Try to run "adb forward" with the port |
| command = "%s forward tcp:%d tcp:%d" % \ |
| (self.adb, p, self.REMOTE_PORT) |
| proc = subprocess.Popen(command.split(), |
| stdout=subprocess.PIPE, |
| stderr=subprocess.PIPE) |
| output, error = proc.communicate() |
| |
| # Check if there is no error |
| if error is None or error.find("error") < 0: |
| port = p |
| break |
| |
| if port is None: |
| raise its.error.Error(self.device_id, " cannot find an available " + |
| "port") |
| |
| # Release the socket as mutex unlock |
| socket_lock.close() |
| |
| # Connect to the socket |
| self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) |
| self.sock.connect((self.IPADDR, port)) |
| self.sock.settimeout(self.SOCK_TIMEOUT) |
| |
| # Reboot the device if needed and wait for the service to be ready for |
| # connection. |
| def __wait_for_service(self): |
| # This also includes the optional reboot handling: if the user |
| # provides a "reboot" or "reboot=N" arg, then reboot the device, |
| # waiting for N seconds (default 30) before returning. |
| for s in sys.argv[1:]: |
| if s[:6] == "reboot": |
| duration = 30 |
| if len(s) > 7 and s[6] == "=": |
| duration = int(s[7:]) |
| print "Rebooting device" |
| _run("%s reboot" % (self.adb)) |
| _run("%s wait-for-device" % (self.adb)) |
| time.sleep(duration) |
| print "Reboot complete" |
| |
| # Flush logcat so following code won't be misled by previous |
| # 'ItsService ready' log. |
| _run('%s logcat -c' % (self.adb)) |
| time.sleep(1) |
| |
| # TODO: Figure out why "--user 0" is needed, and fix the problem. |
| _run('%s shell am force-stop --user 0 %s' % (self.adb, self.PACKAGE)) |
| _run(('%s shell am start-foreground-service --user 0 -t text/plain ' |
| '-a %s') % (self.adb, self.INTENT_START)) |
| |
| # Wait until the socket is ready to accept a connection. |
| proc = subprocess.Popen( |
| self.adb.split() + ["logcat"], |
| stdout=subprocess.PIPE) |
| logcat = proc.stdout |
| while True: |
| line = logcat.readline().strip() |
| if line.find('ItsService ready') >= 0: |
| break |
| proc.kill() |
| |
| def __init__(self, camera_id=None, hidden_physical_id=None): |
| self._camera_id = camera_id |
| self._hidden_physical_id = hidden_physical_id |
| |
| def __enter__(self): |
| # Initialize device id and adb command. |
| self.device_id = get_device_id() |
| self.adb = "adb -s " + self.device_id |
| |
| self.__wait_for_service() |
| self.__init_socket_port() |
| |
| self.__close_camera() |
| self.__open_camera() |
| return self |
| |
| def __exit__(self, type, value, traceback): |
| if hasattr(self, 'sock') and self.sock: |
| self.__close_camera() |
| self.sock.close() |
| return False |
| |
| def __read_response_from_socket(self): |
| # Read a line (newline-terminated) string serialization of JSON object. |
| chars = [] |
| while len(chars) == 0 or chars[-1] != '\n': |
| ch = self.sock.recv(1) |
| if len(ch) == 0: |
| # Socket was probably closed; otherwise don't get empty strings |
| raise its.error.SocketError(self.device_id, 'Problem with socket on device side') |
| chars.append(ch) |
| line = ''.join(chars) |
| jobj = json.loads(line) |
| # Optionally read a binary buffer of a fixed size. |
| buf = None |
| if jobj.has_key("bufValueSize"): |
| n = jobj["bufValueSize"] |
| buf = bytearray(n) |
| view = memoryview(buf) |
| while n > 0: |
| nbytes = self.sock.recv_into(view, n) |
| view = view[nbytes:] |
| n -= nbytes |
| buf = numpy.frombuffer(buf, dtype=numpy.uint8) |
| return jobj, buf |
| |
| def __open_camera(self): |
| # Get the camera ID to open if it is an argument as a single camera. |
| # This allows passing camera=# to individual tests at command line |
| # and camera=#,#,# or an no camera argv with tools/run_all_tests.py. |
| # |
| # In case the camera is a logical multi-camera, to run ITS on the |
| # hidden physical sub-camera, pass camera=[logical ID]:[physical ID] |
| # to an individual test at the command line, and same applies to multiple |
| # camera IDs for tools/run_all_tests.py: camera=#,#:#,#:#,# |
| if not self._camera_id: |
| self._camera_id = 0 |
| for s in sys.argv[1:]: |
| if s[:7] == "camera=" and len(s) > 7: |
| camera_ids = s[7:].split(',') |
| camera_id_combos = parse_camera_ids(camera_ids) |
| if len(camera_id_combos) == 1: |
| self._camera_id = camera_id_combos[0].id |
| self._hidden_physical_id = camera_id_combos[0].sub_id |
| |
| cmd = {"cmdName":"open", "cameraId":self._camera_id} |
| self.sock.send(json.dumps(cmd) + "\n") |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'cameraOpened': |
| raise its.error.Error('Invalid command response') |
| |
| def __close_camera(self): |
| cmd = {"cmdName":"close"} |
| self.sock.send(json.dumps(cmd) + "\n") |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'cameraClosed': |
| raise its.error.Error('Invalid command response') |
| |
| def do_vibrate(self, pattern): |
| """Cause the device to vibrate to a specific pattern. |
| |
| Args: |
| pattern: Durations (ms) for which to turn on or off the vibrator. |
| The first value indicates the number of milliseconds to wait |
| before turning the vibrator on. The next value indicates the |
| number of milliseconds for which to keep the vibrator on |
| before turning it off. Subsequent values alternate between |
| durations in milliseconds to turn the vibrator off or to turn |
| the vibrator on. |
| |
| Returns: |
| Nothing. |
| """ |
| cmd = {} |
| cmd["cmdName"] = "doVibrate" |
| cmd["pattern"] = pattern |
| self.sock.send(json.dumps(cmd) + "\n") |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'vibrationStarted': |
| raise its.error.Error('Invalid command response') |
| |
| def set_audio_restriction(self, mode): |
| """Set the audio restriction mode for this camera device. |
| |
| Args: |
| mode: the audio restriction mode. See CameraDevice.java for valid |
| value. |
| Returns: |
| Nothing. |
| """ |
| cmd = {} |
| cmd["cmdName"] = "setAudioRestriction" |
| cmd["mode"] = mode |
| self.sock.send(json.dumps(cmd) + "\n") |
| data,_ = self.__read_response_from_socket() |
| if data["tag"] != "audioRestrictionSet": |
| raise its.error.Error("Invalid command response") |
| |
| def get_sensors(self): |
| """Get all sensors on the device. |
| |
| Returns: |
| A Python dictionary that returns keys and booleans for each sensor. |
| """ |
| cmd = {} |
| cmd["cmdName"] = "checkSensorExistence" |
| self.sock.send(json.dumps(cmd) + "\n") |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'sensorExistence': |
| raise its.error.Error('Invalid command response') |
| return data['objValue'] |
| |
| def start_sensor_events(self): |
| """Start collecting sensor events on the device. |
| |
| See get_sensor_events for more info. |
| |
| Returns: |
| Nothing. |
| """ |
| cmd = {} |
| cmd["cmdName"] = "startSensorEvents" |
| self.sock.send(json.dumps(cmd) + "\n") |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'sensorEventsStarted': |
| raise its.error.Error('Invalid command response') |
| |
| def get_sensor_events(self): |
| """Get a trace of all sensor events on the device. |
| |
| The trace starts when the start_sensor_events function is called. If |
| the test runs for a long time after this call, then the device's |
| internal memory can fill up. Calling get_sensor_events gets all events |
| from the device, and then stops the device from collecting events and |
| clears the internal buffer; to start again, the start_sensor_events |
| call must be used again. |
| |
| Events from the accelerometer, compass, and gyro are returned; each |
| has a timestamp and x,y,z values. |
| |
| Note that sensor events are only produced if the device isn't in its |
| standby mode (i.e.) if the screen is on. |
| |
| Returns: |
| A Python dictionary with three keys ("accel", "mag", "gyro") each |
| of which maps to a list of objects containing "time","x","y","z" |
| keys. |
| """ |
| cmd = {} |
| cmd["cmdName"] = "getSensorEvents" |
| self.sock.send(json.dumps(cmd) + "\n") |
| timeout = self.SOCK_TIMEOUT + self.EXTRA_SOCK_TIMEOUT |
| self.sock.settimeout(timeout) |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'sensorEvents': |
| raise its.error.Error('Invalid command response') |
| self.sock.settimeout(self.SOCK_TIMEOUT) |
| return data['objValue'] |
| |
| def get_camera_ids(self): |
| """Get a list of camera device Ids that can be opened. |
| |
| Returns: |
| a list of camera ID string |
| """ |
| cmd = {} |
| cmd["cmdName"] = "getCameraIds" |
| self.sock.send(json.dumps(cmd) + "\n") |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'cameraIds': |
| raise its.error.Error('Invalid command response') |
| return data['objValue']['cameraIdArray'] |
| |
| def check_its_version_compatible(self): |
| """Check the java side ItsService is compatible with current host script. |
| Raise ItsException if versions are incompatible |
| |
| Returns: None |
| """ |
| cmd = {} |
| cmd["cmdName"] = "getItsVersion" |
| self.sock.send(json.dumps(cmd) + "\n") |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'ItsVersion': |
| raise its.error.Error('ItsService is incompatible with host python script') |
| server_version = data['strValue'] |
| if self.ITS_SERVICE_VERSION != server_version: |
| raise its.error.Error('Version mismatch ItsService(%s) vs host script(%s)' % ( |
| server_version, ITS_SERVICE_VERSION)) |
| |
| def override_with_hidden_physical_camera_props(self, props): |
| """If current session is for a hidden physical camera, check that it is a valid |
| sub-camera backing the logical camera, and return the |
| characteristics of sub-camera. Otherwise, return "props" directly. |
| |
| Returns: The properties of the hidden physical camera if possible |
| """ |
| if self._hidden_physical_id: |
| e_msg = 'Camera %s is not a logical multi-camera' % self._camera_id |
| assert its.caps.logical_multi_camera(props), e_msg |
| physical_ids = its.caps.logical_multi_camera_physical_ids(props) |
| e_msg = 'Camera %s is not a hidden sub-camera of camera %s' % ( |
| self._hidden_physical_id, self._camera_id) |
| assert self._hidden_physical_id in physical_ids, e_msg |
| props = self.get_camera_properties_by_id(self._hidden_physical_id) |
| return props |
| |
| def get_camera_properties(self): |
| """Get the camera properties object for the device. |
| |
| Returns: |
| The Python dictionary object for the CameraProperties object. |
| """ |
| cmd = {} |
| cmd["cmdName"] = "getCameraProperties" |
| self.sock.send(json.dumps(cmd) + "\n") |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'cameraProperties': |
| raise its.error.Error('Invalid command response') |
| self.props = data['objValue']['cameraProperties'] |
| return data['objValue']['cameraProperties'] |
| |
| def get_camera_properties_by_id(self, camera_id): |
| """Get the camera properties object for device with camera_id |
| |
| Args: |
| camera_id: The ID string of the camera |
| |
| Returns: |
| The Python dictionary object for the CameraProperties object. Empty |
| if no such device exists. |
| |
| """ |
| cmd = {} |
| cmd["cmdName"] = "getCameraPropertiesById" |
| cmd["cameraId"] = camera_id |
| self.sock.send(json.dumps(cmd) + "\n") |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'cameraProperties': |
| raise its.error.Error('Invalid command response') |
| return data['objValue']['cameraProperties'] |
| |
| def do_3a(self, regions_ae=[[0,0,1,1,1]], |
| regions_awb=[[0,0,1,1,1]], |
| regions_af=[[0,0,1,1,1]], |
| do_ae=True, do_awb=True, do_af=True, |
| lock_ae=False, lock_awb=False, |
| get_results=False, |
| ev_comp=0, mono_camera=False): |
| """Perform a 3A operation on the device. |
| |
| Triggers some or all of AE, AWB, and AF, and returns once they have |
| converged. Uses the vendor 3A that is implemented inside the HAL. |
| Note: do_awb is always enabled regardless of do_awb flag |
| |
| Throws an assertion if 3A fails to converge. |
| |
| Args: |
| regions_ae: List of weighted AE regions. |
| regions_awb: List of weighted AWB regions. |
| regions_af: List of weighted AF regions. |
| do_ae: Trigger AE and wait for it to converge. |
| do_awb: Wait for AWB to converge. |
| do_af: Trigger AF and wait for it to converge. |
| lock_ae: Request AE lock after convergence, and wait for it. |
| lock_awb: Request AWB lock after convergence, and wait for it. |
| get_results: Return the 3A results from this function. |
| ev_comp: An EV compensation value to use when running AE. |
| mono_camera: Boolean for monochrome camera. |
| |
| Region format in args: |
| Arguments are lists of weighted regions; each weighted region is a |
| list of 5 values, [x,y,w,h, wgt], and each argument is a list of |
| these 5-value lists. The coordinates are given as normalized |
| rectangles (x,y,w,h) specifying the region. For example: |
| [[0.0, 0.0, 1.0, 0.5, 5], [0.0, 0.5, 1.0, 0.5, 10]]. |
| Weights are non-negative integers. |
| |
| Returns: |
| Five values are returned if get_results is true:: |
| * AE sensitivity; None if do_ae is False |
| * AE exposure time; None if do_ae is False |
| * AWB gains (list); |
| * AWB transform (list); |
| * AF focus position; None if do_af is false |
| Otherwise, it returns five None values. |
| """ |
| print "Running vendor 3A on device" |
| cmd = {} |
| cmd["cmdName"] = "do3A" |
| cmd["regions"] = {"ae": sum(regions_ae, []), |
| "awb": sum(regions_awb, []), |
| "af": sum(regions_af, [])} |
| cmd["triggers"] = {"ae": do_ae, "af": do_af} |
| if lock_ae: |
| cmd["aeLock"] = True |
| if lock_awb: |
| cmd["awbLock"] = True |
| if ev_comp != 0: |
| cmd["evComp"] = ev_comp |
| if self._hidden_physical_id: |
| cmd["physicalId"] = self._hidden_physical_id |
| self.sock.send(json.dumps(cmd) + "\n") |
| |
| # Wait for each specified 3A to converge. |
| ae_sens = None |
| ae_exp = None |
| awb_gains = None |
| awb_transform = None |
| af_dist = None |
| converged = False |
| while True: |
| data,_ = self.__read_response_from_socket() |
| vals = data['strValue'].split() |
| if data['tag'] == 'aeResult': |
| if do_ae: |
| ae_sens, ae_exp = [int(i) for i in vals] |
| elif data['tag'] == 'afResult': |
| if do_af: |
| af_dist = float(vals[0]) |
| elif data['tag'] == 'awbResult': |
| awb_gains = [float(f) for f in vals[:4]] |
| awb_transform = [float(f) for f in vals[4:]] |
| elif data['tag'] == '3aConverged': |
| converged = True |
| elif data['tag'] == '3aDone': |
| break |
| else: |
| raise its.error.Error('Invalid command response') |
| if converged and not get_results: |
| return None,None,None,None,None |
| if (do_ae and ae_sens == None or (not mono_camera and do_awb and awb_gains == None) |
| or do_af and af_dist == None or not converged): |
| |
| raise its.error.Error('3A failed to converge') |
| return ae_sens, ae_exp, awb_gains, awb_transform, af_dist |
| |
| def is_stream_combination_supported(self, out_surfaces): |
| """Query whether a output surfaces combination is supported by the camera device. |
| |
| This function hooks up to the isSessionConfigurationSupported() camera API |
| to query whether a particular stream combination is supported. |
| |
| Refer to do_capture function for specification of out_surfaces field. |
| """ |
| cmd = {} |
| cmd['cmdName'] = 'isStreamCombinationSupported' |
| |
| if not isinstance(out_surfaces, list): |
| cmd['outputSurfaces'] = [out_surfaces] |
| else: |
| cmd['outputSurfaces'] = out_surfaces |
| formats = [c['format'] if 'format' in c else 'yuv' |
| for c in cmd['outputSurfaces']] |
| formats = [s if s != 'jpg' else 'jpeg' for s in formats] |
| |
| self.sock.send(json.dumps(cmd) + '\n') |
| |
| data,_ = self.__read_response_from_socket() |
| if data['tag'] != 'streamCombinationSupport': |
| its.error.Error('Failed to query stream combination') |
| |
| return data['strValue'] == 'supportedCombination' |
| |
| def do_capture(self, cap_request, |
| out_surfaces=None, reprocess_format=None, repeat_request=None): |
| """Issue capture request(s), and read back the image(s) and metadata. |
| |
| The main top-level function for capturing one or more images using the |
| device. Captures a single image if cap_request is a single object, and |
| captures a burst if it is a list of objects. |
| |
| The optional repeat_request field can be used to assign a repeating |
| request list ran in background for 3 seconds to warm up the capturing |
| pipeline before start capturing. The repeat_requests will be ran on a |
| 640x480 YUV surface without sending any data back. The caller needs to |
| make sure the stream configuration defined by out_surfaces and |
| repeat_request are valid or do_capture may fail because device does not |
| support such stream configuration. |
| |
| The out_surfaces field can specify the width(s), height(s), and |
| format(s) of the captured image. The formats may be "yuv", "jpeg", |
| "dng", "raw", "raw10", "raw12", "rawStats" or "y8". The default is a YUV420 |
| frame ("yuv") corresponding to a full sensor frame. |
| |
| Optionally the out_surfaces field can specify physical camera id(s) if the |
| current camera device is a logical multi-camera. The physical camera id |
| must refer to a physical camera backing this logical camera device. |
| |
| Note that one or more surfaces can be specified, allowing a capture to |
| request images back in multiple formats (e.g.) raw+yuv, raw+jpeg, |
| yuv+jpeg, raw+yuv+jpeg. If the size is omitted for a surface, the |
| default is the largest resolution available for the format of that |
| surface. At most one output surface can be specified for a given format, |
| and raw+dng, raw10+dng, and raw+raw10 are not supported as combinations. |
| |
| If reprocess_format is not None, for each request, an intermediate |
| buffer of the given reprocess_format will be captured from camera and |
| the intermediate buffer will be reprocessed to the output surfaces. The |
| following settings will be turned off when capturing the intermediate |
| buffer and will be applied when reprocessing the intermediate buffer. |
| 1. android.noiseReduction.mode |
| 2. android.edge.mode |
| 3. android.reprocess.effectiveExposureFactor |
| |
| Supported reprocess format are "yuv" and "private". Supported output |
| surface formats when reprocessing is enabled are "yuv" and "jpeg". |
| |
| Example of a single capture request: |
| |
| { |
| "android.sensor.exposureTime": 100*1000*1000, |
| "android.sensor.sensitivity": 100 |
| } |
| |
| Example of a list of capture requests: |
| |
| [ |
| { |
| "android.sensor.exposureTime": 100*1000*1000, |
| "android.sensor.sensitivity": 100 |
| }, |
| { |
| "android.sensor.exposureTime": 100*1000*1000, |
| "android.sensor.sensitivity": 200 |
| } |
| ] |
| |
| Examples of output surface specifications: |
| |
| { |
| "width": 640, |
| "height": 480, |
| "format": "yuv" |
| } |
| |
| [ |
| { |
| "format": "jpeg" |
| }, |
| { |
| "format": "raw" |
| } |
| ] |
| |
| The following variables defined in this class are shortcuts for |
| specifying one or more formats where each output is the full size for |
| that format; they can be used as values for the out_surfaces arguments: |
| |
| CAP_RAW |
| CAP_DNG |
| CAP_YUV |
| CAP_JPEG |
| CAP_RAW_YUV |
| CAP_DNG_YUV |
| CAP_RAW_JPEG |
| CAP_DNG_JPEG |
| CAP_YUV_JPEG |
| CAP_RAW_YUV_JPEG |
| CAP_DNG_YUV_JPEG |
| |
| If multiple formats are specified, then this function returns multiple |
| capture objects, one for each requested format. If multiple formats and |
| multiple captures (i.e. a burst) are specified, then this function |
| returns multiple lists of capture objects. In both cases, the order of |
| the returned objects matches the order of the requested formats in the |
| out_surfaces parameter. For example: |
| |
| yuv_cap = do_capture( req1 ) |
| yuv_cap = do_capture( req1, yuv_fmt ) |
| yuv_cap, raw_cap = do_capture( req1, [yuv_fmt,raw_fmt] ) |
| yuv_caps = do_capture( [req1,req2], yuv_fmt ) |
| yuv_caps, raw_caps = do_capture( [req1,req2], [yuv_fmt,raw_fmt] ) |
| |
| The "rawStats" format processes the raw image and returns a new image |
| of statistics from the raw image. The format takes additional keys, |
| "gridWidth" and "gridHeight" which are size of grid cells in a 2D grid |
| of the raw image. For each grid cell, the mean and variance of each raw |
| channel is computed, and the do_capture call returns two 4-element float |
| images of dimensions (rawWidth / gridWidth, rawHeight / gridHeight), |
| concatenated back-to-back, where the first iamge contains the 4-channel |
| means and the second contains the 4-channel variances. Note that only |
| pixels in the active array crop region are used; pixels outside this |
| region (for example optical black rows) are cropped out before the |
| gridding and statistics computation is performed. |
| |
| For the rawStats format, if the gridWidth is not provided then the raw |
| image width is used as the default, and similarly for gridHeight. With |
| this, the following is an example of a output description that computes |
| the mean and variance across each image row: |
| |
| { |
| "gridHeight": 1, |
| "format": "rawStats" |
| } |
| |
| Args: |
| cap_request: The Python dict/list specifying the capture(s), which |
| will be converted to JSON and sent to the device. |
| out_surfaces: (Optional) specifications of the output image formats |
| and sizes to use for each capture. |
| reprocess_format: (Optional) The reprocessing format. If not None, |
| reprocessing will be enabled. |
| |
| Returns: |
| An object, list of objects, or list of lists of objects, where each |
| object contains the following fields: |
| * data: the image data as a numpy array of bytes. |
| * width: the width of the captured image. |
| * height: the height of the captured image. |
| * format: image the format, in [ |
| "yuv","jpeg","raw","raw10","raw12","rawStats","dng"]. |
| * metadata: the capture result object (Python dictionary). |
| """ |
| cmd = {} |
| if reprocess_format != None: |
| cmd["cmdName"] = "doReprocessCapture" |
| cmd["reprocessFormat"] = reprocess_format |
| else: |
| cmd["cmdName"] = "doCapture" |
| |
| if repeat_request is not None and reprocess_format is not None: |
| raise its.error.Error('repeating request + reprocessing is not supported') |
| |
| if repeat_request is None: |
| cmd["repeatRequests"] = [] |
| elif not isinstance(repeat_request, list): |
| cmd["repeatRequests"] = [repeat_request] |
| else: |
| cmd["repeatRequests"] = repeat_request |
| |
| if not isinstance(cap_request, list): |
| cmd["captureRequests"] = [cap_request] |
| else: |
| cmd["captureRequests"] = cap_request |
| if out_surfaces is not None: |
| if not isinstance(out_surfaces, list): |
| cmd["outputSurfaces"] = [out_surfaces] |
| else: |
| cmd["outputSurfaces"] = out_surfaces |
| formats = [c["format"] if "format" in c else "yuv" |
| for c in cmd["outputSurfaces"]] |
| formats = [s if s != "jpg" else "jpeg" for s in formats] |
| else: |
| max_yuv_size = its.objects.get_available_output_sizes( |
| "yuv", self.props)[0] |
| formats = ['yuv'] |
| cmd["outputSurfaces"] = [{"format": "yuv", |
| "width" : max_yuv_size[0], |
| "height": max_yuv_size[1]}] |
| |
| ncap = len(cmd["captureRequests"]) |
| nsurf = 1 if out_surfaces is None else len(cmd["outputSurfaces"]) |
| |
| cam_ids = [] |
| bufs = {} |
| yuv_bufs = {} |
| for i,s in enumerate(cmd["outputSurfaces"]): |
| if self._hidden_physical_id: |
| s['physicalCamera'] = self._hidden_physical_id |
| |
| if 'physicalCamera' in s: |
| cam_id = s['physicalCamera'] |
| else: |
| cam_id = self._camera_id |
| |
| if cam_id not in cam_ids: |
| cam_ids.append(cam_id) |
| bufs[cam_id] = {"raw":[], "raw10":[], "raw12":[], |
| "rawStats":[], "dng":[], "jpeg":[], "y8":[]} |
| |
| for cam_id in cam_ids: |
| # Only allow yuv output to multiple targets |
| if cam_id == self._camera_id: |
| yuv_surfaces = [s for s in cmd["outputSurfaces"] if s["format"]=="yuv"\ |
| and "physicalCamera" not in s] |
| formats_for_id = [s["format"] for s in cmd["outputSurfaces"] if \ |
| "physicalCamera" not in s] |
| else: |
| yuv_surfaces = [s for s in cmd["outputSurfaces"] if s["format"]=="yuv"\ |
| and "physicalCamera" in s and s["physicalCamera"] == cam_id] |
| formats_for_id = [s["format"] for s in cmd["outputSurfaces"] if \ |
| "physicalCamera" in s and s["physicalCamera"] == cam_id] |
| |
| n_yuv = len(yuv_surfaces) |
| # Compute the buffer size of YUV targets |
| yuv_maxsize_1d = 0 |
| for s in yuv_surfaces: |
| if not ("width" in s and "height" in s): |
| if self.props is None: |
| raise its.error.Error('Camera props are unavailable') |
| yuv_maxsize_2d = its.objects.get_available_output_sizes( |
| "yuv", self.props)[0] |
| yuv_maxsize_1d = yuv_maxsize_2d[0] * yuv_maxsize_2d[1] * 3 / 2 |
| break |
| yuv_sizes = [c["width"]*c["height"]*3/2 |
| if "width" in c and "height" in c |
| else yuv_maxsize_1d |
| for c in yuv_surfaces] |
| # Currently we don't pass enough metadta from ItsService to distinguish |
| # different yuv stream of same buffer size |
| if len(yuv_sizes) != len(set(yuv_sizes)): |
| raise its.error.Error( |
| 'ITS does not support yuv outputs of same buffer size') |
| if len(formats_for_id) > len(set(formats_for_id)): |
| if n_yuv != len(formats_for_id) - len(set(formats_for_id)) + 1: |
| raise its.error.Error('Duplicate format requested') |
| |
| yuv_bufs[cam_id] = {size:[] for size in yuv_sizes} |
| |
| raw_formats = 0; |
| raw_formats += 1 if "dng" in formats else 0 |
| raw_formats += 1 if "raw" in formats else 0 |
| raw_formats += 1 if "raw10" in formats else 0 |
| raw_formats += 1 if "raw12" in formats else 0 |
| raw_formats += 1 if "rawStats" in formats else 0 |
| if raw_formats > 1: |
| raise its.error.Error('Different raw formats not supported') |
| |
| # Detect long exposure time and set timeout accordingly |
| longest_exp_time = 0 |
| for req in cmd["captureRequests"]: |
| if "android.sensor.exposureTime" in req and \ |
| req["android.sensor.exposureTime"] > longest_exp_time: |
| longest_exp_time = req["android.sensor.exposureTime"] |
| |
| extended_timeout = longest_exp_time / self.SEC_TO_NSEC + \ |
| self.SOCK_TIMEOUT |
| if repeat_request: |
| extended_timeout += self.EXTRA_SOCK_TIMEOUT |
| self.sock.settimeout(extended_timeout) |
| |
| print "Capturing %d frame%s with %d format%s [%s]" % ( |
| ncap, "s" if ncap>1 else "", nsurf, "s" if nsurf>1 else "", |
| ",".join(formats)) |
| self.sock.send(json.dumps(cmd) + "\n") |
| |
| # Wait for ncap*nsurf images and ncap metadata responses. |
| # Assume that captures come out in the same order as requested in |
| # the burst, however individual images of different formats can come |
| # out in any order for that capture. |
| nbufs = 0 |
| mds = [] |
| physical_mds = [] |
| widths = None |
| heights = None |
| while nbufs < ncap*nsurf or len(mds) < ncap: |
| jsonObj,buf = self.__read_response_from_socket() |
| if jsonObj['tag'] in ['jpegImage', 'rawImage', \ |
| 'raw10Image', 'raw12Image', 'rawStatsImage', 'dngImage', 'y8Image'] \ |
| and buf is not None: |
| fmt = jsonObj['tag'][:-5] |
| bufs[self._camera_id][fmt].append(buf) |
| nbufs += 1 |
| elif jsonObj['tag'] == 'yuvImage': |
| buf_size = numpy.product(buf.shape) |
| yuv_bufs[self._camera_id][buf_size].append(buf) |
| nbufs += 1 |
| elif jsonObj['tag'] == 'captureResults': |
| mds.append(jsonObj['objValue']['captureResult']) |
| physical_mds.append(jsonObj['objValue']['physicalResults']) |
| outputs = jsonObj['objValue']['outputs'] |
| widths = [out['width'] for out in outputs] |
| heights = [out['height'] for out in outputs] |
| else: |
| tagString = unicodedata.normalize('NFKD', jsonObj['tag']).encode('ascii', 'ignore'); |
| for x in ['jpegImage', 'rawImage', \ |
| 'raw10Image', 'raw12Image', 'rawStatsImage', 'yuvImage']: |
| if tagString.startswith(x): |
| if x == 'yuvImage': |
| physicalId = jsonObj['tag'][len(x):] |
| if physicalId in cam_ids: |
| buf_size = numpy.product(buf.shape) |
| yuv_bufs[physicalId][buf_size].append(buf) |
| nbufs += 1 |
| else: |
| physicalId = jsonObj['tag'][len(x):] |
| if physicalId in cam_ids: |
| fmt = x[:-5] |
| bufs[physicalId][fmt].append(buf) |
| nbufs += 1 |
| rets = [] |
| for j,fmt in enumerate(formats): |
| objs = [] |
| if "physicalCamera" in cmd["outputSurfaces"][j]: |
| cam_id = cmd["outputSurfaces"][j]["physicalCamera"] |
| else: |
| cam_id = self._camera_id |
| |
| for i in range(ncap): |
| obj = {} |
| obj["width"] = widths[j] |
| obj["height"] = heights[j] |
| obj["format"] = fmt |
| if cam_id == self._camera_id: |
| obj["metadata"] = mds[i] |
| else: |
| for physical_md in physical_mds[i]: |
| if cam_id in physical_md: |
| obj["metadata"] = physical_md[cam_id] |
| break |
| |
| if fmt == "yuv": |
| buf_size = widths[j] * heights[j] * 3 / 2 |
| obj["data"] = yuv_bufs[cam_id][buf_size][i] |
| else: |
| obj["data"] = bufs[cam_id][fmt][i] |
| objs.append(obj) |
| rets.append(objs if ncap > 1 else objs[0]) |
| self.sock.settimeout(self.SOCK_TIMEOUT) |
| if len(rets) > 1 or (isinstance(rets[0], dict) and |
| isinstance(cap_request, list)): |
| return rets |
| else: |
| return rets[0] |
| |
| def do_capture_with_latency(cam, req, sync_latency, fmt=None): |
| """Helper function to take enough frames with do_capture to allow sync latency. |
| |
| Args: |
| cam: camera object |
| req: request for camera |
| sync_latency: integer number of frames |
| fmt: format for the capture |
| Returns: |
| single capture with the unsettled frames discarded |
| """ |
| caps = cam.do_capture([req]*(sync_latency+1), fmt) |
| return caps[-1] |
| |
| |
| def get_device_id(): |
| """Return the ID of the device that the test is running on. |
| |
| Return the device ID provided in the command line if it's connected. If no |
| device ID is provided in the command line and there is only one device |
| connected, return the device ID by parsing the result of "adb devices". |
| Also, if the environment variable ANDROID_SERIAL is set, use it as device |
| id. When both ANDROID_SERIAL and device argument present, device argument |
| takes priority. |
| |
| Raise an exception if no device is connected; or the device ID provided in |
| the command line is not connected; or no device ID is provided in the |
| command line or environment variable and there are more than 1 device |
| connected. |
| |
| Returns: |
| Device ID string. |
| """ |
| device_id = None |
| |
| # Check if device id is set in env |
| if "ANDROID_SERIAL" in os.environ: |
| device_id = os.environ["ANDROID_SERIAL"] |
| |
| for s in sys.argv[1:]: |
| if s[:7] == "device=" and len(s) > 7: |
| device_id = str(s[7:]) |
| |
| # Get a list of connected devices |
| devices = [] |
| command = "adb devices" |
| proc = subprocess.Popen(command.split(), stdout=subprocess.PIPE) |
| output, error = proc.communicate() |
| for line in output.split(os.linesep): |
| device_info = line.split() |
| if len(device_info) == 2 and device_info[1] == "device": |
| devices.append(device_info[0]) |
| |
| if len(devices) == 0: |
| raise its.error.Error("No device is connected!") |
| elif device_id is not None and device_id not in devices: |
| raise its.error.Error(device_id + " is not connected!") |
| elif device_id is None and len(devices) >= 2: |
| raise its.error.Error("More than 1 device are connected. " + |
| "Use device=<device_id> to specify a device to test.") |
| elif len(devices) == 1: |
| device_id = devices[0] |
| |
| return device_id |
| |
| def report_result(device_id, camera_id, results): |
| """Send a pass/fail result to the device, via an intent. |
| |
| Args: |
| device_id: The ID string of the device to report the results to. |
| camera_id: The ID string of the camera for which to report pass/fail. |
| results: a dictionary contains all ITS scenes as key and result/summary |
| of current ITS run. See test_report_result unit test for |
| an example. |
| Returns: |
| Nothing. |
| """ |
| ACTIVITY_START_WAIT = 1.5 # seconds |
| adb = "adb -s " + device_id |
| |
| # Start ItsTestActivity to receive test results |
| cmd = "%s shell am start %s --activity-brought-to-front" % (adb, ItsSession.ITS_TEST_ACTIVITY) |
| _run(cmd) |
| time.sleep(ACTIVITY_START_WAIT) |
| |
| # Validate/process results argument |
| for scene in results: |
| result_key = ItsSession.RESULT_KEY |
| summary_key = ItsSession.SUMMARY_KEY |
| if result_key not in results[scene]: |
| raise its.error.Error('ITS result not found for ' + scene) |
| if results[scene][result_key] not in ItsSession.RESULT_VALUES: |
| raise its.error.Error('Unknown ITS result for %s: %s' % ( |
| scene, results[result_key])) |
| if summary_key in results[scene]: |
| device_summary_path = "/sdcard/its_camera%s_%s.txt" % ( |
| camera_id, scene) |
| _run("%s push %s %s" % ( |
| adb, results[scene][summary_key], device_summary_path)) |
| results[scene][summary_key] = device_summary_path |
| |
| json_results = json.dumps(results) |
| cmd = "%s shell am broadcast -a %s --es %s %s --es %s %s --es %s \'%s\'" % ( |
| adb, ItsSession.ACTION_ITS_RESULT, |
| ItsSession.EXTRA_VERSION, ItsSession.CURRENT_ITS_VERSION, |
| ItsSession.EXTRA_CAMERA_ID, camera_id, |
| ItsSession.EXTRA_RESULTS, json_results) |
| if len(cmd) > 4095: |
| print "ITS command string might be too long! len:", len(cmd) |
| _run(cmd) |
| |
| def adb_log(device_id, msg): |
| """Send a log message to adb logcat |
| |
| Args: |
| device_id: The ID string of the adb device |
| msg: the message string to be send to logcat |
| |
| Returns: |
| Nothing. |
| """ |
| adb = "adb -s " + device_id |
| cmd = "%s shell log -p i -t \"ItsTestHost\" %s" % (adb, msg) |
| _run(cmd) |
| |
| def get_device_fingerprint(device_id): |
| """ Return the Build FingerPrint of the device that the test is running on. |
| |
| Returns: |
| Device Build Fingerprint string. |
| """ |
| device_bfp = None |
| |
| # Get a list of connected devices |
| |
| com = ('adb -s %s shell getprop | grep ro.build.fingerprint' % device_id) |
| proc = subprocess.Popen(com.split(), stdout=subprocess.PIPE) |
| output, error = proc.communicate() |
| assert error is None |
| |
| lst = string.split( \ |
| string.replace( \ |
| string.replace( \ |
| string.replace(output, |
| '\n', ''), '[', ''), ']', ''), \ |
| ' ') |
| |
| if lst[0].find('ro.build.fingerprint') != -1: |
| device_bfp = lst[1] |
| |
| return device_bfp |
| |
| def parse_camera_ids(ids): |
| """Parse the string of camera IDs into array of CameraIdCombo tuples. |
| """ |
| CameraIdCombo = namedtuple('CameraIdCombo', ['id', 'sub_id']) |
| id_combos = [] |
| for one_id in ids: |
| one_combo = one_id.split(ItsSession.CAMERA_ID_TOKENIZER) |
| if len(one_combo) == 1: |
| id_combos.append(CameraIdCombo(one_combo[0], None)) |
| elif len(one_combo) == 2: |
| id_combos.append(CameraIdCombo(one_combo[0], one_combo[1])) |
| else: |
| assert(False), 'Camera id parameters must be either ID, or ID:SUB_ID' |
| return id_combos |
| |
| |
| def get_build_sdk_version(device_id=None): |
| """Get the build version of the device.""" |
| if not device_id: |
| device_id = get_device_id() |
| cmd = 'adb -s %s shell getprop ro.build.version.sdk' % device_id |
| try: |
| build_sdk_version = int(subprocess.check_output(cmd.split()).rstrip()) |
| print 'Build SDK version: %d' % build_sdk_version |
| except (subprocess.CalledProcessError, ValueError): |
| print 'No build_sdk_version.' |
| assert 0 |
| return build_sdk_version |
| |
| |
| def get_first_api_level(device_id=None): |
| """Get the first API level for device.""" |
| if not device_id: |
| device_id = get_device_id() |
| cmd = 'adb -s %s shell getprop ro.product.first_api_level' % device_id |
| try: |
| first_api_level = int(subprocess.check_output(cmd.split()).rstrip()) |
| print 'First API level: %d' % first_api_level |
| except (subprocess.CalledProcessError, ValueError): |
| print 'No first_api_level. Setting to build version.' |
| first_api_level = get_build_sdk_version(device_id) |
| return first_api_level |
| |
| |
| def _run(cmd): |
| """Replacement for os.system, with hiding of stdout+stderr messages. |
| """ |
| with open(os.devnull, 'wb') as devnull: |
| subprocess.check_call( |
| cmd.split(), stdout=devnull, stderr=subprocess.STDOUT) |
| |
| |
| class __UnitTest(unittest.TestCase): |
| """Run a suite of unit tests on this module. |
| """ |
| |
| """ |
| # TODO: this test currently needs connected device to pass |
| # Need to remove that dependency before enabling the test |
| def test_report_result(self): |
| device_id = get_device_id() |
| camera_id = "1" |
| result_key = ItsSession.RESULT_KEY |
| results = {"scene0":{result_key:"PASS"}, |
| "scene1":{result_key:"PASS"}, |
| "scene2":{result_key:"PASS"}, |
| "scene3":{result_key:"PASS"}, |
| "sceneNotExist":{result_key:"FAIL"}} |
| report_result(device_id, camera_id, results) |
| """ |
| |
| if __name__ == '__main__': |
| unittest.main() |
| |