0
# Live Video Analytics
1

2
Real-time video analysis capabilities with custom operators, analysis definitions, and batch processing workflows. The LiveVideoAnalytics service provides the foundation for building sophisticated video understanding applications with ML-powered processing pipelines.
3

4
## Capabilities
5

6
### Analysis Management
7

8
Create and manage video analysis definitions that specify processing workflows, operators, and output configurations.
9

10
```python { .api }
11
def list_analyses(self, parent: str, page_size: int = None, page_token: str = None, filter: str = None, order_by: str = None) -> ListAnalysesResponse:
12
    """
13
    Lists analyses in a project and location.
14
    
15
    Args:
16
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
17
        page_size (int): Maximum number of analyses to return
18
        page_token (str): Token for pagination
19
        filter (str): Filter expression for analyses
20
        order_by (str): Sort order for results
21
        
22
    Returns:
23
        ListAnalysesResponse: Response with analyses and pagination
24
    """
25

26
def get_analysis(self, name: str) -> Analysis:
27
    """
28
    Retrieves a specific analysis configuration.
29
    
30
    Args:
31
        name (str): Required. Analysis resource path
32
                   "projects/{project}/locations/{location}/analyses/{analysis}"
33
    
34
    Returns:
35
        Analysis: The analysis resource with configuration
36
    """
37

38
def create_analysis(self, parent: str, analysis: Analysis, analysis_id: str) -> Operation:
39
    """
40
    Creates a new video analysis definition.
41
    
42
    Args:
43
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
44
        analysis (Analysis): Required. Analysis configuration
45
        analysis_id (str): Required. ID for the new analysis
46
    
47
    Returns:
48
        Operation: Long-running operation for analysis creation
49
    """
50

51
def update_analysis(self, analysis: Analysis, update_mask: FieldMask = None) -> Operation:
52
    """
53
    Updates an existing analysis configuration.
54
    
55
    Args:
56
        analysis (Analysis): Required. Updated analysis configuration
57
        update_mask (FieldMask): Fields to update
58
    
59
    Returns:
60
        Operation: Long-running operation for analysis update
61
    """
62

63
def delete_analysis(self, name: str) -> Operation:
64
    """
65
    Deletes a video analysis definition.
66
    
67
    Args:
68
        name (str): Required. Analysis resource path to delete
69
    
70
    Returns:
71
        Operation: Long-running operation for deletion
72
    """
73
```
74

75
### Operator Management
76

77
Manage custom and public operators that perform specific video analysis tasks within processing pipelines.
78

79
```python { .api }
80
def list_operators(self, parent: str, page_size: int = None, page_token: str = None, filter: str = None, order_by: str = None) -> ListOperatorsResponse:
81
    """
82
    Lists user-created operators.
83
    
84
    Args:
85
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
86
        page_size (int): Maximum number of operators to return
87
        page_token (str): Token for pagination
88
        filter (str): Filter expression for operators
89
        order_by (str): Sort order for results
90
        
91
    Returns:
92
        ListOperatorsResponse: Response with operators and pagination
93
    """
94

95
def list_public_operators(self, parent: str, page_size: int = None, page_token: str = None, filter: str = None, order_by: str = None) -> ListPublicOperatorsResponse:
96
    """
97
    Lists public operators available for use.
98
    
99
    Args:
100
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
101
        page_size (int): Maximum number to return
102
        page_token (str): Token for pagination
103
        filter (str): Filter expression
104
        order_by (str): Sort order for results
105
        
106
    Returns:
107
        ListPublicOperatorsResponse: Response with public operators
108
    """
109

110
def resolve_operator_info(self, parent: str, queries: List[OperatorQuery]) -> ResolveOperatorInfoResponse:
111
    """
112
    Resolves detailed information about operators.
113
    
114
    Args:
115
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
116
        queries (List[OperatorQuery]): Operator queries to resolve
117
    
118
    Returns:
119
        ResolveOperatorInfoResponse: Resolved operator information
120
    """
121

122
def get_operator(self, name: str) -> Operator:
123
    """
124
    Retrieves operator details and configuration.
125
    
126
    Args:
127
        name (str): Required. Operator resource path
128
                   "projects/{project}/locations/{location}/operators/{operator}"
129
    
130
    Returns:
131
        Operator: The operator resource with configuration
132
    """
133

134
def create_operator(self, parent: str, operator: Operator, operator_id: str) -> Operation:
135
    """
136
    Creates a custom operator for video analysis.
137
    
138
    Args:
139
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
140
        operator (Operator): Required. Operator configuration
141
        operator_id (str): Required. ID for the new operator
142
    
143
    Returns:
144
        Operation: Long-running operation for operator creation
145
    """
146

147
def update_operator(self, operator: Operator, update_mask: FieldMask = None) -> Operation:
148
    """
149
    Updates operator configuration.
150
    
151
    Args:
152
        operator (Operator): Required. Updated operator configuration
153
        update_mask (FieldMask): Fields to update
154
    
155
    Returns:
156
        Operation: Long-running operation for operator update
157
    """
158

159
def delete_operator(self, name: str) -> Operation:
160
    """
161
    Deletes a custom operator.
162
    
163
    Args:
164
        name (str): Required. Operator resource path to delete
165
    
166
    Returns:
167
        Operation: Long-running operation for operator deletion
168
    """
169
```
170

