stegawave

Stegawave Python Client

stegawave is an unofficial Python SDK for the Stegawave forensic watermarking platform. It wraps the public REST API and helps you validate /create-pipeline payloads, manage pipeline lifecycle, and trigger watermark decode jobs without hand-writing HTTP calls.

Installation

pip install stegawave

Quick start

from stegawave import StegawaveClient, models

client = StegawaveClient(api_key="your-api-key")

create_request = models.CreatePipelineRequest(
    name="launch-stream",
    description="Product launch livestream",
    segmentDuration=4,
    input=models.InputConfig(
        Type="SRT_LISTENER",
        whitelist=["0.0.0.0/0"],
        SrtListenerSettings=models.SrtListenerSettings(
            IngestPort=5000,
            MinLatency=2000,
            PassphraseEnabled=True,
            # Passphrase will be auto-generated if not provided
        ),
    ),
    encoder=models.EncoderConfig(
        vodArchive=False,
        Outputs=[
            models.OutputConfig(
                OutputName="cmaf-1080p",
                resolution="1920x1080",
                FramerateNumerator=30,
                FramerateDenominator=1,
                VideoBitrate=7_500_000,
                AudioBitrate=128_000,
            )
        ],
    ),
    packager=models.PackagerConfig(
        originEndpoints=[
            models.OriginEndpoint(
                name="cmaf-hybrid",
                ContainerType="CMAF",
                HlsManifests=[models.HlsManifest(ManifestName="index")],
            )
        ]
    ),
)

session = client.create_pipeline_session(create_request, wait=True)
print(session.event_id)

# Access input endpoints and passphrase
status = session.status
if status.input.endpoints:
    print("Input endpoints:", status.input.endpoints)
if status.input.passphraseEnabled:
    print("Passphrase:", status.input.passphrase)

print("Manifests:")
for url in session.signed_manifest_uris("john_doe"):
    print("  ", url)

Input types

The SDK supports the following input types:

• SRT_LISTENER - SRT listener endpoint (recommended for most use cases)
• SRT_CALLER - SRT caller that connects to remote endpoints
• RTP - RTP/UDP input
• RTP_FEC - RTP with Forward Error Correction
• RIST - Reliable Internet Stream Transport
• ZIXI_PUSH - Zixi push input

Note: RTMP and file-based inputs (HLS, MP4, TS) have been deprecated in favor of the above streaming protocols.

SRT Listener (Recommended)

The SRT_LISTENER input type creates an SRT listener endpoint where you can push your stream. This is the most common input type.

Requirements:

• SrtListenerSettings with IngestPort
• Exactly one CIDR in whitelist array
• Optional passphrase encryption

Example:

models.InputConfig(
    Type="SRT_LISTENER",
    whitelist=["0.0.0.0/0"],
    SrtListenerSettings=models.SrtListenerSettings(
        IngestPort=5000,
        MinLatency=2000,
        MaxLatency=10000,
        PassphraseEnabled=True,
        Passphrase="my-32-character-passphrase!!!!"  # Optional - auto-generated if omitted
    )
)

After creation, retrieve the generated passphrase from the pipeline status:

status = client.get_pipeline(event_id)
if status.input.passphraseEnabled:
    print(f"Generated passphrase: {status.input.passphrase}")

SRT Caller

The SRT_CALLER input type enables MediaLive to initiate outbound SRT connections to remote SRT listener endpoints. This is useful for connecting to external encoders or CDN origins that expose SRT listener ports.

Requirements:

• Provide 1 or 2 SrtCallerSources (for redundancy)
• Each source requires SrtListenerAddress (IP or hostname) and SrtListenerPort
• MediaLive channel class is automatically selected: 1 source → SINGLE_PIPELINE, 2 sources → STANDARD

Example with single source:

models.InputConfig(
    Type="SRT_CALLER",
    SrtCallerSources=[
        models.SrtCallerSource(
            SrtListenerAddress="encoder.example.com",
            SrtListenerPort=9000,
            StreamId="primary-feed"  # Optional
        )
    ]
)

Example with redundant sources:

models.InputConfig(
    Type="SRT_CALLER",
    SrtCallerSources=[
        models.SrtCallerSource(
            SrtListenerAddress="encoder1.example.com",
            SrtListenerPort=9000,
            SrtCallerDecryption=models.SrtCallerDecryption(
                Algorithm="AES256",
                Passphrase="16-char-minimum!"
            )
        ),
        models.SrtCallerSource(
            SrtListenerAddress="encoder2.example.com",
            SrtListenerPort=9000,
            SrtCallerDecryption=models.SrtCallerDecryption(
                Algorithm="AES256",
                Passphrase="16-char-minimum!"
            )
        )
    ]
)

Notes:

