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

Event Management

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

Creation and management of channel events for dynamic overlay insertion, ad breaks, and other time-based modifications to live streams during broadcast.

Capabilities

Creating Events

Creates a new event for a channel that can perform actions like input switching, ad breaks, slate insertion, or mute/unmute operations.

def create_event(
    self,
    request: Union[CreateEventRequest, dict] = None,
    *,
    parent: str = None,
    event: Event = None,
    event_id: str = None,
    retry: OptionalRetry = gapic_v1.method.DEFAULT,
    timeout: Union[float, object] = gapic_v1.method.DEFAULT,
    metadata: Sequence[Tuple[str, Union[str, bytes]]] = ()
) -> Event:
    """
    Creates an event with the provided unique ID in the specified channel.

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

    Returns:
        Event: The created event resource
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

Listing Events

Retrieves a list of events for a channel with pagination and filtering capabilities.

def list_events(
    self,
    request: Union[ListEventsRequest, 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.ListEventsPager:
    """
    Returns a list of all events 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.ListEventsPager: Pager for iterating over event results
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

Getting Event Details

Retrieves detailed information about a specific event including its configuration, state, and execution status.

def get_event(
    self,
    request: Union[GetEventRequest, 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]]] = ()
) -> Event:
    """
    Returns the specified event.

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

    Returns:
        Event: The event resource
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

Deleting Events

Permanently removes an event from a channel, canceling it if currently scheduled or executing.

def delete_event(
    self,
    request: Union[DeleteEventRequest, 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]]] = ()
) -> None:
    """
    Deletes the specified event.

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

    Returns:
        None
        
    Raises:
        google.api_core.exceptions.GoogleAPICallError: If the request fails
    """

Event Configuration Types

Event Resource

class Event:
    """
    Event resource for dynamic channel modifications during live streaming.
    
    Attributes:
        name (str): Event 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
        input_switch (InputSwitchTask): Input switching task configuration
        ad_break (AdBreakTask): Ad break task configuration
        return_to_program (ReturnToProgramTask): Return to program task configuration
        slate (SlateTask): Slate insertion task configuration  
        mute (MuteTask): Audio mute task configuration
        unmute (UnmuteTask): Audio unmute task configuration
        execute_now (bool): Execute immediately upon creation
        execution_time (google.protobuf.timestamp_pb2.Timestamp): Scheduled execution time
        state (State): Current event state
        error (google.rpc.status_pb2.Status): Any execution errors
    """
    
    class State(proto.Enum):
        """Event state enumeration."""
        STATE_UNSPECIFIED = 0
        SCHEDULED = 1
        RUNNING = 2
        SUCCEEDED = 3
        FAILED = 4
        PENDING = 5
        STOPPED = 6

Event Task Types

class InputSwitchTask:
    """
    Task to switch the active input source.
    
    Attributes:
        input_key (str): Input attachment key to switch to
    """

class AdBreakTask:
    """
    Task to insert an ad break with duration and slate content.
    
    Attributes:
        duration (google.protobuf.duration_pb2.Duration): Ad break duration
        assets (MutableSequence[str]): Asset resource names for ad content
    """

class ReturnToProgramTask:
    """
    Task to return to the main program content after an ad break.
    """

class SlateTask:
    """
    Task to display a static slate image or video.
    
    Attributes:
        duration (google.protobuf.duration_pb2.Duration): Slate display duration
        asset (str): Asset resource name containing slate content
    """

class MuteTask:
    """
    Task to mute audio output.
    
    Attributes:
        duration (google.protobuf.duration_pb2.Duration): Mute duration
    """

class UnmuteTask:
    """
    Task to unmute audio output.
    """

Usage Examples

Creating Input Switch Event

from google.cloud.video import live_stream_v1
from google.protobuf import timestamp_pb2
import datetime

client = live_stream_v1.LivestreamServiceClient()

# Schedule input switch for specific time
execution_time = datetime.datetime.now() + datetime.timedelta(minutes=5)
execution_timestamp = timestamp_pb2.Timestamp()
execution_timestamp.FromDatetime(execution_time)

event = live_stream_v1.Event(
    input_switch=live_stream_v1.InputSwitchTask(
        input_key="backup-input"
    ),
    execution_time=execution_timestamp
)

request = live_stream_v1.CreateEventRequest(
    parent="projects/my-project/locations/us-central1/channels/my-channel",
    event_id="switch-to-backup",
    event=event
)

event_result = client.create_event(request=request)
print(f"Created event: {event_result.name}")

Creating Ad Break Event

from google.protobuf import duration_pb2

# Create 30-second ad break with assets
ad_duration = duration_pb2.Duration(seconds=30)

event = live_stream_v1.Event(
    ad_break=live_stream_v1.AdBreakTask(
        duration=ad_duration,
        assets=[
            "projects/my-project/locations/us-central1/assets/commercial-1",
            "projects/my-project/locations/us-central1/assets/commercial-2"
        ]
    ),
    execute_now=True  # Execute immediately
)

request = live_stream_v1.CreateEventRequest(
    parent="projects/my-project/locations/us-central1/channels/my-channel",
    event_id="ad-break-30s",
    event=event
)

event_result = client.create_event(request=request)

Creating Slate Event

# Display slate for 10 seconds
slate_duration = duration_pb2.Duration(seconds=10)

event = live_stream_v1.Event(
    slate=live_stream_v1.SlateTask(
        duration=slate_duration,
        asset="projects/my-project/locations/us-central1/assets/technical-difficulties"
    ),
    execute_now=True
)

request = live_stream_v1.CreateEventRequest(
    parent="projects/my-project/locations/us-central1/channels/my-channel",
    event_id="technical-difficulties",
    event=event
)

Creating Mute/Unmute Events

# Mute audio for 5 seconds
mute_duration = duration_pb2.Duration(seconds=5)

mute_event = live_stream_v1.Event(
    mute=live_stream_v1.MuteTask(
        duration=mute_duration
    ),
    execute_now=True
)

# Create unmute event to restore audio
unmute_event = live_stream_v1.Event(
    unmute=live_stream_v1.UnmuteTask(),
    execute_now=False,
    execution_time=timestamp_pb2.Timestamp()  # Set appropriate time
)

# Create both events
mute_request = live_stream_v1.CreateEventRequest(
    parent="projects/my-project/locations/us-central1/channels/my-channel",
    event_id="mute-audio",
    event=mute_event
)

unmute_request = live_stream_v1.CreateEventRequest(
    parent="projects/my-project/locations/us-central1/channels/my-channel",
    event_id="unmute-audio", 
    event=unmute_event
)

Monitoring Event Execution

# Check event state
event_name = "projects/my-project/locations/us-central1/channels/my-channel/events/my-event"

get_request = live_stream_v1.GetEventRequest(name=event_name)
event = client.get_event(request=get_request)

print(f"Event state: {event.state}")
if event.error.code != 0:
    print(f"Event error: {event.error.message}")

# List all events in channel
list_request = live_stream_v1.ListEventsRequest(
    parent="projects/my-project/locations/us-central1/channels/my-channel"
)

for event in client.list_events(request=list_request):
    print(f"Event {event.name}: {event.state}")

Install with Tessl CLI

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

docs

asset-dvr-management.md

channel-management.md

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