Video to Document with GPT-4 Turbo Vision & Azure AI Spatial Analysis Video Retrieval¶

#%pip install opencv-python
#%pip install moviepy

import base64
import cv2
import datetime
import io
import matplotlib.pyplot as plt
import numpy as np
import openai
import os
import pandas as pd
import requests
import sys
import time

from azure.core.credentials import AzureSasCredential
from azure.storage.blob import BlobClient
from dotenv import load_dotenv
from IPython.display import Video, Markdown, HTML
from moviepy.editor import VideoFileClip
from urllib.parse import quote

load_dotenv("azure.env")

# 1. Azure OpenAI
AOAI_KEY = os.getenv("AOAI_KEY")
AOAI_ENDPOINT = os.getenv("AOAI_ENDPOINT")

# 2. Azure AI Vision
AZURE_CV_KEY = os.getenv("AZURE_CV_KEY")
AZURE_CV_ENDPOINT = os.getenv("AZURE_CV_ENDPOINT")

# 3. Azure Storage Account
# Check the validatity date and "Add, Create, Write, Delete and List" permissions for the SAS Token
AZURE_SAS_TOKEN = os.getenv("AZURE_SAS_TOKEN")
AZURE_STORAGE_ACCOUNT = os.getenv("AZURE_STORAGE_ACCOUNT")
# Container to store the videos
AZURE_STORAGE_CONTAINER = os.getenv("AZURE_STORAGE_CONTAINER")

pd.set_option("display.max_colwidth", False)

GPT_4V_DEPLOYMENT = "gpt-4TurboVision"  # Your GPT4 Turbo Vision deployed model

GPT_4V_ENDPOINT = f"{AOAI_ENDPOINT}/openai/deployments/{GPT_4V_DEPLOYMENT}\
/extensions/chat/completions?api-version=2023-12-01-preview"

AZURE_STORAGE_PATH = "checklist/"  # Directory to create into your blob storage

VIDEO_DIR = "videos"
VIDEO_INDEX = f"ind-{str(datetime.datetime.today().strftime('%d%b%Y-%H%M%S'))}"

def upload_file_to_blob(blob_name: str, source_dir=VIDEO_DIR, file_extension: str = ""):
    """
    Uploads a file to Azure Blob Storage.

    Args:
        blob_name (str): The name of the blob (file) to be uploaded.
        source_dir (str, optional): The directory of the source file. Defaults to VIDEO_DIR.
        file_extension (str, optional): The file extension (including the dot, e.g., '.mp4') to
        be appended to blob_name. Defaults to ''.

    Returns:
        str: The URL of the uploaded blob.

    """
    account_url = "https://{}.blob.core.windows.net/".format(AZURE_STORAGE_ACCOUNT)

    full_blob_name = os.path.join(AZURE_STORAGE_PATH, blob_name + file_extension)
    encoded_blob_name = quote(full_blob_name)  # URL-encode the full_blob_name

    blob_client = BlobClient(
        account_url=account_url,
        container_name=AZURE_STORAGE_CONTAINER,
        blob_name=full_blob_name,
        credential=AzureSasCredential(AZURE_SAS_TOKEN),
        max_block_size=1024 * 1024 * 4,  # 4 MiB
        max_single_put_size=1024 * 1024 * 16,  # 16 MiB
    )

    with open(
        file=os.path.join(source_dir, blob_name + file_extension), mode="rb"
    ) as data:
        blob_client.upload_blob(data=data, overwrite=True, max_concurrency=2)

    return account_url + AZURE_STORAGE_CONTAINER + "/" + encoded_blob_name


def local_videos_to_azure_blob(file_path, video_dir=VIDEO_DIR):
    """
    Upload local videos to Azure Storage Account
    """
    video_path = os.path.join(video_dir, file_path)

    # extract metadata from video file
    with VideoFileClip(video_path) as video:
        duration_mins = round(video.duration / 60, 1)  # Duration in minutes
        fps = round(video.fps)  # Frames per second
        resolution = video.size  # Resolution (width, height)

    file_size_bytes = os.path.getsize(video_path)
    size_mb = round(file_size_bytes / (1024 * 1024), 1)

    # upload video file to blob storage
    blob_url = upload_file_to_blob(file_path, video_dir)

    metadata = {
        "duration_mins": duration_mins,
        "fps": fps,
        "resolution": resolution,
        "document_url": blob_url,
        "size_mb": size_mb,
    }

    return metadata


