model_garden/notebooks/community/vision/video_warehouse_sdk.ipynb

In [ ]:

# Copyright 2023 Google LLC
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     https://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

Video Warehouse SDK Demo

Run in Colab

View on GitHub

Open in Vertex AI Workbench

NOTE: This notebook has been tested in the following environment:

• Python version = 3.10

Overview

Learn how to build a Video Warehouse step by step by using SDK.

Objective

The objective is to demostrate how to use Warehouse SDK to process input videos, index and search.

The steps to perform include:

• Create corpus.

• Create and upload assets using video files from Google Cloud Storage.

• Create index, create index endpoint, and deploy index.

  • This step can take an hour.

• Run transformations to analyze the assets:

  • Speech transformation
    • Use Video intelligence API to run speech transcription and store into warehouse. By default, the speech result can be searched with criteria with "speech" field. You can specify the search criteria field by setting speech_transcript_search_criteria_key in SpeechTransformerInitConfig.
  • OCR transformation
    • Use Video intelligence API to run text detection and store into warehouse. By default, the text detection result can be searched with criteria with "text" field. Specify the search criteria field by setting ocr_search_criteria_key in OcrTransformerInitConfig.
  • Embedding analysis

• Index.

• Search.

• Clean up resources (assets, index, index endpoint, corpus).

Dataset

The dataset will use a collection of videos stored in a Google Cloud Storage bucket: gs://cloud-samples-data/video.

This tutorial shows how to use warehouse SDK to run speech transcription, text detection, embedding analysis for the videos of this dataset, builds search index and provides search capabilities.

Costs

This tutorial uses billable components of Google Cloud:

Vertex AI Vision (Pricing)

Video Intelligence (Pricing)

Installation

Install the following packages required to execute this notebook.

In [ ]:

! gcloud storage cp gs://visionai-artifacts/visionai-0.0.6-py3-none-any.whl .
! pip install visionai-0.0.6-py3-none-any.whl --force-reinstall

Colab only: Uncomment the following cell to restart the kernel.

In [ ]:

# import IPython

# app = IPython.Application.instance()
# app.kernel.do_shutdown(True)

Before you begin

Set up your Google Cloud project

The following steps are required, regardless of your notebook environment.

1. Select or create a Google Cloud project. When you first create an account, you get a $300 free credit towards your compute/storage costs.

2. Make sure that billing is enabled for your project.

Set your project ID

If you don't know your project ID, try the following:

• Run gcloud config list.
• Run gcloud projects list.
• See the support page: Locate the project ID

In [ ]:

PROJECT_ID = "[your-project-id]"  # @param {type:"string"}

# Set the project id
! gcloud config set project {PROJECT_ID}

Authenticate your Google Cloud account

Depending on your Jupyter environment, you may have to manually authenticate. Follow the relevant instructions below.

1. Vertex AI Workbench

• Do nothing as you are already authenticated.

2. Local JupyterLab instance, uncomment and run:

In [ ]:

# ! gcloud auth login

In [ ]:

# ! gcloud auth application-default login

3. Colab, uncomment and run:

In [ ]:

# from google.colab import auth
# auth.authenticate_user(project_id=PROJECT_ID)

Set Up Other Constants

In [ ]:

PROJECT_NUMBER_STR = !gcloud projects describe $PROJECT_ID --format="value(projectNumber)"
PROJECT_NUMBER = int(PROJECT_NUMBER_STR[0])

# Only us-central1 is supported.
# Please note that this region is for VisionAi services. For speech
# transcription, we may not respect the region here.
REGION = "us-central1"

CORPUS_DISPLAY_NAME = "Demo corpus"  # @param {type: "string"}
CORPUS_DESCRIPTION = "Demo corpus to demo warehouse transformations and search"  # @param {type: "string"}

# External users can only access PROD environment.
ENV = "PROD"

INDEX_DISPLAY_NAME = "Demo Index"  # @param {type: "string"}
INDEX_ENDPOINT_DISPLAY_NAME = "Demo Index Endpoint"  # @param {type: "string"}

CLEAN_UP_ASSETS = True  # @param {type: "boolean"}
CLEAN_UP_INDEX = True  # @param {type: "boolean"}
CLEAN_UP_CORPUS = True  # @param {type: "boolean"}

Whether Using Existing Corpus and Index

In [ ]:

