Application Platform

def list_applications(self, parent: str, page_size: int = None, page_token: str = None, filter: str = None, order_by: str = None) -> ListApplicationsResponse:
    """
    Lists applications in a project and location.
    
    Args:
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
        page_size (int): Maximum number of applications to return
        page_token (str): Token for pagination  
        filter (str): Filter expression for applications
        order_by (str): Sort order for results
        
    Returns:
        ListApplicationsResponse: Response with applications and next page token
    """

def get_application(self, name: str) -> Application:
    """
    Retrieves a specific application.
    
    Args:
        name (str): Required. Application resource path
                   "projects/{project}/locations/{location}/applications/{application}"
    
    Returns:
        Application: The application resource
    """

def create_application(self, parent: str, application: Application, application_id: str) -> Operation:
    """
    Creates a new Vision AI application.
    
    Args:
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
        application (Application): Required. Application configuration
        application_id (str): Required. ID for the new application
    
    Returns:
        Operation: Long-running operation for application creation
    """

def update_application(self, application: Application, update_mask: FieldMask = None) -> Operation:
    """
    Updates an existing application configuration.
    
    Args:
        application (Application): Required. Updated application configuration
        update_mask (FieldMask): Fields to update
    
    Returns:
        Operation: Long-running operation for application update
    """

def delete_application(self, name: str, force: bool = None) -> Operation:
    """
    Deletes a Vision AI application.
    
    Args:
        name (str): Required. Application resource path
        force (bool): Force deletion even if application is deployed
    
    Returns:
        Operation: Long-running operation for deletion
    """

def deploy_application(self, name: str) -> Operation:
    """
    Deploys an application to start processing video streams.
    
    Args:
        name (str): Required. Application resource path to deploy
    
    Returns:
        Operation: Long-running operation for deployment
    """

def undeploy_application(self, name: str) -> Operation:
    """
    Undeploys a running application to stop processing.
    
    Args:
        name (str): Required. Application resource path to undeploy
    
    Returns:
        Operation: Long-running operation for undeployment
    """

def add_application_stream_input(self, name: str, application_stream_inputs: List[ApplicationStreamInput]) -> Operation:
    """
    Adds stream input sources to a deployed application.
    
    Args:
        name (str): Required. Application resource path
        application_stream_inputs (List[ApplicationStreamInput]): Stream inputs to add
    
    Returns:
        Operation: Long-running operation for adding stream inputs
    """

def remove_application_stream_input(self, name: str, target_stream_inputs: List[RemoveApplicationStreamInputRequestTargetStreamInput]) -> Operation:
    """
    Removes stream input sources from a deployed application.
    
    Args:
        name (str): Required. Application resource path
        target_stream_inputs (List[RemoveApplicationStreamInputRequestTargetStreamInput]): 
                            Stream inputs to remove
    
    Returns:
        Operation: Long-running operation for removing stream inputs
    """

def update_application_stream_input(self, name: str, application_stream_inputs: List[ApplicationStreamInput]) -> Operation:
    """
    Updates stream input configuration for a deployed application.
    
    Args:
        name (str): Required. Application resource path
        application_stream_inputs (List[ApplicationStreamInput]): Updated stream input configs
    
    Returns:
        Operation: Long-running operation for updating stream inputs
    """

def list_instances(self, parent: str, page_size: int = None, page_token: str = None, filter: str = None, order_by: str = None) -> ListInstancesResponse:
    """
    Lists instances of an application.
    
    Args:
        parent (str): Required. Application resource path
        page_size (int): Maximum number of instances to return
        page_token (str): Token for pagination
        filter (str): Filter expression for instances
        order_by (str): Sort order for results
        
    Returns:
        ListInstancesResponse: Response with instances and pagination
    """

def get_instance(self, name: str) -> Instance:
    """
    Retrieves specific instance details.
    
    Args:
        name (str): Required. Instance resource path
                   "projects/{project}/locations/{location}/applications/{application}/instances/{instance}"
    
    Returns:
        Instance: The instance resource with status and configuration
    """

