Stegawave Python SDK

"""Example script that provisions a Stegawave pipeline and 
   prints ingest + playback details.
"""

import sys
from typing import Iterable

from stegawave import StegawaveClient, models


def build_pipeline_request() -> models.CreatePipelineRequest:
  """Return a sample pipeline configuration mirroring the UI defaults."""
  return models.CreatePipelineRequest(
    name="launch-stream",
    description="Product launch livestream",
    segmentDuration=4,
    autoStart=True,
    input=models.InputConfig(
      Type="SRT_LISTENER",
      whitelist=["0.0.0.0/0"],
      SrtListenerSettings=models.SrtListenerSettings(
        IngestPort=9000,
        MinLatency=2000,
        PassphraseEnabled=True,
      ),
    ),
    encoder=models.EncoderConfig(
      vodArchive=False,
      Outputs=[
        models.OutputConfig(
          OutputName="cmaf-1080p",
          resolution="1920x1080",
          FramerateNumerator=30,
          FramerateDenominator=1,
          VideoBitrate=7_500_000,
          AudioBitrate=128_000,
          SampleRate=48_000,
        ),
        models.OutputConfig(
          OutputName="cmaf-720p",
          resolution="1280x720",
          FramerateNumerator=30,
          FramerateDenominator=1,
          VideoBitrate=5_200_000,
          AudioBitrate=96_000,
          SampleRate=48_000,
        ),
      ],
    ),
    packager=models.PackagerConfig(
      originEndpoints=[
        models.OriginEndpoint(
          name="cmaf-hybrid",
          ContainerType="CMAF",
          description="Public CMAF endpoint",
          HlsManifests=[models.HlsManifest(ManifestName="index")],
          DashManifests=[models.DashManifest(ManifestName="index")],
          StartoverWindowSeconds=86_400,
        ),
        models.OriginEndpoint(
          name="cmaf-live",
          ContainerType="TS",
          description="Live TS endpoint",
          HlsManifests=[models.HlsManifest(ManifestName="index")],
          StartoverWindowSeconds=86_400,
        )
      ]
    ),
  )


def main():

  with StegawaveClient() as client:
    request = build_pipeline_request()
    response = client.create_pipeline(request)
    if not response.valid():
      print("Pipeline creation request was rejected", file=sys.stderr)
      return 1

    event_id = response.eventID
    print(f"Pipeline provisioning started. eventID={event_id}")

    client.wait_for_event(event_id, poll_interval=5.0)
    session = client.get_pipeline_session(event_id)
    status = session.status
    input_details = session.get_input()
    manifests = session.get_manifests()
    manifest_urls: Iterable[str] = session.signed_manifest_uris("john_doe")

    print(f"Session details for eventID={event_id}:")
    print(f"  Status: {status.status}")
    print(f"  Created: {status.createdAt}")
    print(f"  Last updated: {status.lastUpdated}")
    print(f"  Input protocol: {input_details.protocol}")
    print(f"  Input URIs: {', '.join(input_details.uris) if input_details.uris else 'Unavailable'}")
    if manifests:
      print("  Manifests:")
      for manifest in manifests:
        print(f"    - {manifest.uri}")
    else:
      print("  Manifests: Unavailable")

    print("\nInput endpoint:")
    print(f"  {input_details.uri or 'Unavailable'}")

    print("\nPlayback manifests:")
    if manifest_urls:
      for url in manifest_urls:
        print(f"  {url}")
    else:
      print("  No CDN endpoints reported yet")

  return 0


if __name__ == "__main__":
  raise SystemExit(main())

"""Example script that demonstrates the /get-pipeline endpoint.

Run with no arguments to list every event:
  python get_pipelines.py

Pass an event ID to fetch a single event:
  python get_pipelines.py example-event-id
"""

import sys

from stegawave import StegawaveClient


