import cv2
import aiohttp
import asyncio
import concurrent.futures
import argparse
import numpy as np


async def camera_source(ws: aiohttp.ClientWebSocketResponse, threadpool: concurrent.futures.ThreadPoolExecutor, src_id: int=0):
    loop = asyncio.get_running_loop()
    try:
        src = await loop.run_in_executor(threadpool, lambda: cv2.VideoCapture(src_id))
        while True:
            _, im = await loop.run_in_executor(threadpool, src.read)
            im = cv2.resize(im, (640, 384))
            enc_param = [int(cv2.IMWRITE_JPEG_QUALITY), 40]
            _, im = await loop.run_in_executor(threadpool, lambda: cv2.imencode('.jpg', im, enc_param))
            await ws.send_bytes(im.tobytes())
    except asyncio.CancelledError:
        pass
    finally:
        src.release()

async def preview_window(queue: asyncio.Queue, threadpool: concurrent.futures.ThreadPoolExecutor):
    loop = asyncio.get_running_loop()
    try:
        while True:
            im = await queue.get()
            im = np.frombuffer(im, dtype=np.uint8)
            im = await loop.run_in_executor(threadpool, lambda: cv2.imdecode(im, cv2.IMREAD_ANYCOLOR))
            cv2.imshow('fpds_remote_preview', im)
            cv2.waitKey(1)
    except asyncio.CancelledError:
        pass
    finally:
        cv2.destroyAllWindows()

async def run_client(
        ws: aiohttp.ClientWebSocketResponse,
        threadpool: concurrent.futures.ThreadPoolExecutor
    ) -> None:
    # --
    dst_queue = asyncio.Queue(maxsize=1)
    src_task = asyncio.create_task(camera_source(ws, threadpool))
    dst_task = asyncio.create_task(preview_window(dst_queue, threadpool))
    try:
        while True:
            im = await ws.receive_bytes()
            await dst_queue.put(im)
    except asyncio.CancelledError:
        await ws.send_str('close')
        src_task.cancel()
        dst_task.cancel()
        await asyncio.wait([src_task, dst_task])

async def amain(url: str):
    with concurrent.futures.ThreadPoolExecutor(max_workers=4) as threadpool:
        async with aiohttp.ClientSession() as session, session.ws_connect(url) as ws:
            await run_client(ws, threadpool)

def main():
    parser = argparse.ArgumentParser('fpds.client')
    parser.add_argument('url', type=str, help='WebSocket endpoint of fpds.server e.g. http://localhost:8181/fpds')
    args = parser.parse_args()
    loop = asyncio.get_event_loop()
    task = loop.create_task(amain(args.url))
    try:
        loop.run_until_complete(task)
    except KeyboardInterrupt:
        task.cancel()
        loop.run_until_complete(asyncio.wait_for(task, timeout=None))
    finally:
        loop.close()

if __name__ == '__main__':
    main()

[(10, 'asyncio.get_running_loop', 'asyncio.get_running_loop', ([], {}), False, 'import asyncio\n'), (25, 'asyncio.get_running_loop', 'asyncio.get_running_loop', ([], {}), False, 'import asyncio\n'), (43, 'asyncio.Queue', 'asyncio.Queue', ([], {'maxsize': '(1)'}), False, 'import asyncio\n'), (62, 'argparse.ArgumentParser', 'argparse.ArgumentParser', (['"""fpds.client"""'], {}), False, 'import argparse\n'), (65, 'asyncio.get_event_loop', 'asyncio.get_event_loop', ([], {}), False, 'import asyncio\n'), (36, 'cv2.destroyAllWindows', 'cv2.destroyAllWindows', ([], {}), False, 'import cv2\n'), (15, 'cv2.resize', 'cv2.resize', (['im', '(640, 384)'], {}), False, 'import cv2\n'), (29, 'numpy.frombuffer', 'np.frombuffer', (['im'], {'dtype': 'np.uint8'}), True, 'import numpy as np\n'), (31, 'cv2.imshow', 'cv2.imshow', (['"""fpds_remote_preview"""', 'im'], {}), False, 'import cv2\n'), (32, 'cv2.waitKey', 'cv2.waitKey', (['(1)'], {}), False, 'import cv2\n'), (58, 'aiohttp.ClientSession', 'aiohttp.ClientSession', ([], {}), False, 'import aiohttp\n'), (54, 'asyncio.wait', 'asyncio.wait', (['[src_task, dst_task]'], {}), False, 'import asyncio\n'), (71, 'asyncio.wait_for', 'asyncio.wait_for', (['task'], {'timeout': 'None'}), False, 'import asyncio\n'), (12, 'cv2.VideoCapture', 'cv2.VideoCapture', (['src_id'], {}), False, 'import cv2\n'), (17, 'cv2.imencode', 'cv2.imencode', (['""".jpg"""', 'im', 'enc_param'], {}), False, 'import cv2\n'), (30, 'cv2.imdecode', 'cv2.imdecode', (['im', 'cv2.IMREAD_ANYCOLOR'], {}), False, 'import cv2\n')]