• No whitelist needed (MediaLive initiates outbound connections)
• Optional StreamId for stream routing at the remote endpoint
• Optional SrtCallerDecryption for encrypted streams
• Passphrase must match the remote listener's encryption passphrase (16-64 characters)

RTP, RIST, and ZIXI

For other professional streaming protocols:

# RTP input
models.InputConfig(Type="RTP")

# RTP with FEC
models.InputConfig(Type="RTP_FEC")

# RIST input
models.InputConfig(Type="RIST")

# Zixi push
models.InputConfig(Type="ZIXI_PUSH")

Refer to the API documentation for protocol-specific configuration options.

Features

• Strongly-typed request and response models for /create-pipeline, /get-pipeline, /pipeline-state, /delete, /token, /decode, /iptv
• High-level PipelineSession workflow helper to provision, poll, and sign manifests in a few lines
• Support for modern streaming protocols: SRT, RTP, RIST, ZIXI
• Automatic passphrase generation for SRT listener inputs
• Convenience helpers for ABR ladders and asynchronous provisioning workflows
• Configurable retries, timeouts, and polling intervals
• First-class error types for authentication, validation, rate limiting, and server-side failures

API Endpoints

The client provides methods for all major API endpoints:

• create_pipeline(request) - Create a new pipeline (POST /create-pipeline)
• get_pipeline(event_id) - Get pipeline details (GET /get-pipeline?eventID=...)
• list_pipelines() - List all pipelines (GET /get-pipeline)
• get_state(event_id) - Get pipeline state (POST /pipeline-state with action=status)
• start_pipeline(event_id) - Start a pipeline (POST /pipeline-state with action=start)
• stop_pipeline(event_id) - Stop a pipeline (POST /pipeline-state with action=stop)
• delete_pipeline(event_id) - Delete a pipeline (DELETE /delete?eventID=...)
• reset_history(event_id) - Reset channel history/DVR window (GET /reset-history?eventID=...)
• fetch_token(user_key, exp_hours) - Generate CDN tokens (POST /token)
• decode_stream(event_id, stream_url) - Trigger watermark decode job (POST /decode)
• query_iptv(...) - Search IPTV streams (POST /iptv)
• get_vod_archive(event_id, expires_in) - Get VOD archive files with presigned URLs (GET /get-vod-archive)

Reset Channel History

Clear the DVR/startover window for a pipeline while preserving its configuration. The pipeline must be stopped before resetting history.

# Stop the pipeline first
client.stop_pipeline(event_id)

# Wait a moment for stop to complete
import time
time.sleep(5)

# Reset the history
response = client.reset_history(event_id)
print(response.message)  # "Channel history reset initiated for evt-123"

# Restart the pipeline if needed
client.start_pipeline(event_id)

Note: The reset operation returns 202 Accepted and processes asynchronously. It typically completes within a few seconds.

Watermark Detection Results

When watermarks are detected in a pipeline stream, the results are available via the detected_users field:

status = client.get_pipeline(event_id)

if status.detected_users:
    print(f"Found {len(status.detected_users)} detected watermarks:")
    for detection in status.detected_users:
        # Each detection is a DetectedUser object with:
        # - user: Human-readable user identifier
        # - user_key: Unique watermark key
        # - similarity: Confidence score (0-1)
        # - detected_at: ISO timestamp of detection
        print(f"  User: {detection.user}")
        print(f"  Key: {detection.user_key}")
        print(f"  Similarity: {detection.similarity}")
        print(f"  Detected at: {detection.detected_at}")

Migration from 0.1.x

If upgrading from version 0.1.x, please note these breaking changes:

Removed endpoints:

• get_passphrase() and rotate_passphrase() - Passphrase management is now integrated into pipeline creation via SrtListenerSettings

Removed input types:

• RTMP_PUSH, RTMP_PULL - Use SRT_LISTENER instead
• HLS, MP4_FILE, TS_FILE - Use streaming protocols (SRT, RTP, RIST, ZIXI)

Response format changes:

• Input endpoints are now an array: status.input.endpoints (was status.input.endpoint)
• CDN endpoints include protocol info: CdnEndpoint objects with protocol and url fields
• Manifests include type info: ManifestInfo objects with type and name fields

See CHANGELOG.md for complete migration guide.

See CHANGELOG.md for complete migration guide.

Configuration

Set your base URL or API key explicitly, or rely on environment variables.

client = StegawaveClient()

Environment variable	Description
STEGAWAVE_API_KEY	API key provided by Stegawave
STEGAWAVE_API_BASE_URL	Override the default https://api.stegawave.com

The SDK automatically injects your API key, validates payload structure using Pydantic models, and surfaces HTTP issues as rich exceptions.

Status

This client is v0.2.0 targeting the November 2025 API schema. Version 0.2.0 introduces breaking changes - see migration guide above. Contributions and issue reports are welcome.

Development

pip install -e .[dev]
pytest

Refer to CHANGELOG.md for planned enhancements and release history.