def main():
  event_id = sys.argv[1] if len(sys.argv) > 1 else None

  with StegawaveClient() as client:
    if not event_id:
      # No CLI argument means hit /get-pipeline without an eventID query string to list everything.
      listing = client.list_pipelines()
      pipelines = listing.pipelines or []
      if not pipelines:
        print("No events returned from /get-pipeline.")
        return 0

      print(f"Retrieved {len(pipelines)} event(s) from /get-pipeline:")
      for pipeline in pipelines:
        descriptor = pipeline.eventID or pipeline.name or "unknown"
        status = pipeline.status or "unknown"
        created = pipeline.createdAt or "unknown"
        print(f"- {descriptor} (status={status}, createdAt={created})")
      return 0

    # When an event ID is provided we include it as the query string to narrow the response.
    status = client.get_pipeline(event_id)
    print(f"Details for eventID={status.eventID}:")
    print(f"  Status: {status.status}")
    if status.input and status.input.endpoints:
      print(f"  Input endpoints: {', '.join(status.input.endpoints)}")
    if status.input and status.input.passphrase:
      print(f"  Passphrase: {status.input.passphrase}")
    if status.cdn and status.cdn.endpoints:
      print("  CDN endpoints:")
      for uri in status.cdn.endpoints:
        print(f"    - {uri}")
    if status.detected_users:
      print(f"  Detected users: {status.detected_users}")
    if status.lastUpdated:
      print(f"  Last updated: {status.lastUpdated}")

  return 0


if __name__ == "__main__":
  raise SystemExit(main())

"""Example script that sends a start command for an existing Stegawave pipeline."""

import sys

from stegawave import StegawaveClient


def main():
  if len(sys.argv) < 2:
    print("Usage: python start_pipeline.py ", file=sys.stderr)
    return 1

  event_id = sys.argv[1]
  WAIT_READY = True
  TIMEOUT = 60.0
  POLL_INTERVAL = 5.0

  with StegawaveClient() as client:
    response = client.start_pipeline(event_id)
    print(f"Start requested for eventID={response.eventID} (state={response.state or 'unknown'})")

    if WAIT_READY:
      status = client.wait_for_event(
        event_id,
        timeout=TIMEOUT,
        poll_interval=POLL_INTERVAL,
      )
      print("Pipeline is ready:")
      print(f"  Status: {status.status}")
      if status.input and status.input.endpoints:
        print(f"  Input endpoints: {', '.join(status.input.endpoints)}")
      if status.cdn and status.cdn.endpoints:
        print("  CDN endpoints:")
        for uri in status.cdn.endpoints:
          print(f"    - {uri}")

  return 0


if __name__ == "__main__":
  raise SystemExit(main())

"""Example script that deletes a Stegawave pipeline by eventID."""

import sys

from stegawave import StegawaveClient


def main():
  if len(sys.argv) < 2:
    print("Usage: python delete_pipeline.py ", file=sys.stderr)
    return 1

  event_id = sys.argv[1]

  with StegawaveClient() as client:
    response = client.delete_pipeline(event_id)
    print(f"Delete requested for eventID={response.eventID} (status={response.status})")
    if response.note:
      print(f"Note: {response.note}")

  return 0

if __name__ == "__main__":
  raise SystemExit(main())

"""Example script that fetches viewer playback tokens for one or more user keys."""


import sys
from stegawave import StegawaveClient


def main():
  user_keys = ["user1", "user2", "user3"]
  with StegawaveClient() as client:
    response = client.fetch_token(user_keys, exp_hours=0)
    for user_key, token in response.tokens.items():
      print(f"{user_key}: {token}")

  return 0


if __name__ == "__main__":
  raise SystemExit(main())

"""Example script that wires IPTV catalog results into a decode job."""

import sys

from stegawave import StegawaveClient, models


