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
|
"""Provides a raw console to test module and demonstrate usage."""
import argparse
import asyncio
import logging
import anthemav
__all__ = ("console", "monitor")
async def console(loop, log):
"""Connect to receiver and show events as they occur.
Pulls the following arguments from the command line (not method arguments):
:param host:
Hostname or IP Address of the device.
:param port:
TCP port number of the device.
:param verbose:
Show debug logging.
"""
parser = argparse.ArgumentParser(description=console.__doc__)
parser.add_argument("--host", default="127.0.0.1", help="IP or FQDN of AVR")
parser.add_argument("--port", default="14999", help="Port of AVR")
parser.add_argument("--verbose", "-v", action="count")
args = parser.parse_args()
if args.verbose:
level = logging.DEBUG
else:
level = logging.INFO
logging.basicConfig(level=level)
def log_callback(message):
"""Receives event callback from Anthem Protocol class."""
log.info("Callback invoked: %s" % message)
host = args.host
port = int(args.port)
log.info("Connecting to Anthem AVR at %s:%i" % (host, port))
conn = await anthemav.Connection.create(
host=host, port=port, loop=loop, update_callback=log_callback
)
log.info("Power state is " + str(conn.protocol.power))
conn.protocol.power = True
await asyncio.sleep(5)
log.info("Power state is " + str(conn.protocol.power))
log.info("Model is %s", conn.protocol.model)
log.info("Number of zone is %s", len(conn.protocol.zones))
log.info("Volume is " + str(conn.protocol.zones[1].volume))
def monitor():
"""Wrapper to call console with a loop."""
log = logging.getLogger(__name__)
loop = asyncio.get_event_loop()
asyncio.ensure_future(console(loop, log))
loop.run_forever()
|
