All Products
Search
Document Center

Object Storage Service:Manage LiveChannels by using OSS SDK for Python

Last Updated:Jan 05, 2024

This topic describes the common operations that you can perform on LiveChannels by using Object Storage Service (OSS) SDK for Python. For example, you can create, list, and delete LiveChannels.

Environment preparation

  • Python 3.6

    Note

    The examples in this topic are provided based on Python 3.6. The examples are also applicable to Python 2.6, 2.7, 3.3, 3.4, and 3.5.

  • aliyun-oss-python-sdk 2.9.0

  • OBS Studio

  • IDE

Create a LiveChannel

Before you can upload audio and video data by using Real-Time Messaging Protocol (RTMP), you must call the PutLiveChannel operation to create a LiveChannel. The response to the PutLiveChannel request includes the URL that is used to ingest streams to the LiveChannel and the URL that is used to play the ingested streams.

Note

You can use the returned URLs to ingest and play streams. You can also perform operations based on the returned LiveChannel name, such as querying stream ingest status, querying stream ingest records, and disabling stream ingest.

The following sample code provides an example on how to create a LiveChannel:

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# Create a bucket. 
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# Create and configure a LiveChannel. 
# Set the name of the channel to test_rtmp_live. Set the m3u8 file generated by live streaming to test.m3u8. The index file includes three TS files, and the duration of each TS file is 5 seconds. The value of 5 seconds in this example is the recommended value. The actual duration varies based on the key frame of the video file. 
channel_name = "test_rtmp_live"
playlist_name = "test.m3u8"

create_result = bucket.create_live_channel(
        channel_name,
        oss2.models.LiveChannelInfo(
            status = 'enabled',
            description = 'LiveChannel used for the test',
            target = oss2.models.LiveChannelInfoTarget(
                playlist_name = playlist_name,
                frag_count = 3,
                frag_duration = 5)))

List and delete LiveChannels

The following sample code provides an example on how to list and delete LiveChannels:

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# Create a bucket. 
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# List the LiveChannels that meet the requirements. 
# List all LiveChannels that meet the requirements in the bucket. 
# param: prefix (type: str) specifies the prefix contained in the name of the LiveChannels that you want to list. If you do not specify this parameter, all LiveChannels are listed. 
# return: class:`ListLiveChannelResult <oss2.models.ListLiveChannelResult>`
for info in oss2.LiveChannelIterator(bucket, prefix="test"):
    print(info.name)

# Delete a LiveChannel. 
bucket.delete_live_channel(info.name)

Configure the status of a LiveChannel

The following sample code provides an example on how to configure the status of a LiveChannel:

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# Create a bucket. 
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# Enable or disable a LiveChannel. 
bucket.put_live_channel_status(channel_name, 'enabled')
bucket.put_live_channel_status(channel_name, 'disabled')

Query RTMP ingest URLs and signatures

The following sample code provides an example on how to query RTMP ingest URLs and signatures:

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# Create a bucket. 
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# Query the ingest URL and streaming URL. 
# After you create a LiveChannel, query play_url that is used to ingest the stream and publish_url that is used to watch the stream. play_url is the RTMP ingest URL. If the access control list (ACL) of your bucket is not public-read-write, you must include a signature in the URL. The following code provides an example. publish_url is the URL of the M3U8 file that is generated by stream ingest. 
# Query create_result by using create_live_channel(). 
publish_url = create_result.publish_url
play_url = create_result.play_url
print("Ingest URL:",publish_url)
print("Streaming URL:",play_url)

# After you query the ingest URL and streaming URL, you can ingest the stream to OSS and watch the stream. If the bucket ACL is not public-read-write, you need to sign the URL. If the bucket ACL is public-read-write, you can use publish_url to ingest the stream. 
# The value of expires in the example indicates the period of time in seconds before the expiration time of the stream that you want to ingest. 
# All parameters are involved in the signature. 
# After you query signed_url, you can use stream ingest tools to ingest streams to OSS. OSS checks the value of expires only when a stream is connected to OSS. A stream that is connected to OSS is not interrupted even if the duration of the stream exceeds the value specified by expires. 
signed_url = bucket.sign_rtmp_url(channel_name, playlist_name, expires=3600)
print(signed_url)

