tessl/pypi-google-cloud-video-live-stream

Google Cloud Video Live Stream API client library that transcodes mezzanine live signals into direct-to-consumer streaming formats, including Dynamic Adaptive Streaming over HTTP (DASH/MPEG-DASH), and HTTP Live Streaming (HLS), for multiple device platforms.

—

Pending

Overview

Eval results

Files

Clip Management

Name: tessl/pypi-google-cloud-video-live-stream
Author: tessl

Creation and management of video clips extracted from live streams or DVR sessions, allowing users to create highlights and shareable content segments.

Capabilities

Listing Clips

Retrieves a list of video clips with pagination and filtering capabilities.

def list_clips(
    self,
    request: Union[ListClipsRequest, dict] = None,
    *,
    parent: str = None,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Union[float, object] = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> pagers.ListClipsPager:
    """
    Returns a list of all clips in the specified channel.

    Args:
        request: The request object containing parent, page_size, page_token, filter, order_by
        parent: Required. The parent channel (projects/{project}/locations/{location}/channels/{channel})
        retry: Retry configuration for the request
        timeout: Request timeout in seconds
        metadata: Additional metadata for the request

    Returns:
        pagers.ListClipsPager: Pager for iterating over clip results
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

Getting Clip Details

Retrieves detailed information about a specific clip including its manifest URI, time range, and processing state.

def get_clip(
    self,
    request: Union[GetClipRequest, dict] = None,
    *,
    name: str = None,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Union[float, object] = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> Clip:
    """
    Returns the specified clip.

    Args:
        request: The request object containing name
        name: Required. The clip name (projects/{project}/locations/{location}/channels/{channel}/clips/{clip})
        retry: Retry configuration for the request
        timeout: Request timeout in seconds
        metadata: Additional metadata for the request

    Returns:
        Clip: The clip resource
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

Creating Clips

Creates a new video clip from a specified time range within a live stream or DVR session.

def create_clip(
    self,
    request: Union[CreateClipRequest, dict] = None,
    *,
    parent: str = None,
    clip: Clip = None,
    clip_id: str = None,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Union[float, object] = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> operation.Operation:
    """
    Creates a clip with the provided unique ID in the specified channel.

    Args:
        request: The request object containing parent, clip_id, clip, and request_id
        parent: Required. The parent channel (projects/{project}/locations/{location}/channels/{channel})
        clip: Required. The clip resource to create
        clip_id: Required. The ID to use for the clip (must be unique within parent)
        retry: Retry configuration for the request
        timeout: Request timeout in seconds
        metadata: Additional metadata for the request

    Returns:
        google.api_core.operation.Operation: Long-running operation
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

Deleting Clips

Permanently removes a video clip and all associated manifest and segment data.

def delete_clip(
    self,
    request: Union[DeleteClipRequest, dict] = None,
    *,
    name: str = None,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Union[float, object] = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> operation.Operation:
    """
    Deletes the specified clip.

    Args:
        request: The request object containing name and request_id
        name: Required. The clip name (projects/{project}/locations/{location}/channels/{channel}/clips/{clip})
        retry: Retry configuration for the request
        timeout: Request timeout in seconds
        metadata: Additional metadata for the request

    Returns:
        google.api_core.operation.Operation: Long-running operation
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

Clip Configuration Types

Clip Resource

class Clip:
    """
    Clip resource represents a video segment extracted from a live stream.
    
    Attributes:
        name (str): Clip resource name
        create_time (google.protobuf.timestamp_pb2.Timestamp): Creation timestamp
        update_time (google.protobuf.timestamp_pb2.Timestamp): Last update timestamp
        labels (MutableMapping[str, str]): User-defined labels
        start_time_offset (google.protobuf.duration_pb2.Duration): Start time offset from stream beginning
        end_time_offset (google.protobuf.duration_pb2.Duration): End time offset from stream beginning
        clip_manifest_uri (str): Output only. URI of the clip's HLS manifest
        state (State): Current clip processing state
        error (google.rpc.status_pb2.Status): Any processing errors
    """
    
    class State(proto.Enum):
        """Clip state enumeration."""
        STATE_UNSPECIFIED = 0
        CREATING = 1
        ACTIVE = 2
        FAILED = 3

Request Types

class CreateClipRequest:
    """
    Request message for CreateClip.
    
    Attributes:
        parent (str): Required. The parent channel resource name
        clip_id (str): Required. The ID to use for the clip
        clip (Clip): Required. The clip resource to create
        request_id (str): Optional. Idempotency key for request deduplication
    """

class ListClipsRequest:
    """
    Request message for ListClips.
    
    Attributes:
        parent (str): Required. The parent channel resource name
        page_size (int): Optional. Maximum number of clips to return per page
        page_token (str): Optional. Token for pagination
        filter (str): Optional. Filter expression for clips
        order_by (str): Optional. Field to sort clips by
    """

class ListClipsResponse:
    """
    Response message for ListClips.
    
    Attributes:
        clips (MutableSequence[Clip]): List of clips
        next_page_token (str): Token for next page of results
        unreachable (MutableSequence[str]): Locations that could not be reached
    """

class GetClipRequest:
    """
    Request message for GetClip.
    
    Attributes:
        name (str): Required. The clip resource name
    """

class DeleteClipRequest:
    """
    Request message for DeleteClip.
    
    Attributes:
        name (str): Required. The clip resource name
        request_id (str): Optional. Idempotency key for request deduplication
    """

Usage Examples

Creating a Clip from Live Stream

from google.cloud.video import live_stream_v1
from google.protobuf import duration_pb2

client = live_stream_v1.LivestreamServiceClient()

# Create 30-second clip starting 5 minutes into the stream
start_offset = duration_pb2.Duration(seconds=5*60)  # 5 minutes
end_offset = duration_pb2.Duration(seconds=5*60 + 30)  # 5 minutes 30 seconds

clip = live_stream_v1.Clip(
    start_time_offset=start_offset,
    end_time_offset=end_offset
)

request = live_stream_v1.CreateClipRequest(
    parent="projects/my-project/locations/us-central1/channels/my-channel",
    clip_id="highlight-clip-1",
    clip=clip
)

operation = client.create_clip(request=request)
clip_result = operation.result()
print(f"Created clip: {clip_result.name}")
print(f"Clip manifest URI: {clip_result.clip_manifest_uri}")

Creating Multiple Clips for Highlights

# Create multiple highlight clips
highlights = [
    {"start": 60, "duration": 30, "name": "goal-1"},      # 1:00-1:30
    {"start": 300, "duration": 45, "name": "goal-2"},     # 5:00-5:45
    {"start": 600, "duration": 20, "name": "save-1"},     # 10:00-10:20
]

clips_created = []

for highlight in highlights:
    start_offset = duration_pb2.Duration(seconds=highlight["start"])
    end_offset = duration_pb2.Duration(seconds=highlight["start"] + highlight["duration"])
    
    clip = live_stream_v1.Clip(
        start_time_offset=start_offset,
        end_time_offset=end_offset,
        labels={
            "type": "highlight",
            "category": highlight["name"].split("-")[0]
        }
    )
    
    request = live_stream_v1.CreateClipRequest(
        parent="projects/my-project/locations/us-central1/channels/sports-broadcast",
        clip_id=f"highlight-{highlight['name']}",
        clip=clip
    )
    
    operation = client.create_clip(request=request)
    clip_result = operation.result()
    clips_created.append(clip_result)
    print(f"Created {highlight['name']}: {clip_result.clip_manifest_uri}")

Monitoring Clip Processing

# Check clip processing status
clip_name = "projects/my-project/locations/us-central1/channels/my-channel/clips/my-clip"

get_request = live_stream_v1.GetClipRequest(name=clip_name)
clip = client.get_clip(request=get_request)

print(f"Clip state: {clip.state}")
if clip.state == live_stream_v1.Clip.State.ACTIVE:
    print(f"Clip ready: {clip.clip_manifest_uri}")
elif clip.state == live_stream_v1.Clip.State.FAILED:
    print(f"Clip failed: {clip.error.message}")
elif clip.state == live_stream_v1.Clip.State.CREATING:
    print("Clip is still processing...")

Listing and Managing Clips

# List all clips in a channel
list_request = live_stream_v1.ListClipsRequest(
    parent="projects/my-project/locations/us-central1/channels/my-channel",
    page_size=50
)

print("Available clips:")
for clip in client.list_clips(request=list_request):
    duration_seconds = clip.end_time_offset.seconds - clip.start_time_offset.seconds
    print(f"- {clip.name}: {duration_seconds}s clip ({clip.state})")
    if clip.state == live_stream_v1.Clip.State.ACTIVE:
        print(f"  Manifest: {clip.clip_manifest_uri}")

# Filter clips by label
filtered_request = live_stream_v1.ListClipsRequest(
    parent="projects/my-project/locations/us-central1/channels/my-channel",
    filter='labels.type="highlight"'
)

print("\nHighlight clips:")
for clip in client.list_clips(request=filtered_request):
    print(f"- {clip.name}")

Clip Cleanup Operations

# Delete old clips
import datetime
from google.protobuf import timestamp_pb2

# Delete clips older than 7 days
cutoff_time = datetime.datetime.now() - datetime.timedelta(days=7)
cutoff_timestamp = timestamp_pb2.Timestamp()
cutoff_timestamp.FromDatetime(cutoff_time)

list_request = live_stream_v1.ListClipsRequest(
    parent="projects/my-project/locations/us-central1/channels/my-channel"
)

for clip in client.list_clips(request=list_request):
    if clip.create_time < cutoff_timestamp:
        print(f"Deleting old clip: {clip.name}")
        delete_request = live_stream_v1.DeleteClipRequest(name=clip.name)
        delete_operation = client.delete_clip(request=delete_request)
        delete_operation.result()  # Wait for completion
        print(f"Deleted: {clip.name}")

Creating Clips with Custom Time Ranges

# Create clip from DVR session with precise timing
from google.protobuf import timestamp_pb2

# Clip from 2:15:30 to 2:18:45 (2 hours 15 minutes 30 seconds to 2 hours 18 minutes 45 seconds)
start_seconds = 2*3600 + 15*60 + 30  # 2:15:30
end_seconds = 2*3600 + 18*60 + 45    # 2:18:45

clip = live_stream_v1.Clip(
    start_time_offset=duration_pb2.Duration(seconds=start_seconds),
    end_time_offset=duration_pb2.Duration(seconds=end_seconds),
    labels={
        "event": "interview_segment",
        "speaker": "guest_expert",
        "topic": "technology_trends"
    }
)

request = live_stream_v1.CreateClipRequest(
    parent="projects/my-project/locations/us-central1/channels/interview-show",
    clip_id="tech-trends-interview",
    clip=clip
)

operation = client.create_clip(request=request)
clip_result = operation.result()
print(f"Interview clip created: {clip_result.clip_manifest_uri}")

Install with Tessl CLI

npx tessl i tessl/pypi-google-cloud-video-live-stream

docs

asset-dvr-management.md

channel-management.md