171
### Process Management
172

173
Manage analysis processes that execute video analysis workflows on specific data sources with batch and streaming modes.
174

175
```python { .api }
176
def list_processes(self, parent: str, page_size: int = None, page_token: str = None, filter: str = None, order_by: str = None) -> ListProcessesResponse:
177
    """
178
    Lists processes in a project and location.
179
    
180
    Args:
181
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
182
        page_size (int): Maximum number of processes to return
183
        page_token (str): Token for pagination
184
        filter (str): Filter expression for processes
185
        order_by (str): Sort order for results
186
        
187
    Returns:
188
        ListProcessesResponse: Response with processes and pagination
189
    """
190

191
def get_process(self, name: str) -> Process:
192
    """
193
    Retrieves process details and status.
194
    
195
    Args:
196
        name (str): Required. Process resource path
197
                   "projects/{project}/locations/{location}/processes/{process}"
198
    
199
    Returns:
200
        Process: The process resource with configuration and status
201
    """
202

203
def create_process(self, parent: str, process: Process, process_id: str) -> Operation:
204
    """
205
    Creates a new analysis process.
206
    
207
    Args:
208
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
209
        process (Process): Required. Process configuration
210
        process_id (str): Required. ID for the new process
211
    
212
    Returns:
213
        Operation: Long-running operation for process creation
214
    """
215

216
def update_process(self, process: Process, update_mask: FieldMask = None) -> Operation:
217
    """
218
    Updates process configuration.
219
    
220
    Args:
221
        process (Process): Required. Updated process configuration
222
        update_mask (FieldMask): Fields to update
223
    
224
    Returns:
225
        Operation: Long-running operation for process update
226
    """
227

228
def delete_process(self, name: str) -> Operation:
229
    """
230
    Deletes an analysis process.
231
    
232
    Args:
233
        name (str): Required. Process resource path to delete
234
    
235
    Returns:
236
        Operation: Long-running operation for process deletion
237
    """
238

239
def batch_run_process(self, parent: str, requests: List[CreateProcessRequest]) -> Operation:
240
    """
241
    Runs multiple processes in batch mode.
242
    
243
    Args:
244
        parent (str): Required. Parent resource path "projects/{project}/locations/{location}"
245
        requests (List[CreateProcessRequest]): Process creation requests to execute
246
    
247
    Returns:
248
        Operation: Long-running operation for batch process execution
249
    """
250
```
251

252
## Types
253

254
### Analysis Resources
255

256
```python { .api }
257
class Analysis:
258
    """Video analysis definition and configuration."""
259
    name: str  # Resource name
260
    analysis_definition: AnalysisDefinition  # Analysis workflow definition
261
    input_streams_mapping: Dict[str, str]  # Input stream mappings
262
    output_streams_mapping: Dict[str, str]  # Output stream mappings
263
    disable_event_watch: bool  # Disable event watching
264
    create_time: Timestamp  # Creation timestamp
265
    update_time: Timestamp  # Last update timestamp
266
    labels: Dict[str, str]  # Resource labels
267

268
class AnalysisDefinition:
269
    """Definition of analysis workflow and operators."""
270
    analyzers: List[AnalyzerDefinition]  # Analyzer configurations
271
    
272
class AnalyzerDefinition:
273
    """Individual analyzer configuration within analysis."""
274
    analyzer: str  # Analyzer identifier
275
    operator: str  # Operator resource name
276
    inputs: List[AnalyzerInput]  # Input configurations
277
    attrs: Dict[str, AttributeValue]  # Analyzer attributes
278
    debug_options: AnalyzerDebugOptions  # Debug configuration
279

280
class AnalyzerInput:
281
    """Input configuration for analyzer."""
282
    input: str  # Input identifier
283
    
284
class AttributeValue:
285
    """Attribute value for analyzer configuration."""
286
    # Union field oneof value:
287
    i: int  # Integer value
288
    f: float  # Float value  
289
    b: bool  # Boolean value
290
    s: bytes  # String/bytes value
291

292
class AnalyzerDebugOptions:
293
    """Debug options for analyzer."""
294
    environment_variables: Dict[str, str]  # Environment variables
295
```
296

