pipelogic.worker

The entry point for every Python component.

from pipelogic.worker import run, config, virtual_in, virtual_out

run(worker_function, **options)

Registers your worker function with the runtime and blocks until the backend closes.

Synchronous (default) mode

from pipelogic.worker import run

def double(x):
    return x * 2

run(double)

One message in per stream per tick, one message out per stream per tick. The function arguments are positional, in the order declared by component.yml.

Multi-output

A function whose component declares output_types returns a tuple in declared order:

worker:
  input_type: Image
  output_types:
    - BoundingBox
    - Image

Stateful

Pass initial_state and accept a state kwarg. Return {"output": ..., "state": ...}. State persists across restarts.

def counter(message, state):
    new_state = state + 1
    return {"output": new_state, "state": new_state}

run(counter, initial_state=0)

Virtual input

Define a virtual_input parameter on your worker. The runtime detects the parameter name via inspect.signature and supplies a queue you can push to from another thread:

import threading
from pipelogic.worker import run, virtual_in

def from_camera(virtual_input):
    return virtual_input[0]

def producer():
    while True:
        virtual_in.push(grab_frame(), blocking=False)

threading.Thread(target=producer, daemon=True).start()
run(from_camera)

Virtual output

from pipelogic.worker import run, virtual_out

def emit_externally(image):
    return {"output": image, "virtual_output": [some_payload]}

def consumer(item):
    upload_to_external_system(item)

virtual_out.set_on_push(consumer)
run(emit_externally, use_virtual_output=True)

Streaming mode

Pass initial_pull_request and the worker now receives lists of messages and must return a pull_request:

from pipelogic.worker import run
from pipelogic.types import InputsPullRequest, InputsPullOption, StreamPullShift

INITIAL = InputsPullRequest(InputsPullOption([
    StreamPullShift.from_fixed(0, 1, 1),
]))

def chunked(messages):
    return {
        "output": [process(m) for m in messages],
        "pull_request": INITIAL,
    }

run(chunked, initial_pull_request=INITIAL)

StreamPullShift.from_fixed(skip, min_pull, max_pull) declares how many messages to skip and how many to receive. Use streaming mode when stream rates are asymmetric — e.g. an audio summarizer that pulls one second of audio at a time and emits one summary every N seconds.

config

Singleton exposing config_schema parameters from component.yml as attributes:

from pipelogic.worker import config

threshold = float(config.threshold)
mean = tuple(config.mean)             # (Double, Double, Double) -> tuple of floats
language = str(config.language)

The runtime auto-converts each value to the wrapper type that matches the declared type: field in component.yml.

config.sync()

Picks up live updates pushed from the platform UI. Returns a dict of changed keys:

def expensive_worker(image):
    changed = config.sync()
    if "model_path" in changed:
        reload_model()
    return run_inference(image)

Most components don't need this — re-reading the value each tick is enough.

virtual_in: VirtualInputStream

Module-level singleton bound to the worker's virtual input queue.

virtual_out: VirtualOutputStream

Module-level singleton bound to the worker's virtual output queue.

Streaming-mode primitives

from pipelogic.worker import (
    InputsPullRequest, InputsPullOption, StreamPullShift,
    VirtualInputStream, VirtualOutputStream,
)

What's next

• Concepts — how the run loop fits into the broader system.
• API: cv — the typed wrappers you'll most often return from a worker.
• Recipes — stateful counter, sliding-window smoother, virtual-input camera.