demo.ipynb

{
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/nateraw/openai-vision-api-for-videos/blob/main/demo.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "l2TWhQUvSSkO"
      },
      "source": [
        "# Trying OpenAI's Vision API on Videos\n",
        "\n",
        "OpenAI just launched their [Vision API](https://platform.openai.com/docs/guides/vision) - a multimodal LLM that understands language and images that acts as a helpful assistant.\n",
        "\n",
        "The debut model,  `gpt-4-vision-preview`, has a whopping [128k context window](https://platform.openai.com/docs/models/gpt-4-and-gpt-4-turbo) 🤯. This means that in one pass of the model, it can process up to 128,000 tokens.\n",
        "\n",
        "In this notebook, we'll put that context to use by passing frames from videos sequentially into it, and see what it can do.\n",
        "\n",
        "# If you find this notebook helpful...\n",
        "\n",
        "Please consider supporting me by:\n",
        "\n",
        "- giving [the repo](https://github.com/nateraw/openai-vision-api-for-videos) a ⭐️\n",
        "- Following me on [GitHub](https://github.com/nateraw) and/or [Twitter](https://x.com/_nateraw). ❤️\n",
        "\n",
        "Share issues/feature requests [here](https://github.com/nateraw/openai-vision-api-for-videos/issues)."
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "vjqShaFRUYjh"
      },
      "source": [
        "<details>\n",
        "  <summary><strong>⚠️ Note About Costs (Click to expand)</strong></summary>\n",
        "  <p>Pricing is done by token, just like with language models. However, Vision tokens are counted differently. The amount of tokens a single image is billed for depends on a wide variety of factors. This is outside of the scope of this notebook, but I encourage you to read <a href=\"https://platform.openai.com/docs/guides/vision/calculating-costs\">this rather complicated explanation</a> for more info so you (hopefully 😅) understand how you'll be billed for the Vision API.</p>\n",
        "\n",
        "  <p>You should be mindful of your spending, and make sure you have a spend limit set to avoid any bank-draining whoopsies.</p>\n",
        "\n",
        "</details>"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "SjmK-DmvLnPs"
      },
      "source": [
        "# Install dependencies"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "xQaz5E_WT8km"
      },
      "outputs": [],
      "source": [
        "%%capture\n",
        "! pip install openai yt-dlp av replicate"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "KFzILElgODI7"
      },
      "source": [
        "# Setup Env\n",
        "\n",
        "<!-- TODO maybe use this instead of form -->\n",
        "On the left panel, select the 🔑 icon to add secrets:\n",
        "\n",
        "- `OPENAI_API_KEY`: Key from OpenAI, which you can find [here](https://platform.openai.com/account/api-keys) when logged in.\n",
        "- `REPLICATE_API_TOKEN`: (only required for transcription) A key from your account on Replicate, which can be found [here](https://replicate.com/account/api-tokens) if logged in.\n",
        "\n",
        "<img src=\"https://huggingface.co/datasets/nateraw/documentation-images/resolve/main/Screen%20Shot%202023-11-07%20at%202.38.30%20AM.png?download=true\">"
      ]
    },
    {
      "cell_type": "code",
      "source": [
        "import os\n",
        "from google.colab import userdata\n",
        "\n",
        "os.environ[\"OPENAI_API_KEY\"] = userdata.get(\"OPENAI_API_KEY\")\n",
        "if not os.environ.get(\"OPENAI_API_KEY\"):\n",
        "    raise ValueError(\"OPENAI_API_KEY not set\")\n",
        "\n",
        "os.environ[\"REPLICATE_API_TOKEN\"] = userdata.get(\"REPLICATE_API_TOKEN\")"
      ],
      "metadata": {
        "id": "amILf9cxOm6I"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "BOOiuvJJLqZX"
      },
      "source": [
        "# Download YouTube video to analyze\n",
        "\n",
        "Note the format of the URL is:\n",
        "\n",
        "`https://www.youtube.com/watch?v=<YouTube-ID>`\n",
        "\n",
        "If you use a different format, you might have a bad time."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "cellView": "form",
        "id": "buO-SHt6IhyE"
      },
      "outputs": [],
      "source": [
        "import subprocess\n",
        "import re\n",
        "from pathlib import Path\n",
        "from datetime import datetime\n",
        "\n",
        "from IPython.display import YouTubeVideo\n",
        "\n",
        "youtube_url = \"https://www.youtube.com/watch?v=S60GxA9JpLk\"  # @param {type:\"string\"}\n",
        "data_dir = Path(\"./videos\")\n",
        "data_dir.mkdir(exist_ok=True)\n",
        "\n",
        "# check for existing matches based on the YTID\n",
        "ytid = re.search(r'v=(.*)', youtube_url).group(1)\n",
        "matches = list(data_dir.glob(f'*_{ytid}_*'))\n",
        "if matches:\n",
        "    filepath = Path(matches[-1])\n",
        "    print(f\"Found existing file {filepath}\")\n",
        "else:\n",
        "    datetime_prefix = datetime.now().strftime(\"%Y%m%d%H%M%S\")\n",
        "    output_template = str(data_dir / f'{datetime_prefix}_%(id)s_%(title)s.%(ext)s')\n",
        "\n",
        "    command = [\n",
        "        'yt-dlp',\n",
        "        '-f', 'best[ext=mp4]',\n",
        "        '-o', output_template,\n",
        "        youtube_url\n",
        "    ]\n",
        "\n",
        "    process = subprocess.run(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, universal_newlines=True)\n",
        "    filepath = next(data_dir.glob(f'{datetime_prefix}*.mp4'))\n",
        "    print(f\"Downloaded video path: {filepath}\")\n",
        "\n",
        "\n",
        "YouTubeVideo(ytid)"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "hb1L1fPqazBB"
      },
      "source": [
        "## Define Utilities for Extracting Frames / Hitting the API\n",
        "\n",
        "#### ⚠️ Note\n",
        "You could write this with WAY less code with the `decord` library if you just want to load frames.\n",
        "\n",
        "I'm using `av` because it handles audio + video together quite well, which will be helpful when I extend the ideas in this notebook in the future. Some of this code is modified from [pytorchvideo](https://github.com/facebookresearch/pytorchvideo)."
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "BGeZvuM2bjhw"
      },
      "outputs": [],
      "source": [
        "# @title\n",
        "%matplotlib inline\n",
        "\n",
        "import av\n",
        "import gc\n",
        "from typing import Tuple, List\n",
        "import math\n",
        "import numpy as np\n",
        "from fractions import Fraction\n",
        "import subprocess\n",
        "from pathlib import Path\n",
        "import matplotlib.pyplot as plt\n",
        "import cv2\n",
        "import numpy as np\n",
        "import math\n",
        "from typing import Tuple, Union\n",
        "import base64\n",
        "from PIL import Image\n",
        "from io import BytesIO\n",
        "\n",
        "import openai\n",
        "\n",
        "\n",
        "def display_frames_as_grid(frames, rows=5, cols=5):\n",
        "    # Adjust the number of rows to ensure there are no completely empty rows\n",
        "    total_frames = frames.shape[0]\n",
        "    while rows > 1 and total_frames <= (rows - 1) * cols:\n",
        "        rows -= 1\n",
        "\n",
        "    # Initialize a figure with the adjusted number of rows\n",
        "    fig, axes = plt.subplots(rows, cols, figsize=(cols * 4, rows * 4))\n",
        "\n",
        "    # If there's only one row or column, axes might not be a 2D array\n",
        "    if rows == 1 or cols == 1:\n",
        "        axes = np.array(axes).reshape(-1)\n",
        "\n",
        "    # Flatten axes array for easy indexing\n",
        "    axes = axes.flatten()\n",
        "\n",
        "    # Loop through the grid and add frames with indices\n",
        "    for i in range(rows * cols):\n",
        "        ax = axes[i]\n",
        "        if i < total_frames:  # Check if the current frame exists\n",
        "            ax.imshow(frames[i])\n",
        "            ax.text(0.05, 0.95, f'Frame {i}', color='white', weight='bold',\n",
        "                    transform=ax.transAxes, ha=\"left\", va=\"top\", fontsize=8)\n",
        "        ax.axis('off')  # Hide axes\n",
        "\n",
        "    plt.tight_layout()\n",
        "    plt.show()\n",
        "\n",
        "\n",
        "def _pyav_decode_stream(\n",
        "    container: av.container.input.InputContainer,\n",
        "    start_sec: float,\n",
        "    end_sec: float,\n",
        "    stream: av.video.stream.VideoStream,\n",
        "    perform_seek: bool = True,\n",
        ") -> Tuple[List, float]:\n",
        "    \"\"\"\n",
        "    Decode the video with PyAV decoder.\n",
        "    Args:\n",
        "        container (container): PyAV container.\n",
        "        start_sec (float): the starting second to fetch the frames.\n",
        "        end_sec (float): the ending second of the decoded frames.\n",
        "        stream (stream): PyAV stream.\n",
        "    Returns:\n",
        "        result (np.ndarray): np array of decoded frames.\n",
        "    \"\"\"\n",
        "\n",
        "    start_pts = math.ceil(start_sec / stream.time_base)\n",
        "    end_pts = math.ceil(end_sec / stream.time_base)\n",
        "    # NOTE:\n",
        "    # Don't want to seek if iterating through a video due to slow-downs. I\n",
        "    # believe this is some PyAV bug where seeking after a certain point causes\n",
        "    # major slow-downs\n",
        "    if perform_seek:\n",
        "        # Seeking in the stream is imprecise. Thus, seek to an earlier pts by a\n",
        "        # margin pts.\n",
        "        margin = 1024\n",
        "        seek_offset = max(start_pts - margin, 0)\n",
        "        container.seek(int(seek_offset), any_frame=False, backward=True, stream=stream)\n",
        "\n",
        "    frames = []\n",
        "    stream_info = {'video': 0} if stream.type == 'video' else {'audio': 0}\n",
        "    for frame in container.decode(**stream_info):\n",
        "        if frame.pts >= start_pts and frame.pts < end_pts:\n",
        "            frames.append(frame)\n",
        "        elif frame.pts >= end_pts:\n",
        "            break\n",
        "\n",
        "    if stream.type == 'audio':\n",
        "        return np.concatenate([x.to_ndarray() for x in frames], 1)\n",
        "    else:\n",
        "        return np.stack([x.to_ndarray(format='rgb24') for x in frames])\n",
        "\n",
        "\n",
        "class EncodedVideo(object):\n",
        "    \"\"\"\n",
        "    wrapper around the _pyav_decode_stream fn that keeps the container open while we decode clips.\n",
        "\n",
        "    Inspired by PyTorchVideo's EncodedVideo, but without the torch dependency.\n",
        "    \"\"\"\n",
        "\n",
        "    def __init__(self, container_or_path: Union[str, av.container.input.InputContainer]):\n",
        "        if isinstance(container_or_path, (str, Path)):\n",
        "            self.container = av.open(str(container_or_path))\n",
        "        else:\n",
        "            self.container = container_or_path\n",
        "\n",
        "    def get_clip(self, start_sec, end_sec, decode_audio=True, perform_seek=True):\n",
        "        # Handle video stream\n",
        "        video_stream = self.container.streams.video[0]\n",
        "        info = dict(video_fps=video_stream.average_rate)\n",
        "        video_arr = _pyav_decode_stream(self.container, start_sec, end_sec, video_stream, perform_seek)\n",
        "\n",
        "        # Handle audio stream if desired\n",
        "        audio_arr = None\n",
        "        if decode_audio:\n",
        "            audio_stream = self.container.streams.audio[0]\n",
        "            info['audio_sample_rate'] = audio_stream.rate\n",
        "            audio_arr = _pyav_decode_stream(self.container, start_sec, end_sec, audio_stream, perform_seek)\n",
        "\n",
        "        return video_arr, audio_arr, info\n",
        "\n",
        "    def close(self):\n",
        "        self.container.close()\n",
        "        gc.collect()\n",
        "\n",
        "    def __del__(self):\n",
        "        self.close()\n",
        "\n",
        "\n",
        "# translated to np from pytorchvideo\n",
        "def uniform_temporal_subsample(x: np.ndarray, num_samples: int, temporal_dim: int = -3) -> np.ndarray:\n",
        "    \"\"\"\n",
        "    Uniformly subsamples num_samples indices from the temporal dimension of the video.\n",
        "    When num_samples is larger than the size of temporal dimension of the video, it\n",
        "    will sample frames based on nearest neighbor interpolation.\n",
        "    Args:\n",
        "        x (np.ndarray): A video tensor with dimension larger than one with numpy\n",
        "            array type includes int, long, float, complex, etc.\n",
        "        num_samples (int): The number of equispaced samples to be selected\n",
        "        temporal_dim (int): dimension of temporal to perform temporal subsample.\n",
        "    Returns:\n",
        "        An x-like array with subsampled temporal dimension.\n",
        "    \"\"\"\n",
        "    # Adjust the temporal_dim to work with numpy's axis definition\n",
        "    if temporal_dim < 0:\n",
        "        temporal_dim += x.ndim\n",
        "\n",
        "    t = x.shape[temporal_dim]\n",
        "    assert num_samples > 0 and t > 0\n",
        "    # Sample by nearest neighbor interpolation if num_samples > t.\n",
        "    indices = np.linspace(0, t - 1, num_samples)\n",
        "    indices = np.clip(indices, 0, t - 1).astype(int)\n",
        "    # Use advanced integer indexing to select the frames\n",
        "    return np.take(x, indices, axis=temporal_dim)\n",
        "\n",
        "\n",
        "# translated to numpy from pytorchvideo\n",
        "def _interpolate_opencv(\n",
        "    x: np.ndarray, size: Tuple[int, int], interpolation: str\n",
        ") -> np.ndarray:\n",
        "    \"\"\"\n",
        "    Down/up samples the input numpy array x to the given size with given interpolation\n",
        "    mode.\n",
        "    Args:\n",
        "        x (np.ndarray): the input array to be down/up sampled.\n",
        "        size (Tuple[int, int]): expected output spatial size.\n",
        "        interpolation: mode to perform interpolation, options include `nearest`,\n",
        "            `linear`, `bilinear`, `bicubic`.\n",
        "    \"\"\"\n",
        "    _opencv_np_interpolation_map = {\n",
        "        \"nearest\": cv2.INTER_NEAREST,\n",
        "        \"linear\": cv2.INTER_LINEAR,\n",
        "        \"bilinear\": cv2.INTER_LINEAR,\n",
        "        \"bicubic\": cv2.INTER_CUBIC,\n",
        "    }\n",
        "    assert interpolation in _opencv_np_interpolation_map, \"Invalid interpolation mode.\"\n",
        "    new_h, new_w = size\n",
        "    # Transpose to shape (H, W, C, T)\n",
        "    x = x.transpose(2, 3, 0, 1)\n",
        "    resized_array_list = [\n",
        "        cv2.resize(\n",
        "            x[:, :, :, t],\n",
        "            (new_w, new_h),\n",
        "            interpolation=_opencv_np_interpolation_map[interpolation],\n",
        "        )\n",
        "        for t in range(x.shape[3])\n",
        "    ]\n",
        "    # Stack on the last dimension and then transpose back to (C, T, H, W)\n",
        "    img_array = np.stack(resized_array_list, axis=-1)\n",
        "    return img_array.transpose(2, 3, 0, 1)\n",
        "\n",
        "\n",
        "# translated to np from pytorchvideo\n",
        "def short_side_scale(\n",
        "    x: np.ndarray,\n",
        "    size: int,\n",
        "    interpolation: str = \"bilinear\"\n",
        ") -> np.ndarray:\n",
        "    \"\"\"\n",
        "    Determines the shorter spatial dim of the input (i.e. width or height) and scales\n",
        "    it to the given size. To maintain aspect ratio, the longer side is then scaled\n",
        "    accordingly.\n",
        "    Args:\n",
        "        x (np.ndarray): An array of shape (C, T, H, W).\n",
        "        size (int): The size the shorter side is scaled to.\n",
        "        interpolation (str): Algorithm used for upsampling,\n",
        "            options: 'nearest' | 'linear' | 'bilinear' | 'bicubic'\n",
        "    Returns:\n",
        "        A NumPy array with scaled spatial dims.\n",
        "    \"\"\"\n",
        "    assert len(x.shape) == 4, \"Input must be a 4D array.\"\n",
        "    c, t, h, w = x.shape\n",
        "    if w < h:\n",
        "        new_h = int(math.floor((float(h) / w) * size))\n",
        "        new_w = size\n",
        "    else:\n",
        "        new_h = size\n",
        "        new_w = int(math.floor((float(w) / h) * size))\n",
        "\n",
        "    return _interpolate_opencv(x, size=(new_h, new_w), interpolation=interpolation)\n",
        "\n",
        "\n",
        "def transform(\n",
        "    video_arr,\n",
        "    num_frames: int,\n",
        "    short_side_size: int,\n",
        "):\n",
        "    # T, H, W, C -> C, T, H, W\n",
        "    # uint8 [0, 255] -> float32 [0.0, 1.0]\n",
        "    video_arr = np.transpose(video_arr, (3, 0, 1, 2)).astype(np.float32) / 255.\n",
        "\n",
        "    # Sample T down to num_frames\n",
        "    video_arr = uniform_temporal_subsample(video_arr, num_frames)\n",
        "\n",
        "    # Scale short size of video to short_side_size\n",
        "    video_arr = short_side_scale(video_arr, short_side_size)\n",
        "\n",
        "    # Rescale back to [0, 255] and convert to uint8\n",
        "    video_arr *= 255\n",
        "    video_arr = np.transpose(video_arr.astype(np.uint8), (1, 2, 3, 0))\n",
        "    gc.collect()\n",
        "    return video_arr\n",
        "\n",
        "# Function to encode a numpy array to base64\n",
        "def encode_numpy_image(np_img):\n",
        "    \"\"\"Takes in a np img array and returns a base64 encoded string\"\"\"\n",
        "    pil_img = Image.fromarray(np_img.astype('uint8'), 'RGB')\n",
        "    buffer = BytesIO()\n",
        "    pil_img.save(buffer, format='JPEG')\n",
        "    img_str = buffer.getvalue()\n",
        "    return base64.b64encode(img_str).decode('utf-8')\n",
        "\n",
        "\n",
        "def call_vision_api(frames, prompt: str, max_tokens=200, resize=768, **completion_kwargs):\n",
        "    \"\"\"\n",
        "    frames should be np array with shape (T, H, W, C), dtype np.uint8\n",
        "    prompt should be a string that will be used as prompt alongside the frames.\n",
        "    max_tokens should be an int\n",
        "    \"\"\"\n",
        "    client = openai.Client()\n",
        "\n",
        "    # Encode all frames to base64\n",
        "    base64_frames = [encode_numpy_image(frame) for frame in frames]\n",
        "\n",
        "    # Prepare the content for the prompt\n",
        "    content = [\n",
        "        prompt,\n",
        "        *map(lambda x: {\"image\": x, \"resize\": resize}, base64_frames)\n",
        "    ]\n",
        "\n",
        "    # Prepare the prompt message\n",
        "    prompt_messages = [\n",
        "        {\n",
        "            \"role\": \"user\",\n",
        "            \"content\": content,\n",
        "        },\n",
        "    ]\n",
        "\n",
        "    # Set up the parameters for the API call\n",
        "    params = {\n",
        "        \"model\": \"gpt-4-vision-preview\",\n",
        "        \"messages\": prompt_messages,\n",
        "        \"max_tokens\": max_tokens,\n",
        "        **completion_kwargs,\n",
        "    }\n",
        "\n",
        "    # Make the API call\n",
        "    result = client.chat.completions.create(**params)\n",
        "\n",
        "    # Return the generated description\n",
        "    return result.choices[0].message.content"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Run it!"
      ],
      "metadata": {
        "id": "YQ7ve3J_F4v0"
      }
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "glHvWMDPowGH"
      },
      "outputs": [],
      "source": [
        "start_sec = 55  # @param {type:\"integer\"}\n",
        "duration = 30   # @param {type:\"integer\"}\n",
        "end_sec = start_sec + duration\n",
        "num_frames_to_sample = 15  # @param {type:\"integer\"}\n",
        "short_side_size = 512  # @param {type:\"integer\"}\n",
        "max_tokens = 512  # @param {type:\"integer\"}\n",
        "prompt = \"These are frames from a video. Generate a description/summary that explains what is happening based on what you can see.\\nYour response should **STRICTLY and ONLY** be the description, no titles, headers, or anything else.\" # @param {type: \"string\",description:\"leav blank\"}\n",
        "show_frame_grid = True  # @param {type:\"boolean\"}\n",
        "\n",
        "video_arr, audio_arr, info = EncodedVideo(filepath).get_clip(start_sec, end_sec)\n",
        "print(f\"Original video shape: {video_arr.shape}\")\n",
        "\n",
        "video_arr = transform(video_arr, num_frames_to_sample, short_side_size)\n",
        "print(f\"Final video frames shape: {video_arr.shape}\")\n",
        "\n",
        "print('-' * 80)\n",
        "print()\n",
        "\n",
        "if show_frame_grid:\n",
        "    display_frames_as_grid(video_arr, rows=30, cols=5)\n",
        "\n",
        "out = call_vision_api(video_arr, prompt, max_tokens=max_tokens)\n",
        "out"
      ]
    },
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "jHI4HuzUAhI-"
      },
      "source": [
        "# Incorporating audio transcriptions\n",
        "\n",
        "Let's not forget we have audio we can use too!"
      ]
    },
    {
      "cell_type": "code",
      "execution_count": null,
      "metadata": {
        "id": "RQ_9SejN2als"
      },
      "outputs": [],
      "source": [
        "from IPython.display import Audio\n",
        "\n",
        "Audio(audio_arr, rate=info[\"audio_sample_rate\"])"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "Here we'll try using [WhisperX](https://github.com/m-bain/whisperX) to grab the text transcription of the video clip and see if including that as context in our prompt improves the video summary.\n",
        "\n",
        "We'll use this model via [Replicate's API](https://replicate.com/zeke/zisper) thanks to [@zeke](https://github.com/zeke). ❤️\n",
        "\n",
        "If you want to run the model directly, see the underlying [code for the API here](https://github.com/zeke/zisper)! 👀"
      ],
      "metadata": {
        "id": "TUBhyx_aCNdw"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import os\n",
        "from google.colab import userdata\n",
        "\n",
        "###############################################################################\n",
        "# Token should be set as REPLICATE_API_TOKEN in the Secrets section of Colab\n",
        "# Click the 🔑 icon on the left sidebar to add it/check if you already added it\n",
        "###############################################################################\n",
        "\n",
        "replicate_api_token = userdata.get(\"REPLICATE_API_TOKEN\")\n",
        "os.environ[\"REPLICATE_API_TOKEN\"] = replicate_api_token"
      ],
      "metadata": {
        "id": "0kTjrHLBPUeI"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "Note - you might run into a [cold boot](https://replicate.com/docs/how-does-replicate-work#cold-boots) if the model hasn't been used in a while. Once it boots up, it'll respond much faster in subsequent runs."
      ],
      "metadata": {
        "id": "AJ14hPiCYlFb"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "import json\n",
        "import time\n",
        "import io\n",
        "from scipy.io.wavfile import write\n",
        "import replicate\n",
        "\n",
        "\n",
        "client = replicate.Client(replicate_api_token)\n",
        "\n",
        "model = client.models.get(\"zeke/zisper\")\n",
        "version = model.latest_version\n",
        "\n",
        "audio_file_to_transcribe = \"audio.wav\"\n",
        "write(audio_file_to_transcribe, info[\"audio_sample_rate\"], audio_arr.T)\n",
        "\n",
        "prediction = client.predictions.create(\n",
        "    version=version,\n",
        "    input=dict(\n",
        "        audio=open(audio_file_to_transcribe, 'rb'),\n",
        "        debug=False,\n",
        "        batch_size=2,\n",
        "        only_text=False,\n",
        "        align_output=False\n",
        "    )\n",
        ")\n",
        "\n",
        "\n",
        "while prediction.completed_at is None:\n",
        "    prediction.reload()\n",
        "    prediction.status\n",
        "    time.sleep(1)\n",
        "\n",
        "# Warning, haven't tried on responses with multiple chunks, may be wrong join logic here\n",
        "transcription = [x['text'] for x in json.loads(prediction.output)]\n",
        "transcription = \"\".join(transcription).strip()\n",
        "transcription"
      ],
      "metadata": {
        "id": "SAu-nH-oB8Ym"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Hit the Vision API with the images AND transcription from the video"
      ],
      "metadata": {
        "id": "UAW0_GVSLwB_"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "prompt_with_transcriptions = f\"\"\"\\\n",
        "These are frames from a video. Generate a description/summary that explains what is happening based on what you can see, as well as the video's audio transcription below.\n",
        "\n",
        "Transcription: {transcription}\n",
        "\n",
        "Your response should **STRICTLY and ONLY** be the summary, no titles, headers, or anything else.\\\n",
        "\"\"\"\n",
        "\n",
        "out = call_vision_api(\n",
        "    video_arr,\n",
        "    prompt_with_transcriptions,\n",
        "    max_tokens=max_tokens\n",
        ")\n",
        "out"
      ],
      "metadata": {
        "id": "d0ZiIvXJCccg"
      },
      "execution_count": null,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Conclusion\n",
        "\n",
        "There's a LOT more you can do with this API. I'll list some ideas below you can go try for yourself :)\n",
        "\n",
        "I'll try to keep updating this notebook as I find time. Feel free to drop issues/feature requests on GitHub [here](https://github.com/nateraw/openai-vision-api-for-videos/issues).\n",
        "\n",
        "#### Some ideas:\n",
        "\n",
        "- Set up a live webcam and have the model answer questions about what you're doing\n",
        "  - 🔥 implemented [here](https://twitter.com/skalskip92/status/1721694286440468849) by [Piotr Skalski](https://github.com/SkalskiP)\n",
        "- Embed summaries of clips from videos and do video search (on one or a LOT of videos)\n",
        "- Use the API to extract good frames from videos with a characteristic style, then use them to fine-tune a image generative model, such as SDXL.\n",
        "- Use workflow similar to this one to write executive summaries of Ted Talks/Podcasts, or perhaps write review notes for lectures.\n",
        "\n",
        "# If you found this notebook helpful...\n",
        "\n",
        "Please consider supporting me by:\n",
        "\n",
        "- giving [the repo](https://github.com/nateraw/openai-vision-api-for-videos) a ⭐️\n",
        "- Following me on [GitHub](https://github.com/nateraw) and/or [Twitter](https://x.com/_nateraw). ❤️\n",
        "\n"
      ],
      "metadata": {
        "id": "0pvL2U2zPsJH"
      }
    }
  ],
  "metadata": {
    "accelerator": "GPU",
    "colab": {
      "machine_shape": "hm",
      "provenance": [],
      "collapsed_sections": [
        "hb1L1fPqazBB"
      ],
      "authorship_tag": "ABX9TyNDEGJ8tGf6WdLs63qcI+Vu",
      "include_colab_link": true
    },
    "kernelspec": {
      "display_name": "Python 3",
      "name": "python3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "nbformat": 4,
  "nbformat_minor": 0
}