def create_application_instances(self, name: str, application_instances: List[ApplicationInstance]) -> Operation:
    """
    Creates new instances for an application.
    
    Args:
        name (str): Required. Application resource path
        application_instances (List[ApplicationInstance]): Instance configurations to create
    
    Returns:
        Operation: Long-running operation for instance creation
    """

def delete_application_instances(self, name: str, instance_ids: List[str]) -> Operation:
    """
    Deletes application instances.
    
    Args:
        name (str): Required. Application resource path
        instance_ids (List[str]): IDs of instances to delete
    
    Returns:
        Operation: Long-running operation for instance deletion
    """

def update_application_instances(self, name: str, application_instances: List[UpdateApplicationInstancesRequestUpdateApplicationInstance]) -> Operation:
    """
    Updates configuration for application instances.
    
    Args:
        name (str): Required. Application resource path
        application_instances (List[UpdateApplicationInstancesRequestUpdateApplicationInstance]):
                              Instance updates to apply
    
    Returns:
        Operation: Long-running operation for instance updates
    """

def list_drafts(self, parent: str, page_size: int = None, page_token: str = None, filter: str = None, order_by: str = None) -> ListDraftsResponse:
    """
    Lists drafts for an application.
    
    Args:
        parent (str): Required. Application resource path
        page_size (int): Maximum number of drafts to return
        page_token (str): Token for pagination
        filter (str): Filter expression for drafts
        order_by (str): Sort order for results
        
    Returns:
        ListDraftsResponse: Response with drafts and pagination
    """

def get_draft(self, name: str) -> Draft:
    """
    Retrieves a specific draft configuration.
    
    Args:
        name (str): Required. Draft resource path
                   "projects/{project}/locations/{location}/applications/{application}/drafts/{draft}"
    
    Returns:
        Draft: The draft resource with configuration
    """

def create_draft(self, parent: str, draft: Draft, draft_id: str) -> Operation:
    """
    Creates a new draft version of an application.
    
    Args:
        parent (str): Required. Application resource path
        draft (Draft): Required. Draft configuration
        draft_id (str): Required. ID for the new draft
    
    Returns:
        Operation: Long-running operation for draft creation
    """

def update_draft(self, draft: Draft, update_mask: FieldMask = None) -> Operation:
    """
    Updates an existing draft configuration.
    
    Args:
        draft (Draft): Required. Updated draft configuration
        update_mask (FieldMask): Fields to update
    
    Returns:
        Operation: Long-running operation for draft update
    """

def delete_draft(self, name: str) -> Operation:
    """
    Deletes an application draft.
    
    Args:
        name (str): Required. Draft resource path to delete
    
    Returns:
        Operation: Long-running operation for draft deletion
    """

def list_processors(self, parent: str, page_size: int = None, page_token: str = None, filter: str = None, order_by: str = None) -> ListProcessorsResponse:
    """
    Lists available processors.
    
    Args:
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
        page_size (int): Maximum number of processors to return
        page_token (str): Token for pagination
        filter (str): Filter expression for processors
        order_by (str): Sort order for results
        
    Returns:
        ListProcessorsResponse: Response with processors and pagination
    """

def list_prebuilt_processors(self, parent: str, page_size: int = None, page_token: str = None) -> ListPrebuiltProcessorsResponse:
    """
    Lists prebuilt processors available for use.
    
    Args:
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
        page_size (int): Maximum number to return
        page_token (str): Token for pagination
        
    Returns:
        ListPrebuiltProcessorsResponse: Response with prebuilt processors
    """

def get_processor(self, name: str) -> Processor:
    """
    Retrieves processor details and configuration.
    
    Args:
        name (str): Required. Processor resource path
                   "projects/{project}/locations/{location}/processors/{processor}"
    
    Returns:
        Processor: The processor resource with configuration
    """

def create_processor(self, parent: str, processor: Processor, processor_id: str) -> Operation:
    """
    Creates a custom processor for video analysis.
    
    Args:
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
        processor (Processor): Required. Processor configuration
        processor_id (str): Required. ID for the new processor
    
    Returns:
        Operation: Long-running operation for processor creation
    """