297
### Operator Resources
298

299
```python { .api }
300
class Operator:
301
    """Custom operator for video analysis."""
302
    name: str  # Resource name
303
    operator_definition: OperatorDefinition  # Operator implementation
304
    docker_image: str  # Docker image for operator
305
    create_time: Timestamp  # Creation timestamp
306
    update_time: Timestamp  # Last update timestamp
307
    labels: Dict[str, str]  # Resource labels
308

309
class OperatorDefinition:
310
    """Definition of operator implementation."""
311
    operator: str  # Operator identifier
312
    input_args: List[OperatorInputArg]  # Input argument definitions
313
    output_args: List[OperatorOutputArg]  # Output argument definitions
314
    attrs: List[OperatorAttribute]  # Operator attributes
315
    resources: ResourceSpecification  # Resource requirements
316
    short_description: str  # Brief description
317
    description: str  # Detailed description
318

319
class OperatorInputArg:
320
    """Input argument definition for operator.""" 
321
    argument: str  # Argument name
322
    type: str  # Argument type
323
    
324
class OperatorOutputArg:
325
    """Output argument definition for operator."""
326
    argument: str  # Argument name  
327
    type: str  # Argument type
328

329
class OperatorAttribute:
330
    """Attribute definition for operator."""
331
    attribute: str  # Attribute name
332
    type: str  # Attribute type
333
    default_value: AttributeValue  # Default value
334

335
class OperatorQuery:
336
    """Query for resolving operator information."""
337
    operator: str  # Operator identifier to query
338
    tag: str  # Operator version tag
339
    registry: Registry  # Registry containing operator
340

341
class Registry:
342
    """Registry configuration for operators."""
343
    # Union field oneof registry:
344
    public_registry: PublicRegistry  # Public registry
345
    private_registry: PrivateRegistry  # Private registry
346

347
class PublicRegistry:
348
    """Public operator registry configuration."""
349
    pass
350

351
class PrivateRegistry:
352
    """Private operator registry configuration."""  
353
    pass
354
```
355

356
### Process Resources
357

358
```python { .api }
359
class Process:
360
    """Analysis process configuration and status."""
361
    name: str  # Resource name
362
    analysis: str  # Analysis resource name
363
    attribute_overrides: List[str]  # Attribute overrides
364
    run_status: RunStatus  # Current run status
365
    run_mode: RunMode  # Execution mode
366
    event_id: str  # Associated event ID
367
    batch_id: str  # Batch execution ID  
368
    retry_count: int  # Number of retries
369
    create_time: Timestamp  # Creation timestamp
370
    update_time: Timestamp  # Last update timestamp
371
    labels: Dict[str, str]  # Resource labels
372

373
class CreateProcessRequest:
374
    """Request for creating a process."""
375
    parent: str  # Parent resource path
376
    process_id: str  # Process identifier
377
    process: Process  # Process configuration
378

379
class ResourceSpecification:
380
    """Resource requirements specification."""
381
    resource_type: ResourceSpecificationType  # Type of resource
382
    # Union field oneof resource:
383
    cpu: ResourceSpecificationCPU  # CPU resources
384
    gpu: ResourceSpecificationGPU  # GPU resources
385
    
386
class ResourceSpecificationCPU:
387
    """CPU resource specification."""
388
    cpu_limit: float  # CPU limit
389
    memory_limit: int  # Memory limit in bytes
390

391
class ResourceSpecificationGPU:
392
    """GPU resource specification."""
393
    gpu_limit: int  # Number of GPUs
394
    gpu_type: str  # GPU type specification
395
```
396

397
### Enums
398

399
```python { .api }
400
class RunStatus(Enum):
401
    """Process execution status."""
402
    RUN_STATUS_UNSPECIFIED = 0
403
    INITIALIZING = 1  # Process initializing
404
    RUNNING = 2  # Process running
405
    COMPLETED = 3  # Process completed successfully
406
    FAILED = 4  # Process failed
407
    PENDING = 5  # Process pending execution
408
    CANCELLING = 6  # Process being cancelled
409
    CANCELLED = 7  # Process cancelled
410

411
class RunMode(Enum):
412
    """Process execution mode."""
413
    RUN_MODE_UNSPECIFIED = 0  
414
    LIVE = 1  # Live/streaming mode
415
    OFFLINE = 2  # Batch/offline mode
416

417
class ResourceSpecificationType(Enum):
418
    """Resource specification types."""
419
    RESOURCE_SPECIFICATION_TYPE_UNSPECIFIED = 0
420
    CPU = 1  # CPU resources
421
    GPU = 2  # GPU resources
422
    CUSTOM = 3  # Custom resource type
423
```
424

