Example for wrapping asyncio loop in thread about buttplug-py HOT 1 CLOSED

import concurrent
import asyncio
import time

def synchronous_code():
    time.sleep(100)
    return 1

async def asynchronous_code(device):
    while True:
        await device.send_vibrate_cmd(0.5)
        await asyncio.sleep(1)
        await device.send_stop_device_cmd()
        await asyncio.sleep(1)

async def main():
    device = get_device()
    asyncio.create_task(asynchronous_code(device))

    while True:
        loop = asyncio.get_running_loop()
        result = await loop.run_in_executor(concurrent.futures.ThreadPoolExecutor(), synchronous_code)

asyncio.run(main())