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, 11, 10, 37), 'asyncio.get_running_loop', 'asyncio.get_running_loop', ({}, {}), '()', False, 'import asyncio\n'), ((25, 11, 25, 37), 'asyncio.get_running_loop', 'asyncio.get_running_loop', ({}, {}), '()', False, 'import asyncio\n'), ((43, 16, 43, 40), 'asyncio.Queue', 'asyncio.Queue', (), '', False, 'import asyncio\n'), ((62, 13, 62, 51), 'argparse.ArgumentParser', 'argparse.ArgumentParser', ({(62, 37, 62, 50): '"""fpds.client"""'}, {}), "('fpds.client')", False, 'import argparse\n'), ((65, 11, 65, 35), 'asyncio.get_event_loop', 'asyncio.get_event_loop', ({}, {}), '()', False, 'import asyncio\n'), ((36, 8, 36, 31), 'cv2.destroyAllWindows', 'cv2.destroyAllWindows', ({}, {}), '()', False, 'import cv2\n'), ((15, 17, 15, 43), 'cv2.resize', 'cv2.resize', ({(15, 28, 15, 30): 'im', (15, 32, 15, 42): '(640, 384)'}, {}), '(im, (640, 384))', False, 'import cv2\n'), ((29, 17, 29, 50), 'numpy.frombuffer', 'np.frombuffer', (), '', True, 'import numpy as np\n'), ((31, 12, 31, 49), 'cv2.imshow', 'cv2.imshow', ({(31, 23, 31, 44): '"""fpds_remote_preview"""', (31, 46, 31, 48): 'im'}, {}), "('fpds_remote_preview', im)", False, 'import cv2\n'), ((32, 12, 32, 26), 'cv2.waitKey', 'cv2.waitKey', ({(32, 24, 32, 25): '(1)'}, {}), '(1)', False, 'import cv2\n'), ((58, 19, 58, 42), 'aiohttp.ClientSession', 'aiohttp.ClientSession', ({}, {}), '()', False, 'import aiohttp\n'), ((54, 14, 54, 48), 'asyncio.wait', 'asyncio.wait', ({(54, 27, 54, 47): '[src_task, dst_task]'}, {}), '([src_task, dst_task])', False, 'import asyncio\n'), ((71, 32, 71, 68), 'asyncio.wait_for', 'asyncio.wait_for', (), '', False, 'import asyncio\n'), ((12, 61, 12, 85), 'cv2.VideoCapture', 'cv2.VideoCapture', ({(12, 78, 12, 84): 'src_id'}, {}), '(src_id)', False, 'import cv2\n'), ((17, 67, 17, 102), 'cv2.imencode', 'cv2.imencode', ({(17, 80, 17, 86): '""".jpg"""', (17, 88, 17, 90): 'im', (17, 92, 17, 101): 'enc_param'}, {}), "('.jpg', im, enc_param)", False, 'import cv2\n'), ((30, 64, 30, 101), 'cv2.imdecode', 'cv2.imdecode', ({(30, 77, 30, 79): 'im', (30, 81, 30, 100): 'cv2.IMREAD_ANYCOLOR'}, {}), '(im, cv2.IMREAD_ANYCOLOR)', False, 'import cv2\n')]