buttplug-py py3.7 implementation example

buttplug-test.py

from buttplug.client import (ButtplugClientWebsocketConnector, ButtplugClient,
                             ButtplugClientDevice)
import asyncio
# import signal
# import sys


async def cancel_me():
    print('cancel_me(): before sleep')

    try:
        await asyncio.sleep(3600)
    except asyncio.CancelledError:
        pass


# def signal_handler(sig, frame):
#     print('You pressed Ctrl+C!')
#     sys.exit(0)

async def run_vibration(dev: ButtplugClientDevice):
    await dev.send_vibrate_cmd(0.5)


def device_added(emitter, dev: ButtplugClientDevice):
    print("Device Added!")
    print(dev.name)
    asyncio.create_task(run_vibration(dev))


async def main():
    # signal.signal(signal.SIGINT, signal_handler)
    connector = ButtplugClientWebsocketConnector("ws://127.0.0.1:12345")
    client = ButtplugClient("Test Client", connector)
    client.device_added_handler += device_added
    await client.connect()
    await client.start_scanning()
    task = asyncio.create_task(cancel_me())
    try:
        await task
    except asyncio.CancelledError:
        pass
    await client.stop_scanning()
    await client.disconnect()

asyncio.run(main(), debug=True)