Stegawave Python SDK

Why this SDK

The stegawave package wraps the public Stegawave REST API with typed models, ergonomic helpers, and production-ready retry logic. Use it to provision pipelines, monitor readiness, manage viewer tokens, and automate decode workflows from one cohesive client.

Typed payloads

Create and validate requests with pydantic models such as models.CreatePipelineRequest, eliminating brittle hand-written dictionaries.

Workflow helpers

PipelineSession waits for provisioning, normalizes ingest/manifests, and signs CDN URLs so you can go from event ID to playback links quickly.

Full API coverage

Endpoints for /create, /get, /state, /delete, /token, /decode, /iptv, and passphrase rotation map to dedicated methods.

Purpose-built errors

Network hiccups, auth failures, validation issues, and rate limits surface as descriptive subclasses of StegawaveError.

Installation & Setup

Install from PyPI and either pass credentials explicitly or rely on environment variables. The client enforces an API key at construction time.

pip install stegawave

Variable	Purpose	Default
`STEGAWAVE_API_KEY`	Primary authentication token	None (required)

from stegawave import StegawaveClient

# Explicit credentials
client = StegawaveClient(api_key="your-api-key", timeout=20.0, retry_attempts=5)

# Or rely on environment variables
with StegawaveClient() as client:
    print("Client ready")

Quick start pipeline

Provision a pipeline that mirrors the UI defaults, wait for readiness, and collect ingest plus signed playback URLs.

from stegawave import StegawaveClient, models

request = models.CreatePipelineRequest(
    name="launch-stream",
    description="Product launch livestream",
    segmentDuration=4,
    input=models.InputConfig(Type="RTMP_PUSH", whitelist=["0.0.0.0/0"]),
    encoder=models.EncoderConfig(
        vodArchive=False,
        output_group=models.OutputGroup(
            Outputs=[
                models.OutputConfig(
                    OutputName="cmaf-1080p",
                    resolution="1920x1080",
                    FramerateNumerator=30,
                    FramerateDenominator=1,
                    VideoBitrate="7.5M",
                ),
                models.OutputConfig(
                    OutputName="cmaf-720p",
                    resolution="1280x720",
                    FramerateNumerator=30,
                    FramerateDenominator=1,
                    VideoBitrate="5.2M",
                ),
            ]
        ),
    ),
    packager=models.PackagerConfig(
        originEndpoints=[
            models.OriginEndpoint(
                name="cmaf-hybrid",
                ContainerType="CMAF",
                HlsManifests=[models.HlsManifest(ManifestName="index")],
                DashManifests=[models.DashManifest(ManifestName="index")],
                StartoverWindowSeconds=86_400,
            )
        ]
    ),
)

with StegawaveClient() as client:
    session = client.create_pipeline_session(request, wait=True)
    print("Input URI:", session.input_uri)
    for playback in session.signed_manifest_uris("viewer-key"):
        print("Manifest:", playback)

PipelineSession tour

The PipelineSession wrapper keeps the latest status cached, handles polling for readiness, normalizes ingest details, and signs CDN manifests for a viewer token.

Monitor readiness

session = client.get_pipeline_session(event_id)
status = session.wait_until_ready(timeout=600, poll_interval=5)
print(status.status, status.createdAt)

Normalize ingest

input_details = session.get_input()
print(input_details.protocol)
print(input_details.uri)
print(input_details.allowed_ips)

Deliver playback

for url in session.signed_manifest_uris("viewer123", exp_hours=4):
    print("Signed:", url)

Model building patterns

Models enforce structure with helpful validation. Compose them incrementally to build tailored pipelines.

SRT listener input

models.InputConfig(
    Type="SRT_LISTENER",
    whitelist=["203.0.113.50/32"],
    SrtListenerSettings=models.SrtListenerSettings(
        IngestPort=9000,
        MinLatency=2000,
        PassphraseEnabled=True,
    ),
)

ABR ladder

models.OutputGroup(
    Outputs=[
        models.OutputConfig(
            resolution="1920x1080",
            FramerateNumerator=60,
            FramerateDenominator=1,
            VideoBitrate="8M",
        ),
        models.OutputConfig(
            resolution="1280x720",
            FramerateNumerator=60,
            FramerateDenominator=1,
            VideoBitrate="4M",
        ),
    ],
)

CMAF + HLS/DASH

