Streaming Analysis

class StreamingVideoIntelligenceServiceClient:
    """
    Synchronous client for Streaming Video Intelligence Service.
    
    Service that implements streaming video analysis using AI.
    """
    
    def __init__(
        self,
        *,
        credentials: Optional[ga_credentials.Credentials] = None,
        transport: Optional[Union[str, StreamingVideoIntelligenceServiceTransport]] = None,
        client_options: Optional[Union[client_options_lib.ClientOptions, dict]] = None,
        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
    ):
        """
        Instantiate the streaming 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 streaming_annotate_video(
        self,
        requests: Iterator[Union[StreamingAnnotateVideoRequest, dict]],
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: Union[float, object] = gapic_v1.method.DEFAULT,
        metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
    ) -> Iterable[StreamingAnnotateVideoResponse]:
        """
        Performs video annotation with streaming requests and responses.

        Parameters:
        - requests: Iterator of StreamingAnnotateVideoRequest objects
        - 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:
        Iterable of StreamingAnnotateVideoResponse objects containing real-time analysis results.
        """

    @classmethod
    def from_service_account_file(
        cls, filename: str, *args, **kwargs
    ) -> StreamingVideoIntelligenceServiceClient:
        """
        Creates an instance of this client using the provided credentials file.

        Parameters:
        - filename: The path to the service account private key json file

        Returns:
        StreamingVideoIntelligenceServiceClient: The constructed client
        """

    @classmethod
    def from_service_account_info(
        cls, info: dict, *args, **kwargs
    ) -> StreamingVideoIntelligenceServiceClient:
        """
        Creates an instance of this client using the provided credentials info.

        Parameters:
        - info: The service account private key info in dict format

        Returns:
        StreamingVideoIntelligenceServiceClient: The constructed client
        """

class StreamingVideoIntelligenceServiceAsyncClient:
    """
    Asynchronous client for Streaming Video Intelligence Service.
    
    Service that implements streaming video analysis using AI.
    """
    
    def __init__(
        self,
        *,
        credentials: Optional[ga_credentials.Credentials] = None,
        transport: Optional[Union[str, StreamingVideoIntelligenceServiceAsyncTransport]] = None,
        client_options: Optional[client_options_lib.ClientOptions] = None,
        client_info: gapic_v1.client_info.ClientInfo = DEFAULT_CLIENT_INFO,
    ):
        """
        Instantiate the async streaming 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 streaming_annotate_video(
        self,
        requests: AsyncIterator[Union[StreamingAnnotateVideoRequest, dict]],
        retry: OptionalRetry = gapic_v1.method.DEFAULT,
        timeout: Union[float, object] = gapic_v1.method.DEFAULT,
        metadata: Sequence[Tuple[str, Union[str, bytes]]] = (),
    ) -> AsyncIterable[StreamingAnnotateVideoResponse]:
        """
        Performs video annotation with streaming requests and responses.

        Parameters:
        - requests: AsyncIterator of StreamingAnnotateVideoRequest objects
        - 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:
        AsyncIterable of StreamingAnnotateVideoResponse objects containing real-time analysis results.
        """

    @classmethod
    def from_service_account_file(
        cls, filename: str, *args, **kwargs
    ) -> StreamingVideoIntelligenceServiceAsyncClient:
        """
        Creates an instance of this client using the provided credentials file.

        Parameters:
        - filename: The path to the service account private key json file

        Returns:
        StreamingVideoIntelligenceServiceAsyncClient: The constructed async client
        """

    @classmethod
    def from_service_account_info(
        cls, info: dict, *args, **kwargs
    ) -> StreamingVideoIntelligenceServiceAsyncClient:
        """
        Creates an instance of this client using the provided credentials info.

        Parameters:
        - info: The service account private key info in dict format

        Returns:
        StreamingVideoIntelligenceServiceAsyncClient: The constructed async client
        """

class StreamingAnnotateVideoRequest:
    """
    The streaming video annotation request message.
    
    Attributes:
        video_config: Provides information to the annotator that specifies how to process video
        input_content: The video data bytes
    """
    
    video_config: StreamingVideoConfig
    input_content: bytes

class StreamingAnnotateVideoResponse:
    """
    StreamingAnnotateVideoResponse contains analysis results for a stream.
    
    Attributes:
        error: If set, returns a google.rpc.Status message that specifies the error for the operation
        annotation_results: Streaming annotation results
        annotation_results_uri: Google Cloud Storage URI that stores annotation results
    """
    
    error: status_pb2.Status
    annotation_results: StreamingVideoAnnotationResults
    annotation_results_uri: str

class StreamingVideoConfig:
    """
    Config for streaming video content.
    
    Attributes:
        feature: Streaming video annotation feature
        storage_config: Config for streaming storage
        automl_action_recognition_config: Config for AUTOML_ACTION_RECOGNITION in streaming mode
        automl_classification_config: Config for AUTOML_CLASSIFICATION in streaming mode
        automl_object_tracking_config: Config for AUTOML_OBJECT_TRACKING in streaming mode
        explicit_content_detection_config: Config for EXPLICIT_CONTENT_DETECTION in streaming mode
        label_detection_config: Config for STREAMING_LABEL_DETECTION in streaming mode
        object_tracking_config: Config for STREAMING_OBJECT_TRACKING in streaming mode
        shot_change_detection_config: Config for SHOT_CHANGE_DETECTION in streaming mode
    """
    
    feature: StreamingFeature
    storage_config: StreamingStorageConfig
    automl_action_recognition_config: StreamingAutomlActionRecognitionConfig
    automl_classification_config: StreamingAutomlClassificationConfig
    automl_object_tracking_config: StreamingAutomlObjectTrackingConfig
    explicit_content_detection_config: StreamingExplicitContentDetectionConfig
    label_detection_config: StreamingLabelDetectionConfig
    object_tracking_config: StreamingObjectTrackingConfig
    shot_change_detection_config: StreamingShotChangeDetectionConfig

