mpdaio-test.py

   1 #!/usr/bin/python3
   2
   3 import logging
   4
   5 from asyncio import run
   6
   7 from mpdaio.connection import ConnectionPool
   8 from mpdaio.exceptions import MPDProtocolError
   9
  10
  11 HELLO_PREFIX = 'OK MPD '
  12
  13 async def _hello(conn):
  14     """Consume HELLO_PREFIX"""
  15     # await conn.drain()
  16     # data = await conn.readline()
  17     data = await conn.readuntil(b'\n')
  18     rcv = data.decode('utf-8')
  19     if not rcv.startswith(HELLO_PREFIX):
  20         raise MPDProtocolError(f'Got invalid MPD hello: "{rcv}"')
  21     logging.debug('consumed hello prefix')
  22     logging.debug('"%s"', rcv)
  23     version = rcv.split('\n')[0][len(HELLO_PREFIX):]
  24     logging.debug('version: %s', version)
  25     return version
  26
  27
  28 async def lookup(pool, server, port, query):
  29     try:
  30         conn = await pool.connect(server, port)
  31         logging.info(conn)
  32     except (ValueError, OSError):
  33         return {}
  34
  35     async with conn:
  36         await _hello(conn)
  37         conn.write(query.encode('utf-8'))
  38         conn.write(b'\n')
  39         await conn.drain()
  40         data = await conn.readuntil(b'OK\n')
  41         rcv = data.decode('utf-8')
  42         logging.info(rcv)
  43     await pool.close()
  44
  45
  46 def main():
  47     logging.basicConfig(level=logging.DEBUG)
  48     pool = ConnectionPool(max_connections=10)
  49     logging.info(pool)
  50     run(lookup(pool, 'hispaniola.lan', 6600,'currentsong'))
  51
  52
  53 if __name__ == '__main__':
  54     main()