stegawave 0.2.15
Python client for the Stegawave media pipeline API
Requires Python: >=3.9
Keywords: aws, media, stegawave, streaming, video

# 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

```bash
pip install stegawave
```

## Quick start

```python
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:**

```python
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:

```python
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:**

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

**Example with redundant sources:**

```python
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:

```python
# 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.

## DRM Configuration

Protect your content with SPEKE 2.0 encryption. Add the `drm` field to any `OriginEndpoint` to enable segment encryption.

**Supported encryption methods:**
- **CMAF**: CENC (Widevine, PlayReady, Irdeto) or CBCS (Widevine, PlayReady, FairPlay)
- **TS**: SAMPLE_AES (FairPlay) or AES_128 (Clear Key)

**Example with Widevine and PlayReady:**

```python
models.OriginEndpoint(
    name="cmaf-drm",
    ContainerType="CMAF",
    HlsManifests=[models.HlsManifest(ManifestName="index")],
    DashManifests=[models.DashManifest(ManifestName="index")],
    drm=models.DrmConfig(
        encryptionMethod="CENC",
        drmSystems=["WIDEVINE", "PLAYREADY"],
        resourceId="asset-12345",
        keyServer=models.DrmKeyServerConfig(
            url="https://kms.example.com/v2/cpix/getKey",
            headers={
                "Authorization": "Bearer your-token"
            },
            query={
                "site-id": "12345"
            }
        ),
        keyRotationIntervalSeconds=300  # Optional: rotate keys every 5 minutes
    )
)
```

**Example with FairPlay:**

```python
models.OriginEndpoint(
    name="hls-fairplay",
    ContainerType="TS",
    HlsManifests=[models.HlsManifest(ManifestName="index")],
    drm=models.DrmConfig(
        encryptionMethod="SAMPLE_AES",
        drmSystems=["FAIRPLAY"],
        resourceId="asset-67890",
        keyServer=models.DrmKeyServerConfig(
            url="https://fairplay.example.com/key-server"
        )
    )
)
```

## 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=...`)
- `schedule_action(event_id, action, scheduled_time)` - Schedule start/stop/delete at specific time (POST `/schedule-actions`)
- `cancel_scheduled_action(event_id, action_id)` - Cancel a scheduled action (POST `/cancel-scheduled-action`)
- `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`)

### Schedule Pipeline Actions

Schedule start, stop, or delete actions to execute at a specific time using EventBridge rules.

```python
# Schedule a stop action for 11 PM UTC
response = client.schedule_action(
    event_id="abc123",
    action="stop",
    scheduled_time="2025-12-20T23:00:00Z"
)
print(f"Scheduled with action ID: {response.actionID}")

# Cancel the scheduled action before it executes
cancel_response = client.cancel_scheduled_action(
    event_id="abc123",
    action_id=response.actionID
)
print(f"Cancelled {cancel_response.action} scheduled for {cancel_response.scheduledTime}")

# View scheduled actions in pipeline details
pipeline = client.get_pipeline("abc123")
for action in pipeline.schedule:
    print(f"{action['action']} scheduled for {action['scheduledTime']}")
```

### Reset Channel History

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

```python
# 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:

```python
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.

```python
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

```bash
pip install -e .[dev]
pytest
```

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