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', False, 'import asyncio\n'), (62, 'argparse.ArgumentParser', 'argparse.ArgumentParser', 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', False, 'import cv2\n'), (29, 'numpy.frombuffer', 'np.frombuffer', True, 'import numpy as np\n'), (31, 'cv2.imshow', 'cv2.imshow', False, 'import cv2\n'), (32, 'cv2.waitKey', 'cv2.waitKey', False, 'import cv2\n'), (58, 'aiohttp.ClientSession', 'aiohttp.ClientSession', False, 'import aiohttp\n'), (54, 'asyncio.wait', 'asyncio.wait', False, 'import asyncio\n'), (71, 'asyncio.wait_for', 'asyncio.wait_for', False, 'import asyncio\n'), (12, 'cv2.VideoCapture', 'cv2.VideoCapture', False, 'import cv2\n'), (17, 'cv2.imencode', 'cv2.imencode', False, 'import cv2\n'), (30, 'cv2.imdecode', 'cv2.imdecode', False, 'import cv2\n')]