> For the complete documentation index, see [llms.txt](https://docs.degirum.com/llms.txt). Markdown versions of documentation pages are available by appending `.md` to page URLs; this page is available as [Markdown](https://docs.degirum.com/degirum-tools/streams.md).

# Streams

DeGirum Tools API Reference Guide. Streaming toolkit for building multi-threaded pipelines.

{% hint style="info" %}
This API Reference is based on DeGirum Tools version 1.2.0.
{% endhint %}

## Streaming Toolkit Overview <a href="#streaming-toolkit-overview" id="streaming-toolkit-overview"></a>

This module provides a streaming toolkit for building multi-threaded processing pipelines, where data (images, video frames, or arbitrary objects) flows through a series of *processing blocks* called gizmos. The toolkit allows you to:

* Acquire or generate data from one or more sources (e.g., camera feeds, video files).
* Process the data in a pipeline (possibly in parallel), chaining multiple gizmos together.
* Optionally display or save the processed data, or feed it into AI inference models.
* Orchestrate everything in a [Composition](/degirum-tools/streams/streams_base.md#composition), which manages the lifecycle (threads) of all connected gizmos.

### Core Concepts <a href="#core-concepts" id="core-concepts"></a>

1. **Stream**:
   * Represents a queue of data items [StreamData](/degirum-tools/streams/streams_base.md#streamdata), such as frames from a camera or images from a directory.
   * Gizmos push (`put`) data into Streams or read (`get`) data from them.
   * Streams can optionally drop data (the oldest item) if they reach a specified maximum queue size, preventing pipeline bottlenecks.
2. **Gizmo**:
   * A gizmo is a discrete processing node in the pipeline.
   * Each gizmo runs in its own thread, pulling data from its input stream(s), processing it, and pushing results to its output stream(s).
   * Example gizmos include:
     * Video-sourcing gizmos that read frames from a webcam or file.
     * AI inference gizmos that run a model on incoming frames.
     * Video display or saving gizmos that show or store processed frames.
     * Gizmos that perform transformations (resizing, cropping, analyzing) on data.
   * Gizmos communicate via Streams. A gizmo output Stream can feed multiple downstream gizmos.
   * Gizmos keep a list of input streams that they are connected to.
   * Gizmos own their input streams.
3. **Composition**:
   * A [Composition](/degirum-tools/streams/streams_base.md#composition) is a container that holds and manages multiple gizmos (and their Streams).
   * Once gizmos are connected, you can call `composition.start()` to begin processing. Each gizmo `run()` method executes in a dedicated thread.
   * Call `composition.stop()` to gracefully stop processing and wait for threads to finish.
4. **StreamData** and **StreamMeta**:
   * Each item in the pipeline is encapsulated by a [StreamData](/degirum-tools/streams/streams_base.md#streamdata) object, which holds:
     * `data`: The actual payload (e.g., an image array, a frame).
     * `meta`: A [StreamMeta](/degirum-tools/streams/streams_base.md#streammeta) object that can hold extra metadata from each gizmo (e.g., a detection result, timestamps, bounding boxes, etc.).
       * Gizmos can append to [StreamMeta](/degirum-tools/streams/streams_base.md#streammeta) so that metadata accumulates across the pipeline.
5. **Metadata Flow (StreamMeta)**:
   * How [StreamMeta](/degirum-tools/streams/streams_base.md#streammeta) works:
     * [StreamMeta](/degirum-tools/streams/streams_base.md#streammeta) itself is a container that can hold any number of "meta info" objects.
     * Each meta info object is "tagged" with one or more string tags, such as `"dgt_video"`, `"dgt_inference"`, etc.
     * You append new meta info by calling `meta.append(my_info, [list_of_tags])`.
     * You can retrieve meta info objects by searching with `meta.find("tag")` (returns *all* matches) or `meta.find_last("tag")` (returns the *most recent* match).
     * **Important**: A gizmo generally clones (`.clone()`) the incoming [StreamMeta](/degirum-tools/streams/streams_base.md#streammeta) before appending its own metadata to avoid upstream side effects.
     * This design lets each gizmo add new metadata, while preserving what was provided by upstream gizmos.
   * High-Level Example:
     * A camera gizmo outputs frames with meta tagged `"dgt_video"` containing properties like FPS, width, height, etc.
     * An AI inference gizmo downstream takes `StreamData(data=frame, meta=...)`, runs inference, then:
       1. Clones the metadata container.
       2. Appends its inference results under the `"dgt_inference"` tag.
     * If *two* AI gizmos run in series, both will append metadata with the same `"dgt_inference"` tag. A later consumer can call `meta.find("dgt_inference")` to get both sets of results or `meta.find_last("dgt_inference")` to get the most recent result.

### Basic Usage Example <a href="#basic-usage-example" id="basic-usage-example"></a>

A simple pipeline might look like this:

{% code overflow="wrap" %}

```python
import time
import cv2
import degirum_tools.streams as streams

# Create gizmos. If you are on a laptop or have a webcam attached,
# VideoSourceGizmo(0) will typically select the default camera.
video_source = streams.VideoSourceGizmo(0)
video_display = streams.VideoDisplayGizmo("Camera Preview")

# Connect them
video_source >> video_display

# Build composition
composition = streams.Composition(video_source, video_display)
composition.start(wait=False)  # Don't block main thread

try:
    start_time = time.time()
    while time.time() - start_time < 10:  # Run for 10 seconds
        cv2.waitKey(5)  # Let OpenCV handle window events
finally:
    composition.stop()
    cv2.destroyAllWindows()
```

{% endcode %}

### Key Steps <a href="#key-steps" id="key-steps"></a>

1. **Create** your gizmos (e.g., `VideoSourceGizmo`, `VideoDisplayGizmo`, AI inference gizmos, etc.).
2. **Connect** them together using the `>>` operator (or `connect_to()` method) to form a processing graph. E.g.:

{% code overflow="wrap" %}

```
   source >> processor >> sink
```

{% endcode %}

3. **Initialize** a [Composition](/degirum-tools/streams/streams_base.md#composition) with the top-level gizmo(s).
4. **Start** the [Composition](/degirum-tools/streams/streams_base.md#composition) to launch each gizmo in its own thread.
5. (Optional) **Wait** for the pipeline to finish or perform other tasks. You can query statuses, queue sizes, or get partial results in real time.
6. **Stop** the pipeline when done.

### Advanced Topics <a href="#advanced-topics" id="advanced-topics"></a>

* **Non-blocking vs Blocking**: Streams can drop items if configured (`allow_drop=True`) to handle real-time feeds.
* **Multiple Inputs or Outputs**: Some gizmos can have multiple input streams and/or broadcast results to multiple outputs.
* **Error Handling**: If any gizmo encounters an error, the [Composition](/degirum-tools/streams/streams_base.md#composition) can stop the whole pipeline, allowing you to handle exceptions centrally.

For practical code examples, see the `dgstreams_demo.ipynb` notebook in the PySDKExamples.

## Functions <a href="#functions" id="functions"></a>

#### load\_composition(description, ...) <a href="#load_composition" id="load_composition"></a>

`load_composition(description, global_context=None, local_context=None)`

Load a [Composition](/degirum-tools/streams/streams_base.md#composition) of gizmos and connections from a description.

The description can be provided as a JSON/YAML file path, a YAML string, or a Python dictionary conforming to the JSON schema defined in `composition_definition_schema`.

Composition Description Schema (YAML):

{% code overflow="wrap" %}

```yaml
type: object
additionalProperties: false
required: [gizmos, connections]
properties:
    vars:
        type: object
        description: The collection of variables, keyed by variable name
        additionalProperties: false
        patternProperties:
            "^[a-zA-Z_][a-zA-Z0-9_]*$":
                oneOf:
                  - type: [string, number, boolean, array]
                    description: The variable value; can be $(expression) to evaluate
                  - type: object
                    description: The only object key is the class name to instantiate; value are the constructor parameters
                    additionalProperties: false
                    minProperties: 1
                    maxProperties: 1
                    patternProperties:
                        "^[a-zA-Z_][a-zA-Z0-9_.]*$":
                            type: object
                            description: The constructor parameters of the object
                            additionalProperties: true
    gizmos:
        type: object
        description: The collection of gizmos, keyed by gizmo instance name
        additionalProperties: false
        patternProperties:
            "^[a-zA-Z_][a-zA-Z0-9_]*$":
                oneOf:
                  - type: string
                    description: The gizmo class name to instantiate (if no parameters are needed)
                  - type: object
                    description: The only object key is the class name to instantiate; value are the constructor parameters
                    additionalProperties: false
                    minProperties: 1
                    maxProperties: 1
                    patternProperties:
                        "^[a-zA-Z_][a-zA-Z0-9_.]*$":
                            type: object
                            description: The constructor parameters of the gizmo
                            additionalProperties: true
    connections:
        type: array
        description: The list of connections between gizmos
        items:
            type: array
            description: The connection between gizmos
            items:
                oneOf:
                    - type: string
                    - type: array
                      description: Gizmo with input index
                      prefixItems:
                        - type: string
                        - type: number
                      items: false
```

{% endcode %}

Parameters:

| Name             | Type          | Description                                                                                                   | Default    |
| ---------------- | ------------- | ------------------------------------------------------------------------------------------------------------- | ---------- |
| `description`    | `str or dict` | A YAML string or dict describing the Composition, or a path to a .json/.yaml file containing the description. | *required* |
| `global_context` | `dict`        | Global context for evaluating expressions in the description (variables, etc.). Defaults to None.             | `None`     |
| `local_context`  | `dict`        | Local context for evaluating expressions. Defaults to None.                                                   | `None`     |

Returns:

| Name          | Type          | Description                                                                                                           |
| ------------- | ------------- | --------------------------------------------------------------------------------------------------------------------- |
| `Composition` | `Composition` | A [Composition](/degirum-tools/streams/streams_base.md#composition) object representing the described gizmo pipeline. |