tessl/pypi-google-cloud-videointelligence
Python Client for Video Intelligence API that enables developers to make videos searchable and discoverable by extracting metadata through machine learning.
—
Pending
video-analysis.mddocs/
Video Analysis
Core client functionality for analyzing videos with Google's AI capabilities. The Video Intelligence API provides both synchronous and asynchronous clients for processing videos with various machine learning features.
Capabilities
VideoIntelligenceServiceClient
Primary synchronous client for video analysis operations. All video analysis operations are long-running and return Operation objects that must be polled for completion.
class VideoIntelligenceServiceClient:
"""
Synchronous client for Video Intelligence Service.
Service that implements the Video Intelligence API.
"""
def __init__(
self,
*,
credentials: Optional[ga_credentials.Credentials] = None,
transport: Optional[Union[str, VideoIntelligenceServiceTransport]] = None,
client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
):
"""
Instantiate the video intelligence service client.
Parameters:
- credentials: The authorization credentials to attach to requests
- transport: The transport to use for communicating with the service
- client_options: Optional client options
- client_info: The client info used to send a user-agent string
"""
def annotate_video(
self,
request: Optional[Union[AnnotateVideoRequest, dict]] = None,
*,
input_uri: Optional[str] = None,
features: Optional[MutableSequence[Feature]] = None,
retry: OptionalRetry = gapic_v1.method.DEFAULT,
timeout: Union[float, object] = gapic_v1.method.DEFAULT,
metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
) -> operation.Operation:
"""
Performs asynchronous video annotation. Progress and results can be
retrieved through the google.longrunning.Operations interface.
Parameters:
- request: The request object or dict
- input_uri: Input video location (Google Cloud Storage URI)
- features: Requested video annotation features
- retry: Designation of what errors, if any, should be retried
- timeout: The timeout for this request
- metadata: Strings which should be sent along with the request as metadata
Returns:
Operation object representing the long-running operation.
Operation.metadata contains AnnotateVideoProgress (progress).
Operation.response contains AnnotateVideoResponse (results).
"""
@classmethod
def from_service_account_file(
cls, filename: str, *args, **kwargs
) -> VideoIntelligenceServiceClient:
"""
Creates an instance of this client using the provided credentials file.
Parameters:
- filename: The path to the service account private key json file
Returns:
VideoIntelligenceServiceClient: The constructed client
"""
@classmethod
def from_service_account_info(
cls, info: dict, *args, **kwargs
) -> VideoIntelligenceServiceClient:
"""
Creates an instance of this client using the provided credentials info.
Parameters:
- info: The service account private key info in dict format
Returns:
VideoIntelligenceServiceClient: The constructed client
"""
@property
def transport(self) -> VideoIntelligenceServiceTransport:
"""Returns the transport used by the client instance."""VideoIntelligenceServiceAsyncClient
Asynchronous version of the video intelligence client for non-blocking operations. Provides the same functionality as the synchronous client but with async/await support.
class VideoIntelligenceServiceAsyncClient:
"""
Asynchronous client for Video Intelligence Service.
Service that implements the Video Intelligence API.
"""
def __init__(
self,
*,
credentials: Optional[ga_credentials.Credentials] = None,
transport: Optional[Union[str, VideoIntelligenceServiceAsyncTransport]] = None,
client_options: Optional[client_options_lib.ClientOptions] = None,
client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
):
"""
Instantiate the async video intelligence service client.
Parameters:
- credentials: The authorization credentials to attach to requests
- transport: The transport to use for communicating with the service
- client_options: Optional client options
- client_info: The client info used to send a user-agent string
"""
async def annotate_video(
self,
request: Optional[Union[AnnotateVideoRequest, dict]] = None,
*,
input_uri: Optional[str] = None,
features: Optional[MutableSequence[Feature]] = None,
retry: OptionalRetry = gapic_v1.method.DEFAULT,
timeout: Union[float, object] = gapic_v1.method.DEFAULT,
metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
) -> operation_async.AsyncOperation:
"""
Performs asynchronous video annotation.
Parameters:
- request: The request object or dict
- input_uri: Input video location (Google Cloud Storage URI)
- features: Requested video annotation features
- retry: Designation of what errors, if any, should be retried
- timeout: The timeout for this request
- metadata: Strings which should be sent along with the request as metadata
Returns:
AsyncOperation object representing the long-running operation.
"""
@classmethod
def from_service_account_file(
cls, filename: str, *args, **kwargs
) -> VideoIntelligenceServiceAsyncClient:
"""
Creates an instance of this client using the provided credentials file.
Parameters:
- filename: The path to the service account private key json file
Returns:
VideoIntelligenceServiceAsyncClient: The constructed async client
"""
@classmethod
def from_service_account_info(
cls, info: dict, *args, **kwargs
) -> VideoIntelligenceServiceAsyncClient:
"""
Creates an instance of this client using the provided credentials info.
Parameters:
- info: The service account private key info in dict format
Returns:
VideoIntelligenceServiceAsyncClient: The constructed async client
"""Usage Examples
Basic Video Analysis
from google.cloud import videointelligence
# Create client
client = videointelligence.VideoIntelligenceServiceClient()
# Analyze video with multiple features
features = [
videointelligence.Feature.LABEL_DETECTION,
videointelligence.Feature.SHOT_CHANGE_DETECTION,
videointelligence.Feature.EXPLICIT_CONTENT_DETECTION
]
operation = client.annotate_video(
request={
"features": features,
"input_uri": "gs://your-bucket/your-video.mp4",
}
)
# Wait for completion and get results
result = operation.result(timeout=600)Async Video Analysis
import asyncio
from google.cloud import videointelligence
async def analyze_video():
# Create async client
client = videointelligence.VideoIntelligenceServiceAsyncClient()
# Analyze video
operation = await client.annotate_video(
request={
"features": [videointelligence.Feature.FACE_DETECTION],
"input_uri": "gs://your-bucket/your-video.mp4",
}
)
# Wait for completion
result = await operation.result(timeout=600)
return result
# Run async analysis
result = asyncio.run(analyze_video())Service Account Authentication
from google.cloud import videointelligence
# Using service account file
client = videointelligence.VideoIntelligenceServiceClient.from_service_account_file(
"path/to/service-account-key.json"
)
# Using service account info dict
service_account_info = {
"type": "service_account",
"project_id": "your-project-id",
# ... other service account fields
}
client = videointelligence.VideoIntelligenceServiceClient.from_service_account_info(
service_account_info
)Path Helper Methods
The client provides utility methods for constructing common resource paths:
# Static methods for path construction
@staticmethod
def common_billing_account_path(billing_account: str) -> str:
"""Return a fully-qualified billing_account string."""
@staticmethod
def common_folder_path(folder: str) -> str:
"""Return a fully-qualified folder string."""
@staticmethod
def common_organization_path(organization: str) -> str:
"""Return a fully-qualified organization string."""
@staticmethod
def common_project_path(project: str) -> str:
"""Return a fully-qualified project string."""
@staticmethod
def common_location_path(project: str, location: str) -> str:
"""Return a fully-qualified location string."""Transport Options
The client supports multiple transport protocols:
- gRPC (default): High-performance binary protocol
- gRPC AsyncIO: Asynchronous gRPC support
- REST: HTTP/1.1 REST API for environments where gRPC is not available
Transport can be specified during client initialization:
# Using specific transport
client = videointelligence.VideoIntelligenceServiceClient(
transport="rest" # or "grpc", "grpc_asyncio"
)Install with Tessl CLI
npx tessl i tessl/pypi-google-cloud-videointelligence