1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
|
# Copyright 2022 Google LLC
#
# 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
#
# https://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.
"""Map Bluetooth PTS Man Machine Interface to Pandora gRPC calls."""
__version__ = "0.0.1"
from typing import List
import time
import sys
import grpc
from mmi2grpc.a2dp import A2DPProxy
from mmi2grpc._helpers import format_proxy
from pandora.host_grpc import Host
GRPC_PORT = 8999
MAX_RETRIES = 10
class IUT:
"""IUT class.
Handles MMI calls from the PTS and routes them to corresponding profile
proxy which translates MMI calls to gRPC calls to the IUT.
"""
def __init__(
self, test: str, args: List[str], port: int = GRPC_PORT, **kwargs):
"""Init IUT class for a given test.
Args:
test: PTS test id.
args: test arguments.
port: gRPC port exposed by the IUT test server.
"""
self.port = port
self.test = test
# Profile proxies.
self._a2dp = None
def __enter__(self):
"""Resets the IUT when starting a PTS test."""
# 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:
Host(channel).Reset(wait_for_ready=True)
def __exit__(self, exc_type, exc_value, exc_traceback):
self._a2dp = None
@property
def address(self) -> bytes:
"""Bluetooth MAC address of the IUT."""
with grpc.insecure_channel(f'localhost:{self.port}') as channel:
tries = 0
while True:
try:
return Host(channel).ReadLocalAddress(
wait_for_ready=True).address
except grpc.RpcError or grpc._channel._InactiveRpcError:
tries += 1
if tries >= MAX_RETRIES:
raise
else:
print('Retry', tries, 'of', MAX_RETRIES)
time.sleep(1)
def interact(self,
pts_address: bytes,
profile: str,
test: str,
interaction: str,
description: str,
style: str,
**kwargs) -> str:
"""Routes MMI calls to corresponding profile proxy.
Args:
pts_address: Bluetooth MAC addres of the PTS in bytes.
profile: Bluetooth profile.
test: PTS test id.
interaction: MMI name.
description: MMI description.
style: MMI popup style, unused for now.
"""
print(f'{profile} mmi: {interaction}', file=sys.stderr)
# Handles A2DP and AVDTP MMIs.
if profile in ('A2DP', 'AVDTP'):
if not self._a2dp:
self._a2dp = A2DPProxy(
grpc.insecure_channel(f'localhost:{self.port}'))
return self._a2dp.interact(
test, interaction, description, pts_address)
# Handles unsupported profiles.
code = format_proxy(profile, interaction, description)
error_msg = (
f'Missing {profile} proxy and mmi: {interaction}\n'
f'Create a {profile.lower()}.py in mmi2grpc/:\n\n{code}\n'
f'Then, instantiate the corresponding proxy in __init__.py\n'
f'Finally, create a {profile.lower()}.proto in proto/pandora/'
f'and generate the corresponding interface.')
assert False, error_msg
|