def list_video_indexes(vision_api_endpoint, vision_api_key):
    """
    List indexes
    """
    url = f"{vision_api_endpoint}/computervision/retrieval/indexes?api-version=2023-05-01-preview"
    headers = {"Ocp-Apim-Subscription-Key": vision_api_key}
    response = requests.get(url, headers=headers)

    return response.json()


def delete_video_index(vision_api_endpoint, vision_api_key, index_name):
    """
    Delete video index
    """
    url = f"{vision_api_endpoint}/computervision/retrieval/indexes/{index_name}?api-version=2023-05-01-preview"
    headers = {
        "Ocp-Apim-Subscription-Key": vision_api_key,
        "Content-Type": "application/json",
    }
    response = requests.delete(url, headers=headers)

    return response


def create_video_index(vision_api_endpoint, vision_api_key, index_name, metadata=False):
    url = f"{vision_api_endpoint}/computervision/retrieval/indexes/{index_name}?api-version=2023-05-01-preview"
    headers = {
        "Ocp-Apim-Subscription-Key": vision_api_key,
        "Content-Type": "application/json",
    }
    data = {"features": [{"name": "vision", "domain": "generic"}, {"name": "speech"}]}

    if metadata:
        data["metadataSchema"] = metadata
    response = requests.put(url, headers=headers, json=data)

    return response


def wait_for_ingestion_completion(
    vision_api_endpoint, vision_api_key, index_name, max_retries=30
):
    """
    Wait ingestion completion
    """
    url = f"{vision_api_endpoint}/computervision/retrieval/indexes/{index_name}/ingestions?api-version=2023-05-01-preview"
    headers = {"Ocp-Apim-Subscription-Key": vision_api_key}
    retries = 0

    while retries < max_retries:
        time.sleep(10)
        response = requests.get(url, headers=headers)

        if response.status_code == 200:
            state_data = response.json()
            if state_data["value"][0]["state"] == "Completed":
                print(state_data)
                print("Ingestion completed.")
                print(response)
                return True
            elif state_data["value"][0]["state"] == "Failed":
                print(state_data)
                print("Ingestion failed.")
                return False
        retries += 1

    return False


def get_indexed_video_documents(vision_api_endpoint, vision_api_key, index_name):
    """
    Get indexed videos
    """
    url = f"{vision_api_endpoint}/computervision/retrieval/indexes/{index_name}/documents?api-version=2023-05-01-preview&$top=30"
    headers = {
        "Ocp-Apim-Subscription-Key": vision_api_key
    }  # , "Content-Type": "application/json"}
    response = requests.get(url, headers=headers)

    return response


def find_scene(queryText="", featureFilters=["vision", "speech"]):
    """
    Searches text over vision and/or speech features of an indexed video. It is also called frame locator or video
    retrieval api.
    It returns the relevant frames based on the text input and adname.

    Args:
        queryText: Free text input to search in "vision" and "speech" features
        featureFilters: Video features to search a given text, it is possible to select both or only one of "vision"
        and "speech"
    Returns:
        query_result : Response returned from video retrieval api, which is a list of most relevant frames with their
        confidence scores.
    """
    url = (
        AZURE_CV_ENDPOINT
        + "/computervision/retrieval/indexes/"
        + VIDEO_INDEX
        + ":queryByText?api-version=2023-05-01-preview"
    )

    headers = {
        "Content-type": "application/json",
        "Ocp-Apim-Subscription-Key": AZURE_CV_KEY,
    }

    body = {
        "queryText": queryText,
        "dedup": True,
        "filters": {"featureFilters": featureFilters},
    }

    query_result = None

    try:
        r = requests.post(url, json=body, headers=headers)
        query_result = r.json()
    except Exception as error:
        print("Video search operation failed ")
        print(error)

    return query_result


def extract_frame(video_path, timestamp):
    """
    Extract a frame from a video at a given timestamp.
    """
    try:
        # Handle timestamps with fractional seconds
        h, m, s = timestamp.split(":")
        s, ms = (s.split(".") + ["0"])[
            :2
        ]  # Split seconds and milliseconds, default ms to '0' if not present
        frame_time = (int(h) * 3600 + int(m) * 60 + int(s)) * 1000 + int(
            ms[:3]
        )  # Convert to milliseconds

        cap = cv2.VideoCapture(video_path)
        cap.set(cv2.CAP_PROP_POS_MSEC, frame_time)

        success, frame = cap.read()

        if success:
            return cv2.cvtColor(
                frame, cv2.COLOR_BGR2RGB
            )  # Convert color to RGB for matplotlib

    except Exception as e:
        print(f"Error extracting frame: {e}")

    return None


