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()

[((243, 269), 'asyncio.get_running_loop', 'asyncio.get_running_loop', ([], {}), '()\n', (267, 269), False, 'import asyncio\n'), ((903, 929), 'asyncio.get_running_loop', 'asyncio.get_running_loop', ([], {}), '()\n', (927, 929), False, 'import asyncio\n'), ((1484, 1508), 'asyncio.Queue', 'asyncio.Queue', ([], {'maxsize': '(1)'}), '(maxsize=1)\n', (1497, 1508), False, 'import asyncio\n'), ((2189, 2227), 'argparse.ArgumentParser', 'argparse.ArgumentParser', (['"""fpds.client"""'], {}), "('fpds.client')\n", (2212, 2227), False, 'import argparse\n'), ((2385, 2409), 'asyncio.get_event_loop', 'asyncio.get_event_loop', ([], {}), '()\n', (2407, 2409), False, 'import asyncio\n'), ((1294, 1317), 'cv2.destroyAllWindows', 'cv2.destroyAllWindows', ([], {}), '()\n', (1315, 1317), False, 'import cv2\n'), ((472, 498), 'cv2.resize', 'cv2.resize', (['im', '(640, 384)'], {}), '(im, (640, 384))\n', (482, 498), False, 'import cv2\n'), ((1011, 1044), 'numpy.frombuffer', 'np.frombuffer', (['im'], {'dtype': 'np.uint8'}), '(im, dtype=np.uint8)\n', (1024, 1044), True, 'import numpy as np\n'), ((1160, 1197), 'cv2.imshow', 'cv2.imshow', (['"""fpds_remote_preview"""', 'im'], {}), "('fpds_remote_preview', im)\n", (1170, 1197), False, 'import cv2\n'), ((1210, 1224), 'cv2.waitKey', 'cv2.waitKey', (['(1)'], {}), '(1)\n', (1221, 1224), False, 'import cv2\n'), ((2051, 2074), 'aiohttp.ClientSession', 'aiohttp.ClientSession', ([], {}), '()\n', (2072, 2074), False, 'import aiohttp\n'), ((1892, 1926), 'asyncio.wait', 'asyncio.wait', (['[src_task, dst_task]'], {}), '([src_task, dst_task])\n', (1904, 1926), False, 'import asyncio\n'), ((2586, 2622), 'asyncio.wait_for', 'asyncio.wait_for', (['task'], {'timeout': 'None'}), '(task, timeout=None)\n', (2602, 2622), False, 'import asyncio\n'), ((340, 364), 'cv2.VideoCapture', 'cv2.VideoCapture', (['src_id'], {}), '(src_id)\n', (356, 364), False, 'import cv2\n'), ((626, 661), 'cv2.imencode', 'cv2.imencode', (['""".jpg"""', 'im', 'enc_param'], {}), "('.jpg', im, enc_param)\n", (638, 661), False, 'import cv2\n'), ((1109, 1146), 'cv2.imdecode', 'cv2.imdecode', (['im', 'cv2.IMREAD_ANYCOLOR'], {}), '(im, cv2.IMREAD_ANYCOLOR)\n', (1121, 1146), False, 'import cv2\n')]