tessl/pypi-google-cloud-visionai
Google Cloud Vision AI API client library for building and deploying Vertex AI Vision applications
—
Pending
warehouse.mddocs/
Media Asset Warehouse
Comprehensive media asset management with AI-powered analysis, indexing, search, and collection organization. The Warehouse service provides enterprise-scale capabilities for storing, analyzing, and discovering video content with sophisticated search and metadata management.
Capabilities
Asset Management
Create, upload, analyze, and manage media assets with comprehensive metadata and AI-powered content understanding.
def create_asset(self, parent: str, asset: Asset, asset_id: str = None) -> Asset:
"""
Creates a new media asset in a corpus.
Args:
parent (str): Required. Corpus resource path
asset (Asset): Required. Asset configuration and metadata
asset_id (str): Optional. ID for the new asset (auto-generated if not provided)
Returns:
Asset: Created asset with metadata and system-assigned fields
"""
def get_asset(self, name: str) -> Asset:
"""
Retrieves asset details and metadata.
Args:
name (str): Required. Asset resource path
"projects/{project}/locations/{location}/corpora/{corpus}/assets/{asset}"
Returns:
Asset: The asset resource with full metadata
"""
def list_assets(self, parent: str, page_size: int = None, page_token: str = None, filter: str = None) -> ListAssetsResponse:
"""
Lists assets in a corpus.
Args:
parent (str): Required. Corpus resource path
page_size (int): Maximum number of assets to return
page_token (str): Token for pagination
filter (str): Filter expression for assets
Returns:
ListAssetsResponse: Response with assets and pagination
"""
def update_asset(self, asset: Asset, update_mask: FieldMask = None) -> Asset:
"""
Updates asset metadata and configuration.
Args:
asset (Asset): Required. Updated asset configuration
update_mask (FieldMask): Fields to update
Returns:
Asset: Updated asset resource
"""
def delete_asset(self, name: str) -> Operation:
"""
Deletes an asset and associated data.
Args:
name (str): Required. Asset resource path to delete
Returns:
Operation: Long-running operation for asset deletion
"""
def upload_asset(self, requests: Iterator[UploadAssetRequest]) -> Operation:
"""
Uploads asset content via streaming.
Args:
requests (Iterator[UploadAssetRequest]): Stream of upload requests with metadata and content
Returns:
Operation: Long-running operation for upload completion
"""
def generate_retrieval_url(self, name: str) -> GenerateRetrievalUrlResponse:
"""
Generates URL for downloading asset content.
Args:
name (str): Required. Asset resource path
Returns:
GenerateRetrievalUrlResponse: Download URL with expiration
"""
def analyze_asset(self, name: str) -> Operation:
"""
Analyzes asset content with AI for metadata extraction.
Args:
name (str): Required. Asset resource path to analyze
Returns:
Operation: Long-running operation for analysis completion
"""
def ingest_asset(self, requests: Iterator[IngestAssetRequest]) -> Iterator[IngestAssetResponse]:
"""
Ingests asset into warehouse via streaming.
Args:
requests (Iterator[IngestAssetRequest]): Stream of ingest requests
Yields:
IngestAssetResponse: Streaming responses with ingestion status
"""
def clip_asset(self, name: str, temporal_partition: Partition) -> ClipAssetResponse:
"""
Creates clips from video assets based on temporal partitions.
Args:
name (str): Required. Asset resource path
temporal_partition (Partition): Temporal boundaries for clip creation
Returns:
ClipAssetResponse: Information about created clips
"""
def generate_hls_uri(self, name: str, temporal_partitions: List[Partition] = None, live_view_enabled: bool = None) -> GenerateHlsUriResponse:
"""
Generates HLS streaming URI for asset playback.
Args:
name (str): Required. Asset resource path
temporal_partitions (List[Partition]): Specific time ranges to include
live_view_enabled (bool): Enable live view for streaming assets
Returns:
GenerateHlsUriResponse: HLS streaming URI and metadata
"""
def import_assets(self, parent: str, assets_gcs_uri: str) -> Operation:
"""
Bulk imports assets from Google Cloud Storage.
Args:
parent (str): Required. Corpus resource path
assets_gcs_uri (str): Required. GCS URI containing assets to import
Returns:
Operation: Long-running operation for bulk import
"""Search and Indexing
Create and manage search indexes, perform content-based searches, and organize searchable content.
def create_index(self, parent: str, index: Index, index_id: str = None) -> Operation:
"""
Creates a search index for asset discovery.
Args:
parent (str): Required. Corpus resource path
index (Index): Required. Index configuration
index_id (str): Optional. ID for the new index
Returns:
Operation: Long-running operation for index creation
"""
def get_index(self, name: str) -> Index:
"""
Retrieves index details and configuration.
Args:
name (str): Required. Index resource path
"projects/{project}/locations/{location}/corpora/{corpus}/indexes/{index}"
Returns:
Index: The index resource with configuration
"""
def list_indexes(self, parent: str, page_size: int = None, page_token: str = None) -> ListIndexesResponse:
"""
Lists indexes in a corpus.
Args:
parent (str): Required. Corpus resource path
page_size (int): Maximum number of indexes to return
page_token (str): Token for pagination
Returns:
ListIndexesResponse: Response with indexes and pagination
"""
def update_index(self, index: Index, update_mask: FieldMask = None) -> Operation:
"""
Updates index configuration.
Args:
index (Index): Required. Updated index configuration
update_mask (FieldMask): Fields to update
Returns:
Operation: Long-running operation for index update
"""
def delete_index(self, name: str) -> Operation:
"""
Deletes a search index.
Args:
name (str): Required. Index resource path to delete
Returns:
Operation: Long-running operation for index deletion
"""
def index_asset(self, name: str, index: str) -> Operation:
"""
Adds asset to search index for discoverability.
Args:
name (str): Required. Asset resource path
index (str): Required. Index resource path
Returns:
Operation: Long-running operation for asset indexing
"""
def remove_index_asset(self, name: str, index: str) -> Operation:
"""
Removes asset from search index.
Args:
name (str): Required. Asset resource path
index (str): Required. Index resource path
Returns:
Operation: Long-running operation for asset removal
"""
def view_indexed_assets(self, index: str, page_size: int = None, page_token: str = None, filter: str = None) -> ViewIndexedAssetsResponse:
"""
Views assets in a search index.
Args:
index (str): Required. Index resource path
page_size (int): Maximum number of assets to return
page_token (str): Token for pagination
filter (str): Filter expression for indexed assets
Returns:
ViewIndexedAssetsResponse: Response with indexed assets
"""
def search_assets(self, corpus: str, criteria: List[Criteria] = None, facet_selections: List[FacetGroup] = None, result_annotation_keys: List[str] = None, search_query: str = None, page_size: int = None, page_token: str = None) -> SearchAssetsResponse:
"""
Searches assets using various criteria and queries.
Args:
corpus (str): Required. Corpus resource path to search
criteria (List[Criteria]): Search criteria and filters
facet_selections (List[FacetGroup]): Faceted search selections
result_annotation_keys (List[str]): Annotation keys to include in results
search_query (str): Text-based search query
page_size (int): Maximum number of results to return
page_token (str): Token for pagination
Returns:
SearchAssetsResponse: Search results with assets and metadata
"""
def search_index_endpoint(self, index_endpoint: str, criteria: List[Criteria] = None, exclusion_criteria: List[Criteria] = None, page_size: int = None, page_token: str = None) -> SearchIndexEndpointResponse:
"""
Searches using deployed index endpoint.
Args:
index_endpoint (str): Required. Index endpoint resource path
criteria (List[Criteria]): Search criteria
exclusion_criteria (List[Criteria]): Exclusion criteria
page_size (int): Maximum number of results
page_token (str): Token for pagination
Returns:
SearchIndexEndpointResponse: Search results from index endpoint
"""Corpus Management
Create and manage corpora as containers for related assets with shared access policies and organization.
def create_corpus(self, parent: str, corpus: Corpus) -> Operation:
"""
Creates a corpus for organizing related assets.
Args:
parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
corpus (Corpus): Required. Corpus configuration
Returns:
Operation: Long-running operation for corpus creation
"""
def get_corpus(self, name: str) -> Corpus:
"""
Retrieves corpus details and configuration.
Args:
name (str): Required. Corpus resource path
"projects/{project}/locations/{location}/corpora/{corpus}"
Returns:
Corpus: The corpus resource with configuration
"""
def update_corpus(self, corpus: Corpus, update_mask: FieldMask = None) -> Corpus:
"""
Updates corpus configuration.
Args:
corpus (Corpus): Required. Updated corpus configuration
update_mask (FieldMask): Fields to update
Returns:
Corpus: Updated corpus resource
"""
def list_corpora(self, parent: str, page_size: int = None, page_token: str = None) -> ListCorporaResponse:
"""
Lists corpora in a project and location.
Args:
parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
page_size (int): Maximum number of corpora to return
page_token (str): Token for pagination
Returns:
ListCorporaResponse: Response with corpora and pagination
"""
def delete_corpus(self, name: str) -> Corpus:
"""
Deletes a corpus and all contained assets.
Args:
name (str): Required. Corpus resource path to delete
Returns:
Corpus: Deleted corpus resource
"""
def analyze_corpus(self, name: str) -> Operation:
"""
Analyzes entire corpus with AI for insights and metadata.
Args:
name (str): Required. Corpus resource path to analyze
Returns:
Operation: Long-running operation for corpus analysis
"""Collection Management
Organize assets into curated collections for easier discovery and management.
def create_collection(self, parent: str, collection: Collection, collection_id: str) -> Operation:
"""
Creates a collection for organizing assets.
Args:
parent (str): Required. Corpus resource path
collection (Collection): Required. Collection configuration
collection_id (str): Required. ID for the new collection
Returns:
Operation: Long-running operation for collection creation
"""
def get_collection(self, name: str) -> Collection:
"""
Retrieves collection details and metadata.
Args:
name (str): Required. Collection resource path
Returns:
Collection: The collection resource with metadata
"""
def update_collection(self, collection: Collection, update_mask: FieldMask = None) -> Collection:
"""
Updates collection configuration.
Args:
collection (Collection): Required. Updated collection configuration
update_mask (FieldMask): Fields to update
Returns:
Collection: Updated collection resource
"""
def list_collections(self, parent: str, page_size: int = None, page_token: str = None) -> ListCollectionsResponse:
"""
Lists collections in a corpus.
Args:
parent (str): Required. Corpus resource path
page_size (int): Maximum number of collections to return
page_token (str): Token for pagination
Returns:
ListCollectionsResponse: Response with collections and pagination
"""
def delete_collection(self, name: str) -> Operation:
"""
Deletes a collection (assets remain in corpus).
Args:
name (str): Required. Collection resource path to delete
Returns:
Operation: Long-running operation for collection deletion
"""
def add_collection_item(self, item: CollectionItem) -> AddCollectionItemResponse:
"""
Adds asset to a collection.
Args:
item (CollectionItem): Required. Collection item to add
Returns:
AddCollectionItemResponse: Confirmation of item addition
"""
def remove_collection_item(self, item: CollectionItem) -> RemoveCollectionItemResponse:
"""
Removes asset from a collection.
Args:
item (CollectionItem): Required. Collection item to remove
Returns:
RemoveCollectionItemResponse: Confirmation of item removal
"""
def view_collection_items(self, collection: str, page_size: int = None, page_token: str = None) -> ViewCollectionItemsResponse:
"""
Views assets in a collection.
Args:
collection (str): Required. Collection resource path
page_size (int): Maximum number of items to return
page_token (str): Token for pagination
Returns:
ViewCollectionItemsResponse: Response with collection items
"""Annotation Management
Create, manage, and organize annotations for assets with structured metadata schemas.
def create_annotation(self, parent: str, annotation: Annotation, annotation_id: str = None) -> Annotation:
"""
Creates annotation for an asset.
Args:
parent (str): Required. Asset resource path
annotation (Annotation): Required. Annotation data and metadata
annotation_id (str): Optional. ID for the new annotation
Returns:
Annotation: Created annotation with system-assigned fields
"""
def get_annotation(self, name: str) -> Annotation:
"""
Retrieves annotation details and data.
Args:
name (str): Required. Annotation resource path
Returns:
Annotation: The annotation resource with data
"""
def list_annotations(self, parent: str, page_size: int = None, page_token: str = None, filter: str = None) -> ListAnnotationsResponse:
"""
Lists annotations for an asset.
Args:
parent (str): Required. Asset resource path
page_size (int): Maximum number of annotations to return
page_token (str): Token for pagination
filter (str): Filter expression for annotations
Returns:
ListAnnotationsResponse: Response with annotations and pagination
"""
def update_annotation(self, annotation: Annotation, update_mask: FieldMask = None) -> Annotation:
"""
Updates annotation data and metadata.
Args:
annotation (Annotation): Required. Updated annotation
update_mask (FieldMask): Fields to update
Returns:
Annotation: Updated annotation resource
"""
def delete_annotation(self, name: str) -> Empty:
"""
Deletes an annotation.
Args:
name (str): Required. Annotation resource path to delete
Returns:
Empty: Empty response confirming deletion
"""Types
Asset Resources
class Asset:
"""Media asset with metadata and content references."""
name: str # Resource name
ttl: Duration # Time-to-live for asset
asset_gcs_source: AssetSource # Asset source location
create_time: Timestamp # Creation timestamp
update_time: Timestamp # Last update timestamp
labels: Dict[str, str] # Resource labels
class AssetSource:
"""Source location for asset content."""
# Union field oneof source:
asset_gcs_source: GcsSource # Google Cloud Storage source
asset_content_data: bytes # Direct content data
class GcsSource:
"""Google Cloud Storage source configuration."""
uris: List[str] # GCS URIs for asset content
class Corpus:
"""Container for related assets."""
name: str # Resource name
display_name: str # Human-readable name
description: str # Corpus description
default_ttl: Duration # Default TTL for assets
create_time: Timestamp # Creation timestamp
update_time: Timestamp # Last update timestamp
class Collection:
"""Curated collection of assets."""
name: str # Resource name
display_name: str # Human-readable name
description: str # Collection description
create_time: Timestamp # Creation timestamp
update_time: Timestamp # Last update timestamp
class CollectionItem:
"""Item within a collection."""
collection: str # Collection resource path
type: CollectionItemType # Type of collection item
# Union field oneof item:
asset: str # Asset resource path
class CollectionItemType(Enum):
"""Types of collection items."""
COLLECTION_ITEM_TYPE_UNSPECIFIED = 0
ASSET = 1 # Asset itemSearch and Index Resources
class Index:
"""Search index configuration."""
name: str # Resource name
display_name: str # Human-readable name
description: str # Index description
state: IndexState # Current index state
create_time: Timestamp # Creation timestamp
update_time: Timestamp # Last update timestamp
class IndexState(Enum):
"""Index operational states."""
STATE_UNSPECIFIED = 0
CREATING = 1 # Index being created
CREATED = 2 # Index created and ready
UPDATING = 3 # Index being updated
DELETING = 4 # Index being deleted
ERROR = 5 # Index in error state
class Criteria:
"""Search criteria specification."""
# Union field oneof criteria:
field_criteria: FieldCriteria # Field-based criteria
text_criteria: TextCriteria # Text-based criteria
date_time_range_criteria: DateTimeRangeCriteria # Time range criteria
float_range_criteria: FloatRangeCriteria # Numeric range criteria
boolean_criteria: BooleanCriteria # Boolean criteria
feature_criteria: FeatureCriteria # Feature-based criteria
class FieldCriteria:
"""Field-based search criteria."""
field: str # Field name to search
# Union field oneof value:
text_array: StringArray # Text values
int_array: IntArray # Integer values
float_array: FloatArray # Float values
class TextCriteria:
"""Text-based search criteria."""
text_query: str # Text query string
class DateTimeRangeCriteria:
"""Date/time range search criteria."""
date_time_ranges: List[DateTimeRange] # Time ranges to search
class SearchResultItem:
"""Individual search result."""
asset: str # Asset resource path
segments: List[Segment] # Matching segments within asset
relevance_score: float # Relevance score for result
class Segment:
"""Segment within search result."""
start_time_offset: Duration # Start time of segment
end_time_offset: Duration # End time of segment
confidence: float # Confidence score for segmentAnnotation Resources
class Annotation:
"""Annotation data for assets."""
name: str # Resource name
user_specified_annotation: UserSpecifiedAnnotation # User-created annotation
create_time: Timestamp # Creation timestamp
update_time: Timestamp # Last update timestamp
class UserSpecifiedAnnotation:
"""User-specified annotation data."""
key: str # Annotation key/identifier
# Union field oneof value:
partition: Partition # Temporal or spatial partition
classification_annotation: ClassificationAnnotation # Classification data
# Additional annotation value types...
class Partition:
"""Temporal or spatial partition specification."""
# Union field oneof partition_dimension:
temporal_partition: TemporalPartition # Time-based partition
spatial_partition: SpatialPartition # Space-based partition
relative_temporal_partition: RelativeTemporalPartition # Relative time partition
class TemporalPartition:
"""Time-based partition."""
start_time: Timestamp # Start time
end_time: Timestamp # End time
class SpatialPartition:
"""Space-based partition."""
# Union field oneof region:
circle_area: CircleArea # Circular area
class CircleArea:
"""Circular spatial area."""
latitude: float # Center latitude
longitude: float # Center longitude
radius_meter: float # Radius in metersUsage Examples
Complete Asset Management Workflow
from google.cloud import visionai_v1
# Create client
client = visionai_v1.WarehouseClient()
# Step 1: Create corpus
parent = "projects/my-project/locations/us-central1"
corpus = visionai_v1.Corpus(
display_name="Security Videos Corpus",
description="Repository for security camera footage",
default_ttl={"seconds": 86400 * 30} # 30 days retention
)
create_corpus_op = client.create_corpus(parent=parent, corpus=corpus)
corpus_result = create_corpus_op.result()
corpus_path = corpus_result.name
print(f"Created corpus: {corpus_path}")
# Step 2: Upload and create asset
asset = visionai_v1.Asset(
asset_gcs_source=visionai_v1.AssetSource(
asset_gcs_source=visionai_v1.GcsSource(
uris=["gs://my-bucket/security-footage/camera1-20240910.mp4"]
)
),
labels={
"camera_id": "camera-001",
"location": "main-entrance"
}
)
created_asset = client.create_asset(
parent=corpus_path,
asset=asset,
asset_id="camera1-footage-20240910"
)
asset_path = created_asset.name
print(f"Created asset: {asset_path}")
# Step 3: Analyze asset with AI
analyze_op = client.analyze_asset(name=asset_path)
analyze_result = analyze_op.result()
print(f"Asset analysis completed")
# Step 4: Create search index
index = visionai_v1.Index(
display_name="Security Footage Index",
description="Index for searching security camera footage"
)
create_index_op = client.create_index(
parent=corpus_path,
index=index,
index_id="security-footage-index"
)
index_result = create_index_op.result()
index_path = index_result.name
print(f"Created index: {index_path}")
# Step 5: Add asset to index
index_asset_op = client.index_asset(
name=asset_path,
index=index_path
)
index_asset_op.result()
print(f"Asset indexed successfully")Advanced Search Operations
def search_security_footage():
"""Example of advanced search operations."""
client = visionai_v1.WarehouseClient()
corpus_path = "projects/my-project/locations/us-central1/corpora/security-videos"
# Multi-criteria search
search_criteria = [
# Search for specific time range
visionai_v1.Criteria(
date_time_range_criteria=visionai_v1.DateTimeRangeCriteria(
date_time_ranges=[
visionai_v1.DateTimeRange(
start_datetime={"seconds": 1725926400}, # Sept 10, 2024 00:00
end_datetime={"seconds": 1725940800} # Sept 10, 2024 04:00
)
]
)
),
# Search for specific camera
visionai_v1.Criteria(
field_criteria=visionai_v1.FieldCriteria(
field="labels.camera_id",
text_array=visionai_v1.StringArray(txt_values=["camera-001", "camera-002"])
)
),
# Text-based search for motion events
visionai_v1.Criteria(
text_criteria=visionai_v1.TextCriteria(
text_query="motion detected person walking"
)
)
]
# Perform search
search_response = client.search_assets(
corpus=corpus_path,
criteria=search_criteria,
result_annotation_keys=["motion_detection", "person_detection"],
page_size=50
)
# Process search results
for result in search_response.search_result_items:
print(f"Found asset: {result.asset}")
print(f"Relevance score: {result.relevance_score}")
for segment in result.segments:
print(f" Matching segment: {segment.start_time_offset} - {segment.end_time_offset}")
print(f" Confidence: {segment.confidence}")
def manage_collections():
"""Example of collection management."""
client = visionai_v1.WarehouseClient()
corpus_path = "projects/my-project/locations/us-central1/corpora/security-videos"
# Create collection for high-priority incidents
collection = visionai_v1.Collection(
display_name="High Priority Incidents",
description="Critical security incidents requiring review"
)
create_collection_op = client.create_collection(
parent=corpus_path,
collection=collection,
collection_id="high-priority-incidents"
)
collection_result = create_collection_op.result()
collection_path = collection_result.name
# Add assets to collection
asset_paths = [
f"{corpus_path}/assets/incident-001",
f"{corpus_path}/assets/incident-002",
f"{corpus_path}/assets/incident-003"
]
for asset_path in asset_paths:
collection_item = visionai_v1.CollectionItem(
collection=collection_path,
type=visionai_v1.CollectionItemType.ASSET,
asset=asset_path
)
client.add_collection_item(item=collection_item)
print(f"Added asset to collection: {asset_path}")
# View collection items
items_response = client.view_collection_items(collection=collection_path)
print(f"Collection contains {len(items_response.items)} items")
def create_detailed_annotations():
"""Example of creating detailed annotations."""
client = visionai_v1.WarehouseClient()
asset_path = "projects/my-project/locations/us-central1/corpora/security-videos/assets/incident-001"
# Create temporal annotation for person detection
person_detection_annotation = visionai_v1.Annotation(
user_specified_annotation=visionai_v1.UserSpecifiedAnnotation(
key="person_detection",
partition=visionai_v1.Partition(
temporal_partition=visionai_v1.TemporalPartition(
start_time={"seconds": 1725928200}, # 00:30:00
end_time={"seconds": 1725928320} # 00:32:00
)
)
)
)
created_annotation = client.create_annotation(
parent=asset_path,
annotation=person_detection_annotation
)
print(f"Created annotation: {created_annotation.name}")
# Create spatial annotation for motion area
motion_area_annotation = visionai_v1.Annotation(
user_specified_annotation=visionai_v1.UserSpecifiedAnnotation(
key="motion_area",
partition=visionai_v1.Partition(
spatial_partition=visionai_v1.SpatialPartition(
circle_area=visionai_v1.CircleArea(
latitude=37.4219999,
longitude=-122.0840575,
radius_meter=50.0
)
)
)
)
)
spatial_annotation = client.create_annotation(
parent=asset_path,
annotation=motion_area_annotation
)
print(f"Created spatial annotation: {spatial_annotation.name}")Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-visionai