# Because it takes ~1h to create and deploy index. A existing index can be
# specified to save time.

# If CORPUS_ID is specified, skip creating a new corpus.
CORPUS_ID = None  # @param {type: "string"}
# If DEPLOYED_INDEX_ID is specified, use existing index instead of creating and
# deploying a new index.
DEPLOYED_INDEX_ID = None  # @param {type: "string"}

Input Video Files

In [ ]:

GCS_FILES = [
    "gs://cloud-samples-data/video/animals.mp4",
    "gs://cloud-samples-data/video/googlework_short.mp4",
    "gs://cloud-samples-data/video/chicago.mp4",
    (
        "gs://cloud-samples-data/video/Machine Learning Solving Problems"
        " Big, Small, and Prickly.mp4"
    ),
    "gs://cloud-samples-data/video/JaneGoodall.mp4",
    "gs://cloud-samples-data/video/gbikes_dinosaur.mp4",
    "gs://cloud-samples-data/video/pizza.mp4",
]

Enable API

In [ ]:

!gcloud services enable videointelligence.googleapis.com
!gcloud services enable visionai.googleapis.com

Config logging

In [ ]:

import logging

logging.basicConfig()
logging.getLogger().setLevel(logging.INFO)
_logger = logging.getLogger("colab")

Import libraries

In [ ]:

import concurrent
import logging

from visionai.python.gapic.visionai import visionai_v1
from visionai.python.net import channel
from visionai.python.warehouse.transformer import \
    asset_indexing_transformer as ait
from visionai.python.warehouse.transformer import (ocr_transformer,
                                                   speech_transformer,
                                                   transformer_factory)
from visionai.python.warehouse.utils import (vod_asset, vod_corpus,
                                             vod_index_endpoint)

Create Warehouse client

In [ ]:

warehouse_endpoint = channel.get_warehouse_service_endpoint(channel.Environment[ENV])
warehouse_client = visionai_v1.WarehouseClient(
    client_options={"api_endpoint": warehouse_endpoint}
)

Create a Corpus or use existing Corpus.

In [ ]:

if CORPUS_ID is None:
    corpus_name = vod_corpus.create_corpus(
        warehouse_client,
        PROJECT_NUMBER,
        REGION,
        CORPUS_DISPLAY_NAME,
        CORPUS_DESCRIPTION,
    ).name
else:
    corpus_name = visionai_v1.WarehouseClient.corpus_path(
        PROJECT_NUMBER, REGION, CORPUS_ID
    )

Create an Executor

In [ ]:

# Creates an executor to upload and transform assets in parallel.
executor = concurrent.futures.ThreadPoolExecutor(max_workers=8)

Create and Upload Assets

In [ ]:

new_asset_futures = []
for gcs_file in GCS_FILES:
    new_asset_futures.append(
        executor.submit(
            vod_asset.create_and_upload_asset,
            warehouse_client,
            gcs_file,
            corpus_name,
        )
    )
done_or_error, _ = concurrent.futures.wait(
    new_asset_futures, return_when="ALL_COMPLETED"
)
asset_names = []
for done_future in done_or_error:
    try:
        asset_names.append(done_future.result())
        _logger.info("Create and upload asset succeeded %s", done_future.result())
    except Exception as e:
        _logger.exception(e)

Prepare Index or use existing Index

In [ ]:

# Create index and index endpoint for the corpus, or use existing index
# and index endpoint if specified.
if DEPLOYED_INDEX_ID is None:
    # Creates index for the corpus.
    index_name = vod_corpus.index_corpus(
        warehouse_client, corpus_name, INDEX_DISPLAY_NAME
    )
    # Creates index endpoint and deploys the created index above to the index
    # endpoint.
    index_endpoint_name = vod_index_endpoint.create_index_endpoint(
        warehouse_client,
        PROJECT_NUMBER,
        REGION,
        INDEX_ENDPOINT_DISPLAY_NAME,
    ).name
    deploy_operation = warehouse_client.deploy_index(
        visionai_v1.DeployIndexRequest(
            index_endpoint=index_endpoint_name,
            deployed_index=visionai_v1.DeployedIndex(
                index=index_name,
            ),
        )
    )
    _logger.info("Wait for index to be deployed %s.", deploy_operation.operation.name)
    # Wait for the deploy index operation. Depends on the data size to be
    # indexed, the timeout may need to be increased.
    deploy_operation.result(timeout=7200)
    _logger.info("Index is deployed.")
