blob: 4d4074d235cecc7124e40b28d0dc76836daabc6c (
plain)
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
|
from typing import Optional
import grpc
import time
import sys
import textwrap
from .a2dp import A2DPProxy
from blueberry.host_grpc import Host
GRPC_PORT = 8999
_a2dp: Optional[A2DPProxy] = None
def run(profile: str, interaction_id: str, test: str, description: str, pts_addr: bytes):
global _a2dp
print(f'{profile} mmi: {interaction_id}', file=sys.stderr)
if profile in ('A2DP', 'AVDTP'):
if not _a2dp:
_a2dp = A2DPProxy(grpc.insecure_channel(f'localhost:{GRPC_PORT}'))
return _a2dp.interact(interaction_id, test, description, pts_addr)
def reset():
global _a2dp
_a2dp = None
with grpc.insecure_channel(f'localhost:{GRPC_PORT}') as channel:
Host(channel).Reset(wait_for_ready=True)
def read_local_address() -> bytes:
with grpc.insecure_channel(f'localhost:{GRPC_PORT}') as channel:
try:
return Host(channel).ReadLocalAddress(wait_for_ready=True).address
except grpc.RpcError:
print('Retry')
time.sleep(5)
return Host(channel).ReadLocalAddress(wait_for_ready=True).address
|
