-
Notifications
You must be signed in to change notification settings - Fork 5
/
Copy pathtest_pybeoplay_async.py
101 lines (81 loc) · 3.47 KB
/
test_pybeoplay_async.py
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
from pybeoplay import BeoPlay
import logging
import asyncio
import aiohttp
LOG = logging.getLogger(__name__)
async def test_volume(gateway : BeoPlay):
await asyncio.sleep(10)
await gateway.async_set_volume(0.40)
await asyncio.sleep(4)
await gateway.async_set_volume(0.37)
async def get_status(gateway : BeoPlay):
while True:
try:
await asyncio.sleep(15)
_on : bool = await gateway.async_get_standby()
print("Status is on: ", str(_on))
except Exception as e:
LOG.info("Exception %s", str(e))
async def test_remote(gateway : BeoPlay):
await gateway.async_turn_on()
await asyncio.sleep(5)
await gateway.async_remote_command('Cursor/Down')
await asyncio.sleep(4)
await gateway.async_remote_command('Stream/Play')
async def test_digits(gateway : BeoPlay):
await asyncio.sleep(10)
await gateway.async_digits('1')
await asyncio.sleep(2)
await gateway.async_digits('0')
await asyncio.sleep(2)
await gateway.async_digits('99')
async def main(host):
timeout = aiohttp.ClientTimeout(total=None, connect=None, sock_connect=None, sock_read=None)
async with aiohttp.ClientSession(timeout = timeout) as session:
gateway = BeoPlay(host, session)
await gateway.async_get_device_info()
print ("Serial Number: " , gateway.serialNumber)
print ("Type Number: ", gateway.typeNumber)
print ("Item Number: ",gateway.itemNumber)
print ("Name: ",gateway.name)
print ("Standby: ", gateway.on)
sources = await gateway.async_get_sources()
print ("Sources: ", sources)
await gateway.async_get_source()
print ("Active Source: ", gateway.source)
modes = await gateway.async_get_sound_modes()
print ("Sound Modes: ", modes)
print ("Active Sound Mode: ", gateway.soundMode)
asyncio.create_task(test_remote(gateway))
asyncio.create_task(test_digits(gateway))
asyncio.create_task(test_volume(gateway))
asyncio.create_task(get_status(gateway))
def callback(json_data):
print ("On State: " , gateway.on)
print ("Source: ", gateway.source)
print ("Min Volume: " , gateway.min_volume)
print ("Max Volume: " , gateway.max_volume)
print ("Volume: " , gateway.volume)
print ("Muted: " , gateway.muted)
print ("State: " , gateway.state)
print ("Media URL: " , gateway.media_url)
print ("Media track: " , gateway.media_track)
print ("Media artist: " , gateway.media_artist)
print ("Media album: " , gateway.media_album)
print ("Media genre: " , gateway.media_genre)
print ("Media country: " , gateway.media_country)
print ("Media languages: " , gateway.media_languages)
print ("Primary experience: " , gateway.primary_experience)
print ("Sound Mode: " , gateway.soundMode)
r = await gateway.async_notificationsTask(callback)
print (str(r))
if __name__ == '__main__':
import sys
ch = logging.StreamHandler(sys.stdout)
logging.basicConfig(level=logging.DEBUG)
ch.setLevel(logging.DEBUG)
LOG.addHandler(ch)
if len(sys.argv) < 2:
quit()
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
asyncio.run(main(sys.argv[1]))