else:
    index_name = "{}/indexes/{}".format(corpus_name, DEPLOYED_INDEX_ID)
    index = warehouse_client.get_index(visionai_v1.GetIndexRequest(name=index_name))
    _logger.info("Use existing index %s.", index)
    if index.state != visionai_v1.Index.State.CREATED:
        _logger.critical("Invalid index. The index state must be Created.")
    if not index.deployed_indexes:
        _logger.critical("Invalid index. The index must be deployed.")
    index_endpoint_name = index.deployed_indexes[0].index_endpoint

Run Transforms

In [ ]:

ocr_config = ocr_transformer.OcrTransformerInitConfig(
    corpus_name=corpus_name,
    env=channel.Environment[ENV],
)

ml_config = transformer_factory.MlTransformersCreationConfig(
    run_embedding=True,
    speech_transformer_init_config=speech_transformer.SpeechTransformerInitConfig(
        corpus_name=corpus_name, language_code="en-US"
    ),
    ocr_transformer_init_config=ocr_config,
)
ml_transformers = transformer_factory.create_ml_transformers(
    warehouse_client, ml_config
)
# Creates indexing transformer to index assets.
asset_indexing_transformer = ait.AssetIndexingTransformer(warehouse_client, index_name)
# Runs the transformers for the assets.
futures = []

for asset_name in asset_names:
    futures.append(
        executor.submit(
            vod_asset.transform_single_asset,
            asset_name,
            ml_transformers,
            asset_indexing_transformer,
        )
    )
done_or_error, _ = concurrent.futures.wait(futures, return_when="ALL_COMPLETED")
for future in done_or_error:
    try:
        future.result()
    except Exception as e:
        _logger.exception(e)

all_transformers = ml_transformers + [asset_indexing_transformer]
for transformer in all_transformers:
    transformer.teardown()

Search

In [ ]:

search_response = warehouse_client.search_index_endpoint(
    visionai_v1.SearchIndexEndpointRequest(
        index_endpoint=index_endpoint_name,
        text_query="dinosaur",
        page_size=10,
    )
)
_logger.info("Search response: %s", search_response)

In [ ]:

cr = visionai_v1.Criteria(
    field="speech", text_array=visionai_v1.StringArray(txt_values=["kid"])
)
search_response = warehouse_client.search_index_endpoint(
    visionai_v1.SearchIndexEndpointRequest(
        index_endpoint=index_endpoint_name,
        text_query="river",
        criteria=[cr],
        page_size=100,
    )
)
_logger.info("Search response: %s", search_response)

In [ ]:

cr = visionai_v1.Criteria(
    field="text", text_array=visionai_v1.StringArray(txt_values=["National Park"])
)
search_response = warehouse_client.search_index_endpoint(
    visionai_v1.SearchIndexEndpointRequest(
        index_endpoint=index_endpoint_name,
        text_query="trees",
        criteria=[cr],
        page_size=100,
    )
)
_logger.info("Search response: %s", search_response)

Cleaning up

In [ ]:

if CLEAN_UP_ASSETS:
    for asset_name in asset_names:
        warehouse_client.delete_asset(visionai_v1.DeleteAssetRequest(name=asset_name))
        _logger.info("Deleted asset %s", asset_name)

if CLEAN_UP_INDEX:
    undeploy_operation = warehouse_client.undeploy_index(
        visionai_v1.UndeployIndexRequest(index_endpoint=index_endpoint_name)
    )
    _logger.info(
        "Wait for index to be undeployed %s.",
        undeploy_operation.operation.name,
    )
    # Wait for the undeploy index operation.
    undeploy_operation.result(timeout=1800)
    _logger.info("Index is undeployed.")
    warehouse_client.delete_index(visionai_v1.DeleteIndexRequest(name=index_name))
    _logger.info("Deleted index %s", index_name)
    warehouse_client.delete_index_endpoint(
        visionai_v1.DeleteIndexEndpointRequest(name=index_endpoint_name)
    )
    _logger.info("Deleted index endpoint %s", index_endpoint_name)

if CLEAN_UP_CORPUS:
    warehouse_client.delete_corpus(visionai_v1.DeleteCorpusRequest(name=corpus_name))
    _logger.info("Deleted corpus %s", corpus_name)