HTTP Control

SGN can expose REST endpoints for runtime monitoring and control of pipeline elements using HTTPControl.

Basic Setup

Wrap your pipeline execution with the HTTPControl context manager:

from sgn import Pipeline
from sgn.control import HTTPControl

p = Pipeline()
# ... build pipeline ...

with HTTPControl() as control:
    p.run()

This starts a Bottle web server in a background thread. The server address is printed to stderr and written to a registry file (registry.txt by default).

HTTP-Enabled Elements

Inherit from HTTPControlSourceElement, HTTPControlTransformElement, or HTTPControlSinkElement to automatically register GET and POST routes for your element:

from dataclasses import dataclass
from sgn import SourcePad, SinkPad, Frame
from sgn.control import HTTPControl, HTTPControlTransformElement


@dataclass
class GainControl(HTTPControlTransformElement):
    """A transform whose gain can be adjusted at runtime via HTTP."""
    gain: float = 1.0

    def pull(self, pad: SinkPad, frame: Frame) -> None:
        self.in_frame = frame

    def internal(self) -> None:
        # Read updates from HTTP POST and write current state to HTTP GET
        state = {"gain": self.gain}
        HTTPControl.exchange_state(self.name, state)
        self.gain = state["gain"]

    def new(self, pad: SourcePad) -> Frame:
        result = None
        if self.in_frame.data is not None:
            result = self.in_frame.data * self.gain
        return Frame(data=result, EOS=self.in_frame.EOS)

REST Endpoints

For an element named "my_elem" with tag None:

Method	URL	Purpose
POST	/post/my_elem	Send JSON to update element state
GET	/get/my_elem	Retrieve current element state
GET	/get/my_elem/<key>	Retrieve a specific field
GET	/get/my_elem/<key>/<key2>	Retrieve a nested field

Updating State

curl -X POST http://localhost:8080/post/my_elem \
     -H "Content-Type: application/json" \
     -d '{"gain": 2.5}'

Reading State

curl http://localhost:8080/get/my_elem
curl http://localhost:8080/get/my_elem/gain

Configuration

Set class-level attributes on HTTPControl before creating the context:

HTTPControl.port = 9090          # Default: 0 (auto-assign)
HTTPControl.host = "0.0.0.0"    # Default: auto-detected hostname
HTTPControl.tag = "v1"          # URL prefix: /v1/get/..., /v1/post/...

with HTTPControl(registry_file="my_registry.txt") as control:
    p.run()

exchange_state()

The HTTPControl.exchange_state(name, state_dict) method handles the bidirectional update: it drains the POST queue, updates matching keys in state_dict, then writes the current state to the GET queue.

Call it from internal() so it runs once per pipeline iteration.