import argparse
import asyncio
import logging
import os
import sys
sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..'))
import hidiocore.client
logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger(__name__)
class MyHidIoClient(hidiocore.client.HidIoClient):
async def on_connect(self, cap, cap_auth):
logger.info("Connected!")
print("Connected API Call", await cap.alive().a_wait())
async def on_disconnect(self):
logger.info("Disconnected!")
async def main(args):
client = MyHidIoClient('Python info gathering example')
_tasks = [ asyncio.gather(
*[client.connect(auth=hidiocore.client.HidIoClient.AUTH_BASIC)],
return_exceptions=True
)
]
while client.retry_connection_status():
if client.capability_hidioserver():
try:
nodes = (await asyncio.wait_for(
client.nodes(),
timeout=2.0
)).nodes
nodes = [n for n in nodes if n.type == 'hidioDaemon']
assert len(nodes) == 1, "There can be only one! ...hidioDaemon"
print(await nodes[0].node.daemon.info().a_wait())
return
except asyncio.TimeoutError:
logger.info("Timeout, trying again.")
continue
await asyncio.sleep(1)
parser = argparse.ArgumentParser(description='Info gathering example for HID-IO')
args = parser.parse_args()
try:
loop = asyncio.get_event_loop()
loop.run_until_complete(main(args))
except KeyboardInterrupt:
logger.warning("Ctrl+C detected, exiting...")
sys.exit(1)