def update_processor(self, processor: Processor, update_mask: FieldMask = None) -> Operation:
    """
    Updates processor configuration.
    
    Args:
        processor (Processor): Required. Updated processor configuration
        update_mask (FieldMask): Fields to update
    
    Returns:
        Operation: Long-running operation for processor update
    """

def delete_processor(self, name: str) -> Operation:
    """
    Deletes a custom processor.
    
    Args:
        name (str): Required. Processor resource path to delete
    
    Returns:
        Operation: Long-running operation for processor deletion
    """

class Application:
    """Vision AI application configuration."""
    name: str  # Resource name
    display_name: str  # Human-readable name
    description: str  # Application description  
    application_configs: ApplicationConfigs  # Configuration settings
    runtime_info: ApplicationRuntimeInfo  # Runtime status information
    state: ApplicationState  # Current application state
    create_time: Timestamp  # Creation timestamp
    update_time: Timestamp  # Last update timestamp

class ApplicationConfigs:
    """Application configuration settings."""
    nodes: List[Node]  # Processing nodes in application graph
    event_delivery_config: ApplicationEventDeliveryConfig  # Event delivery settings

class Node:
    """Processing node in application graph."""
    name: str  # Node name
    display_name: str  # Human-readable name
    node_config: ProcessorConfig  # Node processing configuration
    processor: str  # Processor resource name
    parents: List[InputEdge]  # Input connections
    
class ProcessorConfig:
    """Configuration for a processor node."""
    vertex_automl_vision_config: VertexAutoMLVisionConfig
    vertex_automl_video_config: VertexAutoMLVideoConfig  
    vertex_custom_config: VertexCustomConfig
    general_object_detection_config: GeneralObjectDetectionConfig
    person_blur_config: PersonBlurConfig
    occupancy_count_config: OccupancyCountConfig
    person_vehicle_detection_config: PersonVehicleDetectionConfig
    personal_protective_equipment_detection_config: PersonalProtectiveEquipmentDetectionConfig
    video_stream_input_config: VideoStreamInputConfig
    ai_enabled_devices_input_config: AIEnabledDevicesInputConfig
    media_warehouse_config: MediaWarehouseConfig
    gcs_output_config: GcsOutputConfig
    bigquery_config: BigQueryConfig

class ApplicationInstance:
    """Application instance configuration."""
    instance_id: str  # Instance identifier
    input_resources: List[InputResource]  # Input resource mappings
    output_resources: List[OutputResource]  # Output resource mappings
    processor_to_input_resource_binding: List[ProcessorIOSpec]  # Processor bindings

class Draft:
    """Application draft version."""
    name: str  # Resource name
    application_configs: ApplicationConfigs  # Draft configuration
    create_time: Timestamp  # Creation timestamp
    update_time: Timestamp  # Last update timestamp

class Processor:
    """Custom processor definition.""" 
    name: str  # Resource name
    display_name: str  # Human-readable name
    description: str  # Processor description
    processor_type: ProcessorType  # Type of processor
    model_type: ModelType  # ML model type
    custom_processor_source_info: CustomProcessorSourceInfo  # Source information
    state: ProcessorState  # Current processor state
    processor_io_spec: ProcessorIOSpec  # Input/output specifications
    configuration_typeurl: str  # Configuration type URL
    supported_annotation_types: List[StreamAnnotationType]  # Supported annotations
    supports_post_processing: bool  # Post-processing support
    create_time: Timestamp  # Creation timestamp
    update_time: Timestamp  # Last update timestamp

class ProcessorIOSpec:
    """Processor input/output specifications."""
    graph_input_channel_specs: List[GraphInputChannelSpec]  # Input channel specs
    graph_output_channel_specs: List[GraphOutputChannelSpec]  # Output channel specs
    instance_resource_input_binding_specs: List[InstanceResourceInputBindingSpec]  # Input bindings
    instance_resource_output_binding_specs: List[InstanceResourceOutputBindingSpec]  # Output bindings

class ApplicationStreamInput:
    """Stream input configuration for applications."""
    stream_with_annotation: str  # Stream resource with annotations

class VideoStreamInputConfig:
    """Video stream input configuration."""
    streams: List[str]  # Stream resource names
    streams_with_annotation: List[str]  # Streams with annotation data