425
## Usage Examples
426

427
### Creating Video Analysis Pipeline
428

429
```python
430
from google.cloud import visionai_v1
431

432
# Create client
433
client = visionai_v1.LiveVideoAnalyticsClient()
434

435
# Define paths
436
parent = "projects/my-project/locations/us-central1"
437

438
# Create analysis definition
439
analysis_def = visionai_v1.AnalysisDefinition(
440
    analyzers=[
441
        visionai_v1.AnalyzerDefinition(
442
            analyzer="object-detector",
443
            operator="google/object_detection:latest",
444
            inputs=[
445
                visionai_v1.AnalyzerInput(input="video-stream")
446
            ],
447
            attrs={
448
                "confidence_threshold": visionai_v1.AttributeValue(f=0.7),
449
                "max_detections": visionai_v1.AttributeValue(i=100)
450
            }
451
        ),
452
        visionai_v1.AnalyzerDefinition(
453
            analyzer="person-blur",
454
            operator="google/person_blur:latest", 
455
            inputs=[
456
                visionai_v1.AnalyzerInput(input="video-stream")
457
            ]
458
        )
459
    ]
460
)
461

462
# Create analysis
463
analysis = visionai_v1.Analysis(
464
    analysis_definition=analysis_def,
465
    input_streams_mapping={
466
        "video-stream": "projects/my-project/locations/us-central1/clusters/my-cluster/streams/camera-1"
467
    },
468
    output_streams_mapping={
469
        "analyzed-stream": "projects/my-project/locations/us-central1/clusters/my-cluster/streams/output"
470
    }
471
)
472

473
create_op = client.create_analysis(
474
    parent=parent,
475
    analysis=analysis,
476
    analysis_id="security-analysis"
477
)
478

479
created_analysis = create_op.result()
480
```
481

482
### Custom Operator Creation
483

484
```python
485
# Define custom operator
486
operator_def = visionai_v1.OperatorDefinition(
487
    operator="my-custom-detector",
488
    input_args=[
489
        visionai_v1.OperatorInputArg(
490
            argument="video",
491
            type="VIDEO_STREAM"
492
        )
493
    ],
494
    output_args=[
495
        visionai_v1.OperatorOutputArg(
496
            argument="detections",
497
            type="OBJECT_DETECTION_PREDICTION"
498
        )
499
    ],
500
    attrs=[
501
        visionai_v1.OperatorAttribute(
502
            attribute="model_path",
503
            type="STRING",
504
            default_value=visionai_v1.AttributeValue(s=b"/models/detector.pb")
505
        )
506
    ],
507
    resources=visionai_v1.ResourceSpecification(
508
        resource_type=visionai_v1.ResourceSpecificationType.GPU,
509
        gpu=visionai_v1.ResourceSpecificationGPU(
510
            gpu_limit=1,
511
            gpu_type="nvidia-tesla-t4"
512
        )
513
    ),
514
    short_description="Custom object detector",
515
    description="Custom trained object detection model"
516
)
517

518
# Create operator
519
operator = visionai_v1.Operator(
520
    operator_definition=operator_def,
521
    docker_image="gcr.io/my-project/custom-detector:v1.0"
522
)
523

524
create_op = client.create_operator(
525
    parent=parent,
526
    operator=operator,
527
    operator_id="custom-detector"
528
)
529

530
created_operator = create_op.result()
531
```
532

533
### Batch Process Execution
534

535
```python
536
# Create multiple processes for batch execution
537
process_requests = [
538
    visionai_v1.CreateProcessRequest(
539
        parent=parent,
540
        process_id=f"batch-process-{i}",
541
        process=visionai_v1.Process(
542
            analysis=f"{parent}/analyses/security-analysis",
543
            run_mode=visionai_v1.RunMode.OFFLINE,
544
            batch_id="batch-001"
545
        )
546
    ) 
547
    for i in range(5)
548
]
549

550
# Execute batch
551
batch_op = client.batch_run_process(
552
    parent=parent,
553
    requests=process_requests
554
)
555

556
batch_result = batch_op.result()
557

558
# Monitor process status
559
for request in process_requests:
560
    process_name = f"{parent}/processes/{request.process_id}"
561
    process = client.get_process(name=process_name)
562
    print(f"Process {request.process_id}: {process.run_status}")
563
```