class StreamingFeature(Enum):
    """
    Streaming video annotation feature.
    
    Values:
        STREAMING_FEATURE_UNSPECIFIED: Unspecified streaming feature
        STREAMING_LABEL_DETECTION: Label detection in streaming mode
        STREAMING_SHOT_CHANGE_DETECTION: Shot change detection in streaming mode
        STREAMING_EXPLICIT_CONTENT_DETECTION: Explicit content detection in streaming mode
        STREAMING_OBJECT_TRACKING: Object tracking in streaming mode
        STREAMING_AUTOML_ACTION_RECOGNITION: AutoML action recognition in streaming mode
        STREAMING_AUTOML_CLASSIFICATION: AutoML classification in streaming mode
        STREAMING_AUTOML_OBJECT_TRACKING: AutoML object tracking in streaming mode
    """
    
    STREAMING_FEATURE_UNSPECIFIED = 0
    STREAMING_LABEL_DETECTION = 1
    STREAMING_SHOT_CHANGE_DETECTION = 2
    STREAMING_EXPLICIT_CONTENT_DETECTION = 3
    STREAMING_OBJECT_TRACKING = 4
    STREAMING_AUTOML_ACTION_RECOGNITION = 23
    STREAMING_AUTOML_CLASSIFICATION = 21
    STREAMING_AUTOML_OBJECT_TRACKING = 22

class StreamingVideoAnnotationResults:
    """
    Streaming annotation results corresponding to a portion of the video.
    
    Attributes:
        shot_annotations: Shot annotation results for the portion of the video
        label_annotations: Label annotation results for the portion of the video
        explicit_annotation: Explicit content annotation results for the portion of the video
        object_annotations: Object tracking annotation results for the portion of the video
    """
    
    shot_annotations: MutableSequence[VideoSegment]
    label_annotations: MutableSequence[LabelAnnotation]
    explicit_annotation: ExplicitContentAnnotation
    object_annotations: MutableSequence[ObjectTrackingAnnotation]

from google.cloud import videointelligence_v1p3beta1

def stream_video_analysis():
    # Create streaming client
    client = videointelligence_v1p3beta1.StreamingVideoIntelligenceServiceClient()
    
    # Configure streaming video analysis
    video_config = videointelligence_v1p3beta1.StreamingVideoConfig(
        feature=videointelligence_v1p3beta1.StreamingFeature.STREAMING_LABEL_DETECTION,
        label_detection_config=videointelligence_v1p3beta1.StreamingLabelDetectionConfig()
    )
    
    # Create request generator
    def request_generator():
        # Initial configuration request
        yield videointelligence_v1p3beta1.StreamingAnnotateVideoRequest(
            video_config=video_config
        )
        
        # Stream video data chunks
        with open('video_file.mp4', 'rb') as video_file:
            while True:
                chunk = video_file.read(1024 * 1024)  # 1MB chunks
                if not chunk:
                    break
                yield videointelligence_v1p3beta1.StreamingAnnotateVideoRequest(
                    input_content=chunk
                )
    
    # Process streaming responses
    responses = client.streaming_annotate_video(request_generator())
    
    for response in responses:
        if response.annotation_results:
            for label in response.annotation_results.label_annotations:
                print(f"Detected: {label.entity.description} (confidence: {label.segments[0].confidence})")

import asyncio
from google.cloud import videointelligence_v1p3beta1

async def async_stream_analysis():
    # Create async streaming client
    client = videointelligence_v1p3beta1.StreamingVideoIntelligenceServiceAsyncClient()
    
    # Configure for object tracking
    video_config = videointelligence_v1p3beta1.StreamingVideoConfig(
        feature=videointelligence_v1p3beta1.StreamingFeature.STREAMING_OBJECT_TRACKING,
        object_tracking_config=videointelligence_v1p3beta1.StreamingObjectTrackingConfig()
    )
    
    async def async_request_generator():
        # Configuration request
        yield videointelligence_v1p3beta1.StreamingAnnotateVideoRequest(
            video_config=video_config
        )
        
        # Simulate streaming video data
        for i in range(10):
            # In real usage, this would be actual video data
            fake_video_data = b'video_chunk_' + str(i).encode()
            yield videointelligence_v1p3beta1.StreamingAnnotateVideoRequest(
                input_content=fake_video_data
            )
            await asyncio.sleep(0.1)  # Simulate streaming delay
    
    # Process async streaming responses
    responses = client.streaming_annotate_video(async_request_generator())
    
    async for response in responses:
        if response.annotation_results and response.annotation_results.object_annotations:
            for obj in response.annotation_results.object_annotations:
                print(f"Tracked object: {obj.entity.description}")

# Run async streaming
asyncio.run(async_stream_analysis())

from google.cloud import videointelligence_v1p3beta1

# Multiple feature configuration
video_config = videointelligence_v1p3beta1.StreamingVideoConfig(
    feature=videointelligence_v1p3beta1.StreamingFeature.STREAMING_EXPLICIT_CONTENT_DETECTION,
    explicit_content_detection_config=videointelligence_v1p3beta1.StreamingExplicitContentDetectionConfig(),
    storage_config=videointelligence_v1p3beta1.StreamingStorageConfig(
        enable_storage_annotation_result=True,
        annotation_result_storage_directory="gs://your-bucket/results/"
    )
)