models.OriginEndpoint(
    name="cmaf-hybrid",
    ContainerType="CMAF",
    HlsManifests=[models.HlsManifest(ManifestWindowSeconds=600)],
    DashManifests=[models.DashManifest(ManifestWindowSeconds=600)],
    StartoverWindowSeconds=86_400,
)

Lifecycle operations

Control an existing pipeline by event ID. Responses are typed, so you can inspect structured fields without manual parsing.

from stegawave import StegawaveClient

EVENT_ID = "example-event-id-123456"

with StegawaveClient() as client:
    start = client.start_pipeline(EVENT_ID)
    print("Start state:", start.state)

    ready = client.wait_for_event(EVENT_ID, timeout=120, poll_interval=5)
    print("Pipeline status:", ready.status)

    stop = client.stop_pipeline(EVENT_ID)
    print("Stop state:", stop.state)

    delete = client.delete_pipeline(EVENT_ID)
    print("Delete status:", delete.status)
    if delete.note:
        print("Note:", delete.note)

Viewer tokens & playback

Generate viewer-specific tokens and stitch them into signed CDN URLs. Tokens arrive as a dictionary keyed by the requested user IDs.

with StegawaveClient() as client:
    tokens = client.fetch_token(["alice", "bob"], exp_hours=2)
    session = client.get_pipeline_session("event-id")

    print(tokens.tokens["alice"])
    for manifest in session.signed_manifest_uris("alice", parameter="token"):
        print(manifest)

Decode & IPTV automation

Tie IPTV discovery into watermark decode jobs. The SDK normalizes IPTV results and decode responses so you can audit the full flow.

with StegawaveClient() as client:
    query = models.IptvQueryRequest(
        server="https://iptv.example.com",
        username="ops",
        password="secret",
        channelName="news-channel",
        format="m3u8",
        preferHD=True,
    )
    catalog = client.iptv_query(query)
    stream = catalog.results[0]

    decode_request = models.DecodeJobRequest(
        eventID="event-id",
        input_stream=stream.stream_urls[0],
    )
    decode_response = client.submit_decode_job(decode_request)
    print("Decode message:", decode_response.message)
    if decode_response.clientID:
        print("Client ID:", decode_response.clientID)

Rotate passphrases

For SRT listeners with passphrase enforcement, retrieve the current secret or rotate it (optionally supplying your own 32-character string).

with StegawaveClient() as client:
    current = client.get_passphrase()
    print("Current passphrase:", current.passphrase)

    rotated = client.rotate_passphrase()
    print("New passphrase:", rotated.newPassphrase)

    custom = client.rotate_passphrase(
        models.RotatePassphraseRequest(passphrase="0123456789abcdef0123456789abcdef")
    )
    print("Custom rotation successful:", custom.success)

Error handling

Every failure path produces a descriptive exception with optional payload data.

AuthenticationError – Missing or invalid API key.
AuthorizationError – API key lacks permission for the target resource.
ValidationError – The request payload failed schema checks.
ProvisioningError – Pipeline provisioning failed or timed out.
RateLimitError – Slow down after 429 responses.
NetworkError – Retries exhausted while calling the API.
ServerError – Upstream 5xx responses.
UnexpectedResponseError – Payload shape did not match the models.

Example scripts

Copy, adapt, and run the reference scripts that live alongside this documentation.

Provision a pipeline, await readiness, and emit ingest plus signed manifests.

Loading example…

"""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,
    input=models.InputConfig(
      Type="RTMP_PUSH",
      whitelist=["0.0.0.0/0"],
    ),
    encoder=models.EncoderConfig(
      vodArchive=False,
      output_group=models.OutputGroup(
        Name="cmaf-main",
        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 sends a start command for an existing Stegawave pipeline."""

import sys
from typing import Optional

from stegawave import StegawaveClient


def main():
    
  EVENT_ID = "example-event-id-123456"
  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.endpoint:
        print(f"  Input endpoint: {status.input.endpoint}")
      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 typing import Optional

from stegawave import StegawaveClient

EVENT_ID="example-event-id-123456"

def main():
  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 typing import Optional
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 typing import Optional

from stegawave import StegawaveClient, models


def main():
  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=args.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())

Need deeper coverage? The PyPI package ships unit tests under pypi/tests that exercise client methods and workflow helpers.