def video_chat(
    video_url,
    document_id,
    user_prompt,
    sas_token=AZURE_SAS_TOKEN,
    video_index=VIDEO_INDEX,
    gpt_4v_endpoint=GPT_4V_ENDPOINT,
    openai_api_key=AOAI_KEY,
    azure_cv_endpoint=AZURE_CV_ENDPOINT,
    azure_cv_key=AZURE_CV_KEY,
):
    """
    Video chat using Azure OpenAI GPT-4 Turbo with Vision and Azure AI Vision
    """
    headers = {"Content-Type": "application/json", "api-key": openai_api_key}

    payload = {
        "model": "gpt-4-vision-preview",
        "enhancements": {"video": {"enabled": True}},
        "dataSources": [
            {
                "type": "AzureComputerVisionVideoIndex",
                "parameters": {
                    "computerVisionBaseUrl": f"{azure_cv_endpoint}/computervision",
                    "computerVisionApiKey": azure_cv_key,
                    "indexName": video_index,
                    "videoUrls": [f"{video_url}?{sas_token}"],
                },
            }
        ],
        "messages": [
            {
                "role": "system",
                "content": "You are a helpful assistant to analyze images and videos",
            },
            {
                "role": "user",
                "content": [
                    {"type": "text", "text": user_prompt},
                    {"type": "acv_document_id", "acv_document_id": document_id},
                ],
            },
        ],
        "max_tokens": 1000,
        "temperature": 0,
        "seed": 1,
    }

    try:
        response = requests.post(gpt_4v_endpoint, headers=headers, json=payload)
        response.raise_for_status()
        content = response.json()["choices"][0]["message"]["content"]
        return content

    except requests.RequestException as e:
        raise SystemExit(f"Failed to make the request. Error: {e}")


def display_tiles(matches_df, nrows=2, ncols=3, figsize=(16, 9)):
    """
    Display videos tiles
    """
    fig, axes = plt.subplots(nrows=nrows, ncols=ncols, figsize=figsize, dpi=100)
    axes = axes.flatten()

    for i, (ax, (_, row)) in enumerate(zip(axes, matches_df.iterrows())):
        video_path = f"{VIDEO_DIR}/{row['documentId']}"
        timestamp = row["best"]
        frame = extract_frame(video_path, timestamp)

        if frame is not None:
            ax.imshow(frame)
        else:
            ax.imshow(
                np.zeros((100, 100, 3), dtype=np.uint8)
            )  # Display an empty black frame

        title = f"Top {i+1}, (score: {row['relevance']:.3f}) - Video id: {row['documentId']}"

        # Adjusting the timestamp format to remove sub-second information
        start = row["start"].split(".")[0] if "start" in row else None
        end = row["end"].split(".")[0] if "end" in row else None
        subtitle = f"{start} - {end}" if start and end else "Timestamp not available"

        ax.set_title(f"{title}\n{subtitle}", fontsize=10, pad=10)
        ax.axis("off")

    plt.tight_layout()
    plt.show()


def image_to_html(image, format="jpeg"):
    """
    Image to HTML
    """
    buffer = io.BytesIO()
    plt.imsave(buffer, image, format=format)
    data_uri = base64.b64encode(buffer.getvalue()).decode("utf-8")

    return f'<img src="data:image/{format};base64,{data_uri}" width="200">'


def index_video(video):
    """
    Function to structure video data for indexing
    """
    return {
        "mode": "add",
        "documentId": video["id"],
        "documentUrl": video["document_url"] + "?" + AZURE_SAS_TOKEN,
        "metadata": {
            "filename": video["filename"],
            "duration_mins": str(video["duration_mins"]),
            "size_mb": str(video["size_mb"]),
            "resolution": str(video["resolution"]),
            "fps": str(video["fps"]),
        },
    }


def get_indexes():
    """
    List of existing indexes
    """
    print("List of existing indexes:")
    indexes = list_video_indexes(AZURE_CV_ENDPOINT, AZURE_CV_KEY)
    index_names = [index["name"] for index in indexes["value"]]
    print(index_names)

    return index_names


def delete_index(index_to_delete):
    """
    Delete index
    """
    print(f"Deleting the index {index_to_delete}")
    response = delete_video_index(AZURE_CV_ENDPOINT, AZURE_CV_KEY, index_to_delete)

    if response.status_code == 204:
        print("Done")
    else:
        print("Error")

# list existing indexes
indexes = list_video_indexes(AZURE_CV_ENDPOINT, AZURE_CV_KEY)
index_names = [index["name"] for index in indexes["value"]]