Merge changes I3f250bf8,Ib845ba56 am: 44e071883c am: f2211d05c0 (614711a8) · Commits · e / os / android_packages_modules_Bluetooth

android/pandora/mmi2grpc/mmi2grpc/init.py

+23 −7

Original line number	Diff line number	Diff line
		@@ -26,10 +26,12 @@ from mmi2grpc.a2dp import A2DPProxy
		from mmi2grpc.avrcp import AVRCPProxy
		from mmi2grpc.gatt import GATTProxy
		from mmi2grpc.hfp import HFPProxy
		from mmi2grpc.hid import HIDProxy
		from mmi2grpc.hogp import HOGPProxy
		from mmi2grpc.sdp import SDPProxy
		from mmi2grpc.sm import SMProxy
		from mmi2grpc._helpers import format_proxy
		from mmi2grpc._rootcanal import RootCanal

		from pandora_experimental.host_grpc import Host

		@@ -55,31 +57,40 @@ class IUT:
		"""
		self.port = port
		self.test = test
		self.rootcanal = None

		# Profile proxies.
		self._a2dp = None
		self._avrcp = None
		self._gatt = None
		self._hfp = None
		self._hid = None
		self._hogp = None
		self._sdp = None
		self._sm = None
		self._hogp = None

		def __enter__(self):
		"""Resets the IUT when starting a PTS test."""
		self.rootcanal = RootCanal()
		self.rootcanal.reconnect_phone()

		# Note: we don't keep a single gRPC channel instance in the IUT class
		# because reset is allowed to close the gRPC server.
		with grpc.insecure_channel(f'localhost:{self.port}') as channel:
		self._retry(Host(channel).HardReset)(wait_for_ready=True)

		def __exit__(self, exc_type, exc_value, exc_traceback):
		self.rootcanal.close()
		self.rootcanal = None

		self._a2dp = None
		self._avrcp = None
		self._gatt = None
		self._hfp = None
		self._hid = None
		self._hogp = None
		self._sdp = None
		self._sm = None
		self._hogp = None

		def _retry(self, func):

		@@ -154,6 +165,16 @@ class IUT:
		if not self._hfp:
		self._hfp = HFPProxy(grpc.insecure_channel(f'localhost:{self.port}'))
		return self._hfp.interact(test, interaction, description, pts_address)
		# Handles HID MMIs.
		if profile in ('HID'):
		if not self._hid:
		self._hid = HIDProxy(grpc.insecure_channel(f'localhost:{self.port}'), self.rootcanal)
		return self._hid.interact(test, interaction, description, pts_address)
		# Handles HOGP MMIs.
		if profile in ('HOGP'):
		if not self._hogp:
		self._hogp = HOGPProxy(grpc.insecure_channel(f'localhost:{self.port}'))
		return self._hogp.interact(test, interaction, description, pts_address)
		# Handles SDP MMIs.
		if profile in ('SDP'):
		if not self._sdp:
		@@ -164,11 +185,6 @@ class IUT:
		if not self._sm:
		self._sm = SMProxy(grpc.insecure_channel(f'localhost:{self.port}'))
		return self._sm.interact(test, interaction, description, pts_address)
		# Handles HOGP MMIs.
		if profile in ('HOGP'):
		if not self._hogp:
		self._hogp = HOGPProxy(grpc.insecure_channel(f'localhost:{self.port}'))
		return self._hogp.interact(test, interaction, description, pts_address)

		# Handles unsupported profiles.
		code = format_proxy(profile, interaction, description)

android/pandora/mmi2grpc/mmi2grpc/_rootcanal.py

+22 −15

Original line number	Diff line number	Diff line
		@@ -84,6 +84,9 @@ class RootCanal:
		# discard initialization messages
		self.channel.receive_response()

		def close(self):
		self.channel.close()

		@staticmethod
		def _parse_device_list(raw):
		# time for some cursed parsing!
		@@ -112,8 +115,10 @@ class RootCanal:
		return transport, idxs

		def reconnect_phone(self):
		devices = self.channel.send_command("list", [])
		devices = self._parse_device_list(devices)
		raw_devices = None
		try:
		raw_devices = self.channel.send_command("list", [])
		devices = self._parse_device_list(raw_devices)

		for dev_i, name in enumerate(devices["Devices"]):
		# the default transports are always 0 and 1
		@@ -127,6 +132,8 @@ class RootCanal:
		for phy in target_phys:
		if dev_i not in self._parse_phy(devices["Phys"][phy])[1]:
		self.channel.send_command("add_device_to_phy", [dev_i, phy])
		except Exception as e:
		print(raw_devices, e)

		def disconnect_phy(self):
		# first, list all devices

android/pandora/mmi2grpc/mmi2grpc/hid.py

0 → 100644

+185 −0

Original line number	Diff line number	Diff line
		from threading import Thread
		from time import sleep
		from mmi2grpc._helpers import assert_description, match_description
		from mmi2grpc._proxy import ProfileProxy

		from pandora_experimental.hid_grpc import HID
		from pandora_experimental.host_grpc import Host
		from pandora_experimental.hid_pb2 import HID_REPORT_TYPE_OUTPUT
		from mmi2grpc._rootcanal import RootCanal


		class HIDProxy(ProfileProxy):

		def __init__(self, channel, rootcanal):
		super().__init__()
		self.hid = HID(channel)
		self.host = Host(channel)
		self.rootcanal = rootcanal
		self.connection = None

		@assert_description
		def TSC_MMI_iut_enable_connection(self, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then using the Implementation Under Test (IUT) connect to the
		PTS.
		"""

		self.rootcanal.reconnect_phy_if_needed()
		self.connection = self.host.Connect(address=pts_addr).connection

		return "OK"

		@assert_description
		def TSC_MMI_iut_release_connection(self, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then release the HID connection from the Implementation Under
		Test (IUT) by closing the Interrupt Channel followed by the Control
		Channel.

		Description: This can be done using the anticipated L2CAP
		Disconnection Requests. If the host is unable to perform the connection
		request, the IUT may break the ACL or Baseband Link by going out of
		range.
		"""

		self.host.Disconnect(connection=self.connection)

		return "OK"

		@assert_description
		def TSC_MMI_iut_disable_connection(self, pts_addr: bytes, **kwargs):
		"""
		Disable the connection using the Implementation UnderTest (IUT).

		Note:
		The IUT may either disconnect the Interupt Control Channels or send a
		host initiated virtual cable unplug and wait for the PTS to disconnect
		the channels.
		"""

		self.host.Disconnect(connection=self.connection)
		self.connection = None

		return "OK"

		@assert_description
		def TSC_HID_MMI_iut_accept_connection_ready_confirm(self, **kwargs):
		"""
		Please prepare the IUT to accept connection from PTS and then click OK.
		"""

		self.rootcanal.reconnect_phy_if_needed()

		return "OK"

		@assert_description
		def TSC_MMI_iut_connectable_enter_pw_dev(self, **kwargs):
		"""
		Make the Implementation Under Test (IUT) connectable, then click Ok.
		"""

		self.rootcanal.reconnect_phy_if_needed()

		return "OK"

		@assert_description
		def TSC_HID_MMI_iut_accept_control_channel(self, pts_addr: bytes, **kwargs):
		"""
		Accept the control channel connection from the Implementation Under Test
		(IUT).
		"""

		return "OK"

		@assert_description
		def TSC_MMI_tester_release_connection(self, **kwargs):
		"""
		Place the Implementation Under Test (IUT) in a state which will allow
		the PTS to perform an HID connection release, then click Ok.

		Note: The
		PTS will send an L2CAP disconnect request for the Interrupt channel,
		then the control channel.
		"""

		return "OK"

		@assert_description
		def TSC_MMI_host_iut_prepare_to_receive_pointing_data(self, **kwargs):
		"""
		Place the Implementation Under Test (IUT) in a state to receive and
		verify HID pointing data, then click Ok.
		"""

		return "OK"

		@assert_description
		def TSC_MMI_host_iut_verify_pointing_data(self, **kwargs):
		"""
		Verify that the pointer on the Implementation Under Test (IUT) moved to
		the left (X< 0), then click Ok.
		"""

		# TODO: implement!

		return "OK"

		@assert_description
		def TSC_MMI_host_send_output_report(self, pts_addr: bytes, **kwargs):
		"""
		Send an output report from the HOST.
		"""

		self.hid.SendHostReport(
		address=pts_addr,
		report_type=HID_REPORT_TYPE_OUTPUT,
		report="8", # keyboard enable num-lock
		)

		return "OK"

		@match_description
		def TSC_MMI_verify_output_report(self, **kwargs):
		"""
		Verify that the output report is correct. nnOutput Report =0(?:0\|1)'
		"""

		# TODO: check the report matches the num-lock setting

		return "OK"

		@assert_description
		def TSC_MMI_rf_shield_iut_or_tester(self, pts_addr: bytes, **kwargs):
		"""
		Click Ok, then perform one of the following actions:

		1. Move the PTS
		and Implementation Under Test (IUT) out of range of each other.
		2. Place
		either the PTS or IUT in an RF sheild box.
		"""

		def disconnect():
		sleep(2)
		self.rootcanal.disconnect_phy()

		Thread(target=disconnect).start()

		return "OK"

		@assert_description
		def TSC_MMI_iut_auto_connection(self, pts_addr: bytes, **kwargs):
		"""
		Click OK, then initiate a HID connection automatically from the IUT to
		the PTS
		"""

		def connect():
		sleep(1)
		self.rootcanal.reconnect_phy_if_needed()
		self.connection = self.host.Connect(address=pts_addr).connection

		Thread(target=connect).start()

		return "OK"

android/pandora/server/Android.bp

+3 −0

Original line number	Diff line number	Diff line
		@@ -134,6 +134,7 @@ genrule {
		"proto/pandora_experimental/avrcp.proto",
		"proto/pandora_experimental/gatt.proto",
		"proto/pandora_experimental/hfp.proto",
		"proto/pandora_experimental/hid.proto",
		"proto/pandora_experimental/host.proto",
		"proto/pandora_experimental/security.proto",
		],
		@@ -146,6 +147,8 @@ genrule {
		"pandora_experimental/gatt_pb2.py",
		"pandora_experimental/hfp_grpc.py",
		"pandora_experimental/hfp_pb2.py",
		"pandora_experimental/hid_grpc.py",
		"pandora_experimental/hid_pb2.py",
		"pandora_experimental/host_grpc.py",
		"pandora_experimental/host_pb2.py",
		"pandora_experimental/security_grpc.py",

android/pandora/server/configs/PtsBotTest.xml

+2 −1

Original line number	Diff line number	Diff line
		@@ -32,10 +32,11 @@
		<option name="profile" value="HFP/AG/HFI" />
		<option name="profile" value="HFP/AG/SLC" />
		<option name="profile" value="HFP/AG/TCA" />
		<option name="profile" value="HID/HOS" />
		<option name="profile" value="HOGP/RH" />
		<option name="profile" value="SDP/SR" />
		<option name="profile" value="SM/CEN/EKS" />
		<option name="profile" value="SM/CEN/JW" />
		<option name="profile" value="SM/CEN/KDU" />
		<option name="profile" value="HOGP/RH" />
		</test>
		</configuration>