def main():
  if len(sys.argv) < 2:
    print("Usage: python iptv_decode.py ", file=sys.stderr)
    return 1

  event_id = sys.argv[1]

  with StegawaveClient() as client:
    query = models.IptvQueryRequest(
      server='http://iptv.example.com',
      username='user123',
      password='pass123',
      channelName='example_channel',
      categoryId=None,
      format='m3u8',
      preferHD=False,
      preferUK=False,
      avoidVIP=False,
    )
    iptv_response = client.iptv_query(query)
    if not iptv_response.results:
      print("No IPTV streams returned for the supplied parameters", file=sys.stderr)
      return 1

    first_stream = iptv_response.results[0]
    if not first_stream.stream_urls:
      print("IPTV result did not include any stream URLs", file=sys.stderr)
      return 1

    selected_url = first_stream.stream_urls[0]
    decode_request = models.DecodeJobRequest(eventID=event_id, input_stream=selected_url)
    decode_response = client.submit_decode_job(decode_request)

    # Report the end-to-end selection so operators can verify routing.
    print(f"Submitted decode job for eventID={decode_response.eventID}")
    print(f"  IPTV stream: {first_stream.name} (stream_id={first_stream.stream_id})")
    print(f"  Input URL: {selected_url}")
    if decode_response.clientID:
      print(f"  Client ID: {decode_response.clientID}")
    print(f"  Message: {decode_response.message}")


  return 0


if __name__ == "__main__":
  raise SystemExit(main())

"""Example script that submits a decode job for a stream URL.

Usage:
  python decode.py  
"""

import sys

from stegawave import StegawaveClient, models


def main():
  if len(sys.argv) < 3:
    print("Usage: python decode.py <event_id> <stream_url>", file=sys.stderr)
    return 1

  event_id = sys.argv[1]
  stream_url = sys.argv[2]

  with StegawaveClient() as client:
    decode_request = models.DecodeJobRequest(
      eventID=event_id,
      input_stream=stream_url,
    )
    
    decode_response = client.submit_decode_job(decode_request)
    
    print(f"Decode job submitted successfully")
    print(f"  Event ID: {decode_response.eventID}")
    print(f"  Stream URL: {stream_url}")
    if decode_response.clientID:
      print(f"  Client ID: {decode_response.clientID}")
    print(f"  Message: {decode_response.message}")

  return 0


if __name__ == "__main__":
  raise SystemExit(main())

"""Example script that lists and downloads VOD archive files from a pipeline.

Usage:
  # List archive files only
  python vod_archive.py <event_id>
  
  # List and download all files
  python vod_archive.py <event_id> --download ./archive
"""

import sys

from stegawave import StegawaveClient


def main():
  if len(sys.argv) < 2:
    print("Usage: python vod_archive.py <event_id> [--download <path>]", file=sys.stderr)
    return 1

  event_id = sys.argv[1]
  download = False
  download_path = None
  
  if len(sys.argv) >= 4 and sys.argv[2] == "--download":
    download = True
    download_path = sys.argv[3]

  with StegawaveClient() as client:
    if download:
      # Download all files to the specified directory
      print(f"Downloading VOD archive for eventID={event_id}...")
      response = client.get_vod_archive(
        event_id=event_id,
        download=True,
        download_path=download_path
      )
      print(f"\nDownloaded {response.archive.fileCount} files ({response.archive.totalSize})")
      print(f"Location: {download_path}")
    else:
      # Just list the files with their metadata
      response = client.get_vod_archive(event_id=event_id, expires_in=3600)
      
      print(f"VOD Archive for: {response.eventName} (eventID={response.eventID})")
      print(f"Total files: {response.archive.fileCount}")
      print(f"Total size: {response.archive.totalSize}")
      print(f"URLs expire at: {response.archive.expiresAt}")
      print(f"\nFiles:")
      
      for file in response.files:
        print(f"  [{file.index}] {file.filename}")
        print(f"      Size: {file.size}")
        print(f"      Modified: {file.lastModified}")
        print(f"      URL: {file.url[:80]}...")
        print()

  return 0


if __name__ == "__main__":
  raise SystemExit(main())