- "api: videolivestream"

This directory contains samples for the Live Stream API. Use this API to

transcode live, linear video streams into a variety of formats. The Live Stream

API benefits broadcasters, production companies, businesses, and individuals

looking to transform their live video content for use across a variety of user

devices. For more information, see the

[Live Stream API documentation](https://cloud.google.com/livestream/).

To run the samples, you need to first follow the steps in

[Before you begin](https://cloud.google.com/livestream/docs/how-to/before-you-begin).

For more information on authentication, refer to the

[Authentication Getting Started Guide](https://cloud.google.com/docs/authentication/getting-started).

1. Clone python-docs-samples repository and change directories to the sample directory

you want to use.

$ git clone https://github.com/GoogleCloudPlatform/python-docs-samples.git

1. Install [pip](https://pip.pypa.io/) and

[virtualenv](https://virtualenv.pypa.io/) if you do not already have them. You

may want to refer to the

[Python Development Environment Setup Guide](https://cloud.google.com/python/setup)

for Google Cloud Platform for instructions.

1. Create a virtualenv. Samples are compatible with Python 3.6+.

$ virtualenv env

$ source env/bin/activate

1. Install the dependencies needed to run the samples.

$ pip install -r requirements.txt

Make sure to enable the Live Stream API on the test project. Set the following

environment variable:

* `GOOGLE_CLOUD_PROJECT`

import os

import uuid

from google.api_core.exceptions import FailedPrecondition, NotFound

from google.protobuf import empty_pb2 as empty

import pytest

import create_asset

import delete_asset

import get_asset

import list_assets

import utils

project_name = os.environ["GOOGLE_CLOUD_PROJECT"]

location = "us-central1"

asset_id = f"my-python-test-asset-{uuid.uuid4()}"

asset_uri = "gs://cloud-samples-data/media/ForBiggerEscapes.mp4"

def test_asset_operations(capsys: pytest.fixture) -> None:

# Clean up old resources in the test project

responses = list_assets.list_assets(project_name, location)

for response in responses:

next_asset_id = response.name.rsplit("/", 1)[-1]

if utils.is_resource_stale(response.create_time):

try:

delete_asset.delete_asset(project_name, location, next_asset_id)

except FailedPrecondition as e:

print(f"Ignoring FailedPrecondition, details: {e}")

except NotFound as e:

print(f"Ignoring NotFound, details: {e}")

asset_name_project_id = (

f"projects/{project_name}/locations/{location}/assets/{asset_id}"

)

# Tests

response = create_asset.create_asset(project_name, location, asset_id, asset_uri)

assert asset_name_project_id in response.name

list_assets.list_assets(project_name, location)

out, _ = capsys.readouterr()

assert asset_name_project_id in out

response = get_asset.get_asset(project_name, location, asset_id)

assert asset_name_project_id in response.name

response = delete_asset.delete_asset(project_name, location, asset_id)

assert response == empty.Empty()

import os

import uuid

import pytest

import create_channel

import create_channel_event

import create_input

import delete_channel

import delete_channel_event

import delete_input

import get_channel_event

import list_channel_events

import start_channel

import stop_channel

project_name = os.environ["GOOGLE_CLOUD_PROJECT"]

location = "us-central1"

input_id = f"python-test-input-{uuid.uuid4()}"

channel_id = f"python-test-channel-{uuid.uuid4()}"

event_id = f"python-test-event-{uuid.uuid4()}"

output_bucket_name = f"python-test-bucket-{uuid.uuid4()}"

output_uri = f"gs://{output_bucket_name}/channel-test/"

def test_channel_event_operations(capsys: pytest.fixture) -> None:

# Set up

event_name_project_id = f"projects/{project_name}/locations/{location}/channels/{channel_id}/events/{event_id}"

create_input.create_input(project_name, location, input_id)

create_channel.create_channel(

project_name, location, channel_id, input_id, output_uri

)

start_channel.start_channel(project_name, location, channel_id)

# Tests

response = create_channel_event.create_channel_event(

project_name, location, channel_id, event_id

)

assert event_name_project_id in response.name

response = get_channel_event.get_channel_event(

project_name, location, channel_id, event_id

)

assert event_name_project_id in response.name

list_channel_events.list_channel_events(project_name, location, channel_id)

out, _ = capsys.readouterr()

assert event_name_project_id in out

response = delete_channel_event.delete_channel_event(

project_name, location, channel_id, event_id

)

assert response is None

# Clean up

stop_channel.stop_channel(project_name, location, channel_id)

delete_channel.delete_channel(project_name, location, channel_id)

delete_input.delete_input(project_name, location, input_id)

import os

import uuid

from google.api_core.exceptions import FailedPrecondition, NotFound

from google.protobuf import empty_pb2 as empty

import pytest

import create_channel

import create_channel_with_backup_input

import create_input

import delete_channel

import delete_channel_event

import delete_input

import get_channel

import list_channel_events

import list_channels

import start_channel

import stop_channel

import update_channel

import utils

project_name = os.environ["GOOGLE_CLOUD_PROJECT"]

location = "us-central1"

input_id = f"python-test-input-{uuid.uuid4()}"

updated_input_id = f"python-test-up-input-{uuid.uuid4()}"

channel_id = f"python-test-channel-{uuid.uuid4()}"

output_bucket_name = f"python-test-bucket-{uuid.uuid4()}"

output_uri = f"gs://{output_bucket_name}/channel-test/"

def test_channel_operations(capsys: pytest.fixture) -> None:

# Clean up old resources in the test project

channel_responses = list_channels.list_channels(project_name, location)

for response in channel_responses:

next_channel_id = response.name.rsplit("/", 1)[-1]

input_attachments = response.input_attachments

if utils.is_resource_stale(response.create_time):

try:

event_responses = list_channel_events.list_channel_events(

project_name, location, next_channel_id

)

for response in event_responses:

next_event_id = response.name.rsplit("/", 1)[-1]

try:

delete_channel_event.delete_channel_event(

project_name, location, next_channel_id, next_event_id

)

except NotFound as e:

print(f"Ignoring NotFound, details: {e}")

try:

stop_channel.stop_channel(project_name, location, next_channel_id)

except FailedPrecondition as e:

print(f"Ignoring FailedPrecondition, details: {e}")

try:

delete_channel.delete_channel(

project_name, location, next_channel_id

)

except FailedPrecondition as e:

print(f"Ignoring FailedPrecondition, try to stop channel: {e}")

try:

stop_channel.stop_channel(

project_name, location, next_channel_id

)

except FailedPrecondition as e:

print(f"Ignoring FailedPrecondition, details: {e}")

except NotFound as e:

print(f"Ignoring NotFound, details: {e}")

except NotFound as e:

print(f"Ignoring NotFound, details: {e}")

for input_attachment in input_attachments:

next_input_id = input_attachment.input.rsplit("/", 1)[-1]

try:

delete_input.delete_input(project_name, location, next_input_id)

except NotFound as e:

print(f"Ignoring NotFound, details: {e}")

# Set up

channel_name_project_id = (

f"projects/{project_name}/locations/{location}/channels/{channel_id}"

)

create_input.create_input(project_name, location, input_id)

create_input.create_input(project_name, location, updated_input_id)

# Tests

response = create_channel.create_channel(

project_name, location, channel_id, input_id, output_uri

)

assert channel_name_project_id in response.name

list_channels.list_channels(project_name, location)

out, _ = capsys.readouterr()

assert channel_name_project_id in out

response = update_channel.update_channel(

project_name, location, channel_id, updated_input_id

)

assert channel_name_project_id in response.name

for input_attachment in response.input_attachments:

assert "updated-input" in input_attachment.key

response = get_channel.get_channel(project_name, location, channel_id)

assert channel_name_project_id in response.name

start_channel.start_channel(project_name, location, channel_id)

out, _ = capsys.readouterr()

assert "Started channel" in out

stop_channel.stop_channel(project_name, location, channel_id)

out, _ = capsys.readouterr()

assert "Stopped channel" in out

response = delete_channel.delete_channel(project_name, location, channel_id)

assert response == empty.Empty()

response = create_channel_with_backup_input.create_channel_with_backup_input(

project_name, location, channel_id, input_id, updated_input_id, output_uri

)

assert channel_name_project_id in response.name

# Clean up

delete_channel.delete_channel(project_name, location, channel_id)

delete_input.delete_input(project_name, location, input_id)

delete_input.delete_input(project_name, location, updated_input_id)