class AIEnabledDevicesInputConfig:
    """AI-enabled devices input configuration."""  
    pass  # Configuration for AI-enabled device inputs

class InputResource:
    """Input resource specification."""
    input_resource_key: str  # Resource key identifier
    input_resource_binding: InputResourceBinding  # Resource binding configuration

class InputResourceBinding:
    """Resource binding configuration."""
    resource_type: InputResourceBindingResourceType  # Type of resource
    resource: str  # Resource identifier
    
class OutputResource:
    """Output resource specification."""
    output_resource_key: str  # Resource key identifier  
    output_resource_binding: OutputResourceBinding  # Resource binding configuration

class OutputResourceBinding:
    """Output resource binding configuration."""
    resource_type: OutputResourceBindingResourceType  # Type of resource
    resource: str  # Resource identifier

class ApplicationState(Enum):
    """Application deployment states."""
    STATE_UNSPECIFIED = 0
    CREATED = 1  # Application created but not deployed
    DEPLOYING = 2  # Deployment in progress  
    DEPLOYED = 3  # Successfully deployed and running
    UNDEPLOYING = 4  # Undeployment in progress
    DELETED = 5  # Application deleted
    ERROR = 6  # Error state

class ProcessorType(Enum):
    """Types of processors."""
    PROCESSOR_TYPE_UNSPECIFIED = 0
    PRETRAINED_VERTEX_AI = 1  # Pretrained Vertex AI model
    CUSTOM_PROCESSOR = 2  # Custom processor
    CONNECTOR_LVA = 3  # Live video analytics connector
    CONNECTOR_STREAM = 4  # Stream connector

class ProcessorState(Enum):
    """Processor states."""
    STATE_UNSPECIFIED = 0  
    CREATING = 1  # Processor being created
    ACTIVE = 2  # Processor ready for use
    DELETING = 3  # Processor being deleted
    FAILED = 4  # Processor creation failed

class ModelType(Enum):
    """ML model types."""
    MODEL_TYPE_UNSPECIFIED = 0
    IMAGE_CLASSIFICATION = 1  # Image classification model
    OBJECT_DETECTION = 2  # Object detection model
    VIDEO_CLASSIFICATION = 3  # Video classification model
    VIDEO_OBJECT_TRACKING = 4  # Video object tracking model
    VIDEO_ACTION_RECOGNITION = 5  # Video action recognition model
    OCCUPANCY_COUNTING = 6  # Occupancy counting model
    PERSON_BLUR = 7  # Person blurring model
    VERTEX_CUSTOM = 8  # Custom Vertex AI model

from google.cloud import visionai_v1

# Create client
client = visionai_v1.AppPlatformClient()

# Define paths
parent = "projects/my-project/locations/us-central1"
app_name = f"{parent}/applications/my-video-app"

# Create application
application = visionai_v1.Application(
    display_name="My Video Analytics App",
    description="Real-time video analysis application"
)

create_op = client.create_application(
    parent=parent,
    application=application, 
    application_id="my-video-app"
)

# Wait for creation to complete
created_app = create_op.result()

# Deploy application
deploy_op = client.deploy_application(name=app_name)
deploy_op.result()  # Wait for deployment

# Add stream input
stream_inputs = [
    visionai_v1.ApplicationStreamInput(
        stream_with_annotation="projects/my-project/locations/us-central1/clusters/my-cluster/streams/camera-1"
    )
]

add_input_op = client.add_application_stream_input(
    name=app_name,
    application_stream_inputs=stream_inputs
)
add_input_op.result()

# List available prebuilt processors
prebuilt = client.list_prebuilt_processors(parent=parent)
for processor in prebuilt:
    print(f"Available: {processor.display_name}")

# Create custom processor  
custom_processor = visionai_v1.Processor(
    display_name="Custom Object Detector",
    description="Custom trained object detection model",
    processor_type=visionai_v1.ProcessorType.CUSTOM_PROCESSOR,
    model_type=visionai_v1.ModelType.OBJECT_DETECTION
)

create_processor_op = client.create_processor(
    parent=parent,
    processor=custom_processor,
    processor_id="custom-detector"
)

processor = create_processor_op.result()