Query the stream ingest status of a LiveChannel

The following sample code provides an example on how to query the stream ingest status of a LiveChannel:

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# Create a bucket. 
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# Query the status of the current stream. 
get_status = bucket.get_live_channel_stat(channel_name)
print("Connection time:",get_status.connected_time)
print("IP address of the ingestion client:",get_status.remote_addr )
print("Stream ingest status:",get_status.status )

Generate and view a playlist

You can call PostVodPlaylist to generate a VOD playlist for a specific LiveChannel. OSS queries the TS segments that are generated by the streams ingested to the specified LiveChannel within a specific period of time and converges the segments into an M3U8 playlist.

The following sample code provides an example on how to generate and view a playlist:

import os
import oss2
import time

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# Create a bucket.
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

end_time = int(time.time())
start_time = end_time - 3600
generate_playlist = "my_vod_list.m3u8"
# Generate a VOD playlist. 
# If you want to use the TS segments generated by stream ingest to generate a VOD playlist, you can use the post_vod_playlist method. 
# In this example, the start time is the current time minus 3,600 seconds, and the end time is the current time. This indicates that a playlist is generated for the stream of the previous hour. 
# After this operation is called, a playlist named my_vod_list.m3u8 is generated in OSS. 
bucket.post_vod_playlist(
                channel_name,
                playlist_name,
                start_time = start_time,
                end_time = end_time)

# If you want to view the playlist content within a specific period of time, you can call the get_vod_playlist operation. 
result = bucket.get_vod_playlist(channel_name, start_time=start_time, end_time=end_time)
print("playlist:", result.playlist)

Query the configurations of a LiveChannel

The following sample code provides an example on how to query the configurations of a LiveChannel:

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# Create a bucket. 
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# Query the configurations of a LiveChannel. 
get_result = bucket.get_live_channel(channel_name)
print("-------------------")
print("Configurations of stream ingest:")
print(get_result.description)
print(get_result.status)
print(get_result.target.type)
print(get_result.target.frag_count)
print(get_result.target.frag_duration)
print(get_result.target.playlist_name)
print("-------------------")

Query the stream ingest records of a LiveChannel

You can call the GetLiveChannelHistory operation to query up to 10 most recent stream ingest records of a LiveChannel. The following sample code provides an example on how to query the stream ingest records of a LiveChannel:

import os
import oss2

access_key_id = os.getenv('OSS_TEST_ACCESS_KEY_ID', '**')
access_key_secret = os.getenv('OSS_TEST_ACCESS_KEY_SECRET', '***')
bucket_name = os.getenv('OSS_TEST_BUCKET', '********')
endpoint = os.getenv('OSS_TEST_ENDPOINT', '***')

# Create a bucket. 
bucket = oss2.Bucket(oss2.Auth(access_key_id, access_key_secret), endpoint, bucket_name)

# Query the stream ingest records of a LiveChannel. 
history_result = bucket.get_live_channel_history(channel_name)
print("Stream ingest records:",len(history_result.records))

References

For more information about LiveChannels, see Overview, live_channel.py, and api.py.

FAQ

  • Why am I unable to query information, such as the stream ingest status, client IP address, and connection time?

    Before you call the get_live_channel_stat operation to query the stream ingest status, make sure that the corresponding LiveChannel (channel_name) is in the Live state. The Live state specifies that the client is ingesting streams to the connected ingest URL.

  • Can I call the get_live_channel_history operation to query the start time and end time and remote address of a historical stream ingest?

    Yes. For more information, see GetLiveChannelHistory.

  • What type of LiveChannel information is returned after you call the list_live_channel operation?

    A string. For more information, see GetLiveChannelHistory.

  • What is the format of the end_time parameter in the post_vod_playlist function that generates a VOD playlist?

    An integer. For more information, see PostVodPlaylist.

  • Why is the ['Code': 'InvalidArgument', 'Message': 'No ts file found in specified time span.'] error message returned?

    A VOD playlist can be generated only after the stream ingest files are uploaded.