notebooks/115-async-api/115-async-api.ipynb

{
 "cells": [
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "# Asynchronous Inference with OpenVINO™\n",
    "This notebook demonstrates how to use the [Async API](https://docs.openvino.ai/nightly/openvino_docs_deployment_optimization_guide_common.html) for asynchronous execution with OpenVINO.\n",
    "\n",
    "OpenVINO Runtime supports inference in either synchronous or asynchronous mode. The key advantage of the Async API is that when a device is busy with inference, the application can perform other tasks in parallel (for example, populating inputs or scheduling other requests) rather than wait for the current inference to complete first."
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Imports"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "import sys\n",
    "import cv2\n",
    "import time\n",
    "import numpy as np\n",
    "from openvino.runtime import Core, AsyncInferQueue\n",
    "import openvino.runtime as ov\n",
    "from IPython import display\n",
    "import matplotlib.pyplot as plt\n",
    "sys.path.append(\"../utils\")\n",
    "import notebook_utils as utils"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## Prepare model and data processing \n",
    "\n",
    "### Download test model\n",
    "We use a pre-trained model from OpenVINO's [Open Model Zoo](https://docs.openvino.ai/nightly/model_zoo.html) to start the test. In this case, the model will be executed to detect the person in each frame of the video."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# directory where model will be downloaded\n",
    "base_model_dir = \"model\"\n",
    "\n",
    "# model name as named in Open Model Zoo\n",
    "model_name = \"person-detection-0202\"\n",
    "precision = \"FP16\"\n",
    "model_path = (\n",
    "    f\"model/intel/{model_name}/{precision}/{model_name}.xml\"\n",
    ")\n",
    "download_command = f\"omz_downloader \" \\\n",
    "                   f\"--name {model_name} \" \\\n",
    "                   f\"--precision {precision} \" \\\n",
    "                   f\"--output_dir {base_model_dir} \" \\\n",
    "                   f\"--cache_dir {base_model_dir}\"\n",
    "! $download_command"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Load the model"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "# initialize OpenVINO runtime\n",
    "ie = Core()\n",
    "\n",
    "# read the network and corresponding weights from file\n",
    "model = ie.read_model(model=model_path)\n",
    "\n",
    "# compile the model for the CPU (you can choose manually CPU, GPU, MYRIAD etc.)\n",
    "# or let the engine choose the best available device (AUTO)\n",
    "compiled_model = ie.compile_model(model=model, device_name=\"CPU\")\n",
    "\n",
    "# get input node\n",
    "input_layer_ir = model.input(0)\n",
    "N, C, H, W = input_layer_ir.shape\n",
    "shape = (H, W)"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Create functions for data processing"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def preprocess(image):\n",
    "    \"\"\"\n",
    "    Define the preprocess function for input data\n",
    "    \n",
    "    :param: image: the orignal input frame\n",
    "    :returns:\n",
    "            resized_image: the image processed\n",
    "    \"\"\"\n",
    "    resized_image = cv2.resize(image, shape)\n",
    "    resized_image = cv2.cvtColor(np.array(resized_image), cv2.COLOR_BGR2RGB)\n",
    "    resized_image = resized_image.transpose((2, 0, 1))\n",
    "    resized_image = np.expand_dims(resized_image, axis=0).astype(np.float32)\n",
    "    return resized_image\n",
    "\n",
    "\n",
    "def postprocess(result, image, fps):\n",
    "    \"\"\"\n",
    "    Define the postprocess function for output data\n",
    "    \n",
    "    :param: result: the inference results\n",
    "            image: the orignal input frame\n",
    "            fps: average throughput calculated for each frame\n",
    "    :returns:\n",
    "            image: the image with bounding box and fps message\n",
    "    \"\"\"\n",
    "    detections = result.reshape(-1, 7)\n",
    "    for i, detection in enumerate(detections):\n",
    "        _, image_id, confidence, xmin, ymin, xmax, ymax = detection\n",
    "        if confidence > 0.5:\n",
    "            xmin = int(max((xmin * image.shape[1]), 10))\n",
    "            ymin = int(max((ymin * image.shape[0]), 10))\n",
    "            xmax = int(min((xmax * image.shape[1]), image.shape[1] - 10))\n",
    "            ymax = int(min((ymax * image.shape[0]), image.shape[0] - 10))\n",
    "            cv2.rectangle(image, (xmin, ymin), (xmax, ymax), (0, 255, 0), 2)\n",
    "            cv2.putText(image, str(round(fps, 2)) + \" fps\", (5, 20), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (0, 255, 0), 3) \n",
    "    return image"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Get the test video"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "video_path = \"../data/video/CEO Pat Gelsinger on Leading Intel.mp4\""
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## How to improve the throughput of video processing\n",
    "Below, we compare the performance of the synchronous and async-based approaches:\n",
    "\n",
    "### Sync Mode (default)\n",
    "Let us see how video processing works with the default approach.<br />\n",
    "Using the synchronous approach, the frame is captured with OpenCV and then immediately processed:\n",
    "\n",
    "<img align='center' src=\"https://user-images.githubusercontent.com/91237924/168452573-d354ea5b-7966-44e5-813d-f9053be4338a.png\" alt=\"drawing\" width=\"600\"/><br />\n",
    "\n",
    "```\n",
    "while(true) {\n",
    "// capture frame\n",
    "// populate CURRENT InferRequest\n",
    "// Infer CURRENT InferRequest\n",
    "//this call is synchronous\n",
    "// display CURRENT result\n",
    "}\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def sync_api(source, flip, fps, use_popup, skip_first_frames):\n",
    "    \"\"\"\n",
    "    Define the main function for video processing in sync mode\n",
    "    \n",
    "    :param: source: the video path or the ID of your webcam\n",
    "    :returns:\n",
    "            sync_fps: the inference throughput in sync mode\n",
    "    \"\"\"\n",
    "    frame_number = 0\n",
    "    infer_request = compiled_model.create_infer_request()\n",
    "    player = None\n",
    "    try:\n",
    "        # Create a video player\n",
    "        player = utils.VideoPlayer(source, flip=flip, fps=fps, skip_first_frames=skip_first_frames)\n",
    "        # Start capturing\n",
    "        start_time = time.time()\n",
    "        player.start()\n",
    "        if use_popup:\n",
    "            title = \"Press ESC to Exit\"\n",
    "            cv2.namedWindow(title, cv2.WINDOW_GUI_NORMAL | cv2.WINDOW_AUTOSIZE)\n",
    "        while True:\n",
    "            frame = player.next()\n",
    "            if frame is None:\n",
    "                print(\"Source ended\")\n",
    "                break\n",
    "            resized_frame = preprocess(frame)\n",
    "            infer_request.set_tensor(input_layer_ir, ov.Tensor(resized_frame))\n",
    "            # Start the inference request in synchronous mode \n",
    "            infer_request.infer()\n",
    "            res = infer_request.get_output_tensor(0).data\n",
    "            stop_time = time.time()\n",
    "            total_time = stop_time - start_time\n",
    "            frame_number = frame_number + 1\n",
    "            sync_fps = frame_number / total_time \n",
    "            frame = postprocess(res, frame, sync_fps)\n",
    "            # Display the results\n",
    "            if use_popup:\n",
    "                cv2.imshow(title, frame)\n",
    "                key = cv2.waitKey(1)\n",
    "                # escape = 27\n",
    "                if key == 27:\n",
    "                    break\n",
    "            else:\n",
    "                # Encode numpy array to jpg\n",
    "                _, encoded_img = cv2.imencode(\".jpg\", frame, params=[cv2.IMWRITE_JPEG_QUALITY, 90])\n",
    "                # Create IPython image\n",
    "                i = display.Image(data=encoded_img)\n",
    "                # Display the image in this notebook\n",
    "                display.clear_output(wait=True)\n",
    "                display.display(i)         \n",
    "    # ctrl-c\n",
    "    except KeyboardInterrupt:\n",
    "        print(\"Interrupted\")\n",
    "    # Any different error\n",
    "    except RuntimeError as e:\n",
    "        print(e)\n",
    "    finally:\n",
    "        if use_popup:\n",
    "            cv2.destroyAllWindows()\n",
    "        if player is not None:\n",
    "            # stop capturing\n",
    "            player.stop()\n",
    "        return sync_fps"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Test performance in Sync Mode"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "sync_fps = sync_api(source=video_path, flip=False, fps=30, use_popup=False, skip_first_frames=800)\n",
    "print(f\"average throuput in sync mode: {sync_fps:.2f} fps\")"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Async Mode \n",
    "\n",
    "Let us see how the OpenVINO Async API can improve the overall frame rate of an application. The key advantage of the Async approach is as follows: while a device is busy with the inference, the application can do other things in parallel (for example, populating inputs or scheduling other requests) rather than wait for the current inference to complete first.\n",
    "\n",
    "<img align='center' src=\"https://user-images.githubusercontent.com/91237924/168452572-c2ff1c59-d470-4b85-b1f6-b6e1dac9540e.png\" alt=\"drawing\" width=\"600\"/><br />\n",
    "\n",
    "In the example below, inference is applied to the results of the video decoding. So it is possible to keep multiple infer requests, and while the current request is processed, the input frame for the next is being captured. This essentially hides the latency of capturing, so that the overall frame rate is rather determined only by the slowest part of the pipeline (decoding vs inference) and not by the sum of the stages.\n",
    "\n",
    "```\n",
    "while(true) {\n",
    "// capture frame\n",
    "// populate NEXT InferRequest\n",
    "// start NEXT InferRequest\n",
    "// this call is async and returns immediately\n",
    "// wait for the CURRENT InferRequest\n",
    "// display CURRENT result\n",
    "// swap CURRENT and NEXT InferRequests\n",
    "}\n",
    "```"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def async_api(source, flip, fps, use_popup, skip_first_frames):\n",
    "    \"\"\"\n",
    "    Define the main function for video processing in async mode\n",
    "    \n",
    "    :param: source: the video path or the ID of your webcam\n",
    "    :returns:\n",
    "            async_fps: the inference throughput in async mode\n",
    "    \"\"\"\n",
    "    frame_number = 0\n",
    "    # Create 2 infer requests\n",
    "    curr_request = compiled_model.create_infer_request()\n",
    "    next_request = compiled_model.create_infer_request()\n",
    "    player = None\n",
    "    try:\n",
    "        # Create a video player\n",
    "        player = utils.VideoPlayer(source, flip=flip, fps=fps, skip_first_frames=skip_first_frames)\n",
    "        # Start capturing\n",
    "        start_time = time.time()\n",
    "        player.start()\n",
    "        if use_popup:\n",
    "            title = \"Press ESC to Exit\"\n",
    "            cv2.namedWindow(title, cv2.WINDOW_GUI_NORMAL | cv2.WINDOW_AUTOSIZE)\n",
    "        # Capture CURRENT frame\n",
    "        frame = player.next()\n",
    "        resized_frame = preprocess(frame)\n",
    "        curr_request.set_tensor(input_layer_ir, ov.Tensor(resized_frame))\n",
    "        # Start the CURRENT inference request\n",
    "        curr_request.start_async()\n",
    "        while True:\n",
    "            # Capture NEXT frame\n",
    "            next_frame = player.next()\n",
    "            if next_frame is None:\n",
    "                print(\"Source ended\")\n",
    "                break\n",
    "            resized_frame = preprocess(next_frame)\n",
    "            next_request.set_tensor(input_layer_ir, ov.Tensor(resized_frame))\n",
    "            # Start the NEXT inference request\n",
    "            next_request.start_async()\n",
    "            # Waiting for CURRENT inference result\n",
    "            if curr_request.wait_for(-1) == 1:\n",
    "                res = curr_request.get_output_tensor(0).data\n",
    "                stop_time = time.time()\n",
    "                total_time = stop_time - start_time\n",
    "                frame_number = frame_number + 1\n",
    "                async_fps = frame_number / total_time  \n",
    "                frame = postprocess(res, frame, async_fps)\n",
    "                # Display the results\n",
    "                if use_popup:\n",
    "                    cv2.imshow(title, frame)\n",
    "                    key = cv2.waitKey(1)\n",
    "                    # escape = 27\n",
    "                    if key == 27:\n",
    "                        break\n",
    "                else:\n",
    "                    # Encode numpy array to jpg\n",
    "                    _, encoded_img = cv2.imencode(\".jpg\", frame, params=[cv2.IMWRITE_JPEG_QUALITY, 90])\n",
    "                    # Create IPython image\n",
    "                    i = display.Image(data=encoded_img)\n",
    "                    # Display the image in this notebook\n",
    "                    display.clear_output(wait=True)\n",
    "                    display.display(i)\n",
    "            # Swap CURRENT and NEXT frames\n",
    "            frame = next_frame\n",
    "            # Swap CURRENT and NEXT infer requests\n",
    "            curr_request, next_request = next_request, curr_request         \n",
    "    # ctrl-c\n",
    "    except KeyboardInterrupt:\n",
    "        print(\"Interrupted\")\n",
    "    # Any different error\n",
    "    except RuntimeError as e:\n",
    "        print(e)\n",
    "    finally:\n",
    "        if use_popup:\n",
    "            cv2.destroyAllWindows()\n",
    "        if player is not None:\n",
    "            # stop capturing\n",
    "            player.stop()\n",
    "        return async_fps"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Test the performance in Async Mode"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "async_fps = async_api(source=video_path, flip=False, fps=30, use_popup=False, skip_first_frames=800)\n",
    "print(f\"average throuput in async mode: {async_fps:.2f} fps\")"
   ]
  },
  {
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Compare the performance"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "width = 0.4\n",
    "fontsize = 14\n",
    "\n",
    "plt.rc('font', size=fontsize)\n",
    "fig, ax = plt.subplots(1, 1, figsize=(10, 8))\n",
    "\n",
    "rects1 = ax.bar([0], sync_fps, width, color='#557f2d')\n",
    "rects2 = ax.bar([width], async_fps, width)\n",
    "ax.set_ylabel(\"frames per second\")\n",
    "ax.set_xticks([0, width]) \n",
    "ax.set_xticklabels([\"Sync mode\", \"Async mode\"])\n",
    "ax.set_xlabel(\"Higher is better\")\n",
    "\n",
    "fig.suptitle('Sync mode VS Async mode')\n",
    "fig.tight_layout()\n",
    "\n",
    "plt.show()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "## AsyncInferQueue\n",
    "\n",
    "Asynchronous mode pipelines can be supported with the [AsyncInferQueue](https://docs.openvino.ai/latest/openvino_docs_OV_UG_Python_API_exclusives.html#asyncinferqueue) wrapper class. This class automatically spawns the pool of InferRequest objects (also called “jobs”) and provides synchronization mechanisms to control the flow of the pipeline. It is a simpler way to manage the infer request queue in Asynchronous mode."
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Setting Callback\n",
    "\n",
    "When `callback` is set, any job that ends inference calls upon the Python function. The `callback` function must have two arguments: one is the request that calls the `callback`, which provides the InferRequest API; the other is called “userdata”, which provides the possibility of passing runtime values."
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def callback(infer_request, info) -> None:\n",
    "    \"\"\"\n",
    "    Define the callback function for postprocessing\n",
    "    \n",
    "    :param: infer_request: the infer_request object\n",
    "            info: a tuple includes original frame and starts time\n",
    "    :returns:\n",
    "            None\n",
    "    \"\"\"\n",
    "    global frame_number\n",
    "    global total_time\n",
    "    global inferqueue_fps\n",
    "    stop_time = time.time()\n",
    "    frame, start_time = info\n",
    "    total_time = stop_time - start_time\n",
    "    frame_number = frame_number + 1\n",
    "    inferqueue_fps = frame_number / total_time\n",
    "    \n",
    "    res = infer_request.get_output_tensor(0).data[0]\n",
    "    frame = postprocess(res, frame, inferqueue_fps)\n",
    "    # Encode numpy array to jpg\n",
    "    _, encoded_img = cv2.imencode(\".jpg\", frame, params=[cv2.IMWRITE_JPEG_QUALITY, 90])\n",
    "    # Create IPython image\n",
    "    i = display.Image(data=encoded_img)\n",
    "    # Display the image in this notebook\n",
    "    display.clear_output(wait=True)\n",
    "    display.display(i)"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def inferqueue(source, flip, fps, skip_first_frames) -> None:\n",
    "    \"\"\"\n",
    "    Define the main function for video processing with async infer queue\n",
    "    \n",
    "    :param: source: the video path or the ID of your webcam\n",
    "    :retuns:\n",
    "        None\n",
    "    \"\"\"\n",
    "    # Create infer requests queue\n",
    "    infer_queue = AsyncInferQueue(compiled_model, 2)\n",
    "    infer_queue.set_callback(callback)\n",
    "    player = None\n",
    "    try:\n",
    "        # Create a video player\n",
    "        player = utils.VideoPlayer(source, flip=flip, fps=fps, skip_first_frames=skip_first_frames)\n",
    "        # Start capturing\n",
    "        start_time = time.time()\n",
    "        player.start()\n",
    "        while True:\n",
    "            # Capture frame\n",
    "            frame = player.next()\n",
    "            if frame is None:\n",
    "                print(\"Source ended\")\n",
    "                break\n",
    "            resized_frame = preprocess(frame)\n",
    "            # Start the inference request with async infer queue \n",
    "            infer_queue.start_async({input_layer_ir.any_name: resized_frame}, (frame, start_time))\n",
    "    except KeyboardInterrupt:\n",
    "        print(\"Interrupted\")\n",
    "    # Any different error\n",
    "    except RuntimeError as e:\n",
    "        print(e)\n",
    "    finally:\n",
    "        infer_queue.wait_all()\n",
    "        player.stop()"
   ]
  },
  {
   "attachments": {},
   "cell_type": "markdown",
   "metadata": {},
   "source": [
    "### Test the performance with AsyncInferQueue"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "frame_number = 0\n",
    "total_time = 0\n",
    "inferqueue(source=video_path, flip=False, fps=30, skip_first_frames=800)\n",
    "print(f\"average throughput in async mode with async infer queue: {inferqueue_fps:.2f} fps\")"
   ]
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3.7.9 64-bit",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.7.9"
  },
  "vscode": {
   "interpreter": {
    "hash": "08af04953a73b86b66cc089a637d3d397b0b73ad05ea59846f770cc21ccdacba"
   }
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}