videoflow.core package

Submodules

videoflow.core.compiler module

Turns a videoflow.core.flow.Flow (a built, validated graph) into a list of per-node NodeSpec``s: everything an execution engine needs to launch one worker per node, without any of the live ``Node objects. A NodeSpec is fully JSON-serializable, which is what lets it cross into a separate process or a Kubernetes pod as environment variables / a ConfigMap.

class videoflow.core.compiler.NodeSpec(name: str, node_class: str | None, params: Dict[str, Any], parents: List[str], kind: str, has_children: bool, nb_tasks: int, device_type: str, is_finite: bool, image: str | None = None, partition_by: str | None = None, join_policy: Dict[str, Any] | None = None, component_ref: str | None = None, descriptor: Dict[str, Any] | None = None, command: List[str] | None = None, protocol_version: int | None = None, gpu_count: int = 1, gpu_resource_name: str | None = None, blob_readers: int | None = None)

Bases: object

A flat, serializable description of one node’s deployment.

• Attributes:

  • name: node’s stable name (unique in the flow).

  • node_class: fully-qualified import path, e.g. videoflow.processors.basic.IdentityProcessor.

  • params: dict from node.get_params() — the kwargs to reconstruct it.

  • parents: list of parent node names, in process() positional order.

  • kind: one of producer/processor/consumer.

  • has_children: whether anything downstream consumes this node’s output.

  • nb_tasks: desired replica count (processors only; 1 otherwise).

  • device_type: ‘cpu’ or ‘gpu’ (processors only).

  • gpu_count: GPUs each replica requests (GPU processors only; default 1).

  • gpu_resource_name: extended-resource name each replica requests, or None to use the deploy-time default (nvidia.com/gpu).

  • is_finite: for producers, whether next() self-terminates.

  • image: the container image ref declared on the node, or None (the deploy-time default/override supplies it — see videoflow.deploy.images).

  • blob_readers: how many downstream reads each message this node publishes receives (Σ over children of nb_tasks if partitioned else 1); drives refcounted blob reclamation (PROTOCOL.md BLOB-5). 0 for leaves; None when unknown (a legacy spec), which disables reclamation.

The field order below is the constructor signature — callers pass these positionally (NodeSpec('n', 'pkg.Cls', {}, [], 'processor', ...)), so reordering or inserting a field is a breaking change. Not frozen: a spec is a plain mutable record, and to_dict/from_dict stay explicit because they are the VF_FLOW_SPECS_JSON serialization boundary — asdict() would deep-copy and rewrite the nested params/descriptor values.

blob_readers: int | None = None

command: List[str] | None = None

component_ref: str | None = None

descriptor: Dict[str, Any] | None = None

device_type: str

classmethod from_dict(d: Dict[str, Any]) → NodeSpec

gpu_count: int = 1

gpu_resource_name: str | None = None

has_children: bool

image: str | None = None

is_finite: bool

property is_native: bool

runs its own image entrypoint (no node_class) and must speak the protobuf wire.

Type:

A non-Python component

property is_remote: bool

Loaded from a component descriptor (Python or native), vs a native Python graph node.

join_policy: Dict[str, Any] | None = None

kind: str

name: str

nb_tasks: int

node_class: str | None

params: Dict[str, Any]

parents: List[str]

partition_by: str | None = None

protocol_version: int | None = None

to_dict() → Dict[str, Any]

videoflow.core.compiler.compile_flow(flow: Flow, envelope_version: int | None = None) → List[NodeSpec]

• Arguments:

  • flow: a built videoflow.core.flow.Flow (do NOT call .run() on it first).

  • envelope_version: the wire version this flow will deploy with. The only supported version is 4 (protobuf); an explicit incompatible pin is rejected. Defaults to the ambient DEFAULT_ENVELOPE_VERSION.

• Returns:

  • list of NodeSpec, one per node in the flow’s topological sort.

videoflow.core.compiler.has_native_components(specs: List[NodeSpec]) → bool

Whether any node is a native (non-Python) component — the ones that force the protobuf wire.

videoflow.core.compiler.has_remote_components(specs: List[NodeSpec]) → bool

Whether any node came from a component descriptor (Python or native).

videoflow.core.compiler.specs_from_tasks_data(tasks_data: List[tuple]) → List[NodeSpec]

Converts build_tasks_data output — tuples of (node, parent_names, is_last) — into a list of serializable NodeSpec.

videoflow.core.compiler.validate_wire_compatibility(specs: List[NodeSpec], envelope_version: int | None) → None

The wire is the single language-neutral protobuf envelope (version 4) for every flow. Reject an explicit pin to any other version at compile/deploy time so the failure is actionable here rather than a worker refusing to start. (PROTOCOL.md §4.)

videoflow.core.constants module

videoflow.core.context module

Runtime context optionally handed to a node’s lifecycle/processing methods.

A node method (open/next/process/consume/close) may declare a final ctx (or context) parameter; if it does, the task passes a RuntimeContext so the node can read run identity and set a partition key on its output without depending on any global state. Methods that don’t declare it are called exactly as before, so this is fully backward compatible with existing nodes.

class videoflow.core.context.RuntimeContext(flow_id: str, run_id: str, node_name: str, replica_id: int, logger: Logger, messenger: Messenger | None = None)

Bases: object

• Attributes:

  • flow_id / run_id / node_name / replica_id: identity of this running node.

  • logger: a standard library logger scoped to the node.

property input_info: Dict[str, Any] | None

Per-parent envelope info for the input group currently being processed: {parent_name: {'event_ts': ..., 'metadata': ..., 'trace_id': ..., 'seq': ...}} (None values for parents missing from a quorum emission; lists for collect parents). None for producers. Lets fusion code read each input’s exact event time without changing process() signatures.

set_event_timestamp(value: float) → None

Set the event time (epoch seconds) stamped on this node’s next published output — when the underlying real-world event was captured (a frame’s capture time, a sensor sample’s timestamp). Producers of time-sensitive data should call this from next(); downstream nodes inherit their input group’s event time automatically, so they rarely need to. Time- aligned joins (JoinPolicy(mode='time')) group on this value.

set_partition_key(value: Any) → None

Set the partition key carried on this node’s next published output, so a downstream partitioned node can route by a business key. Applied to the message metadata under the reserved field _partition_key.

videoflow.core.engine module

class videoflow.core.engine.ExecutionEngine

Bases: object

Defines the interface of the execution environment — how tasks are physically started (as local OS processes for development, or as Kubernetes pods in production) and how flow-wide termination and completion are observed.

allocate_and_run_tasks(tasks_data: List[tuple] | None, flow_id: str, flow_type: str, run_id: str) → None

Defines a template with the order of methods that need to run in order to allocate and run tasks.

• Arguments:

  • tasks_data: list of tuples (node, parent_names : [str], is_last : bool), or None when the engine already holds pre-compiled specs.

  • flow_id: stable identifier for this flow (constant across runs).

  • flow_type: ‘realtime’ or ‘batch’.

  • run_id: per-run identifier that scopes this execution’s broker streams.

join_task_processes() → None

Blocking method. It is supposed to make the calling process sleep until all task processes have finished processing.

signal_flow_termination() → None

Signals the execution environment that the flow needs to stop.

class videoflow.core.engine.Messenger

Bases: object

Utility class that tasks use to receive input and write output, over a message broker (see videoflow.messaging.nats_messenger.NATSMessenger for the concrete implementation). A Messenger is bound to exactly one node in the graph. It knows that node’s own broker subject (for publishing) and its real parents’ subjects (for receiving) — routing is by node name, not by position in a topological sort, so it works correctly for arbitrary DAGs (multi-parent joins, multi-producer graphs).

ack_inputs() → None

Acknowledge the input group last returned by receive_message — called by the task only after the node has processed it (and, for a processor, published its output). This ack-after-process ordering is what makes a crash mid-processing safe: the un-acked message is redelivered instead of lost.

check_for_termination() → bool

Returns true if a flow-wide termination signal has been received on the control channel. Used by videoflow.core.task.ProducerTask to stop pulling new input even before it naturally reaches StopIteration.

close() → None

Release any broker resources held by the messenger. Default: no-op.

fail_inputs(exc: BaseException) → None

Report that the node raised while processing the last input group. The messenger decides whether to redeliver (BATCH, up to a retry limit, then dead-letter) or drop (REALTIME).

last_input_info() → Dict[str, Any] | None

Per-parent envelope info (event_ts, metadata, trace_id, seq) for the input group last returned by receive_message (exposed to nodes as ctx.input_info). Default: None.

last_input_key() → str | None

A stable identity for the input group last returned by receive_message, used as a sink idempotency key. Default: None (no idempotency).

publish_message(message: Any, metadata: Dict[str, Any] | None = None) → None

Publishes this node’s own output message. Depending on the flow’s configured retention policy (REALTIME vs BATCH), this may drop the message if downstream consumers are behind.

publish_stop_signal() → None

Publishes a termination marker on this node’s own subject. Unlike publish_message, this is never dropped regardless of retention policy — every downstream consumer must observe it exactly once.

receive_message() → Dict[str, Dict[str, Any]]

Blocks until this node has received a complete input: one message from every real parent, all derived from the same upstream originating event (see the trace_id propagation scheme in the concrete implementation) — or until every parent has signaled termination.

• Returns:

  • a dict {parent_name: {"message": ..., "metadata": ..., "is_stop_signal": bool}} with exactly one entry per real parent of this node.

set_output_event_timestamp(value: float) → None

Set the event time (epoch seconds) stamped on this node’s next published output (via RuntimeContext.set_event_timestamp) — when the underlying real-world event was captured. Producers of time-sensitive data (cameras, sensors) should set this; downstream nodes inherit their input group’s event time automatically. Default: no-op.

set_output_partition_key(value: Any) → None

Set the partition key attached to this node’s next published output (via RuntimeContext.set_partition_key), so a downstream partitioned node can route by a business key. Default: no-op.

videoflow.core.flow module

class videoflow.core.flow.Flow(consumers: List[ConsumerNode], flow_type: str = 'realtime', flow_id: str | None = None)

Bases: object

Represents a flow of data from producer nodes to consumer nodes, over the directed acyclic graph formed by however you’ve wired up Node instances via child(*parents).

• Arguments:

  • consumers: a list of consumer nodes of type videoflow.core.node.ConsumerNode. Producers are discovered automatically by walking parents back from these.

  • flow_type: one of ‘realtime’ or ‘batch’. Controls the message broker’s retention/discard policy for every edge in the flow (drop-when-full vs. block/at-least-once).

  • flow_id: a stable identifier for this flow, used to namespace broker subjects and Kubernetes resources. Auto-generated if not given.

property flow_id: str

property flow_type: str

join() → None

Blocking method. Will make the process that calls this method block until the flow finishes running naturally.

run(execution_engine: ExecutionEngine, run_id: str | None = None) → None

Starts the flow using the given ExecutionEngine (e.g. videoflow.engines.local.LocalProcessEngine or videoflow.engines.kubernetes.KubernetesExecutionEngine). Non-blocking: returns once tasks have been allocated/started.

• Arguments:

  • run_id: optional stable per-run id (auto-generated if omitted). Every broker stream/subject for this run is namespaced by it, so re-running the same flow never collides with a previous run’s streams.

property run_id: str | None

The id of the most recent (or in-progress) run, or None before run().

stop() → None

Blocking method. Stops the flow. Makes the execution environment send a flow termination signal.

tasks_data() → List[tuple]

topological_sort() → list

Returns the topologically-sorted list of nodes in this flow. Exposed so the Kubernetes manifest-generation CLI (videoflow.deploy.cli) can inspect the graph without needing to call .run().

videoflow.core.flow.build_tasks_data(graph_engine: GraphEngine) → List[tuple]

Turns a validated GraphEngine into the list of (node, parent_names, is_last) tuples that both the local execution engine and the Kubernetes compiler (videoflow.core.compiler) use to allocate one task per node.

• Returns:

  • tasks_data: list of tuples (node, parent_names : [str], is_last : bool)

videoflow.core.graph module

class videoflow.core.graph.GraphEngine(producers: List[ProducerNode], consumers: List[ConsumerNode])

Bases: object

Validates and topologically sorts a computation graph.

• Arguments:

  • producers: list of ProducerNode instances that are the roots of the graph. Any number of producers is supported (a flow may ingest from several independent sources, e.g. multiple cameras, and fan them into shared downstream processors).

  • consumers: list of ConsumerNode instances that are the leaves of the graph.

• Raises:

  • AttributeError if any of producers is not a ProducerNode.

  • ValueError if the graph has a cycle, if any consumer is unreachable from the given producers, or if two or more nodes share the same name.

topological_sort() → list

videoflow.core.node module

class videoflow.core.node.ConsumerNode(metadata: bool = False, name: str | None = None, join_policy: JoinPolicy | dict | None = None, idempotent: bool = False, **kwargs: Any)

Bases: Leaf

• Arguments:

  • metadata (boolean): By default is False. If True, instead of receiving output of parent nodes, receives metadata produced by parent nodes.

  • name (str): see Node.

consume(item: Any) → None

Method definition that needs to be implemented by subclasses.

• Arguments:

  • item: the item being received as input (or consumed).

property idempotent: bool

If True (and the flow has a Redis blob/idempotency store), the consumer’s side effects are deduplicated across redelivery.

property join_policy: JoinPolicy | None

property metadata: bool

property partition_by: str | None

class videoflow.core.node.FunctionProcessorNode(processor_function: Callable[[Any], Any], nb_tasks: int = 1, device_type: str = 'cpu', name: str | None = None, **kwargs: Any)

Bases: ProcessorNode

get_params() → NoReturn

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

process(inp: Any) → Any

Method definition that needs to be implemented by subclasses.

• Arguments:

  • inp: object or list of objects being received for processing from parent nodes.

• Returns:

  • the output being consumed by child nodes.

videoflow.core.node.JoinPolicyArg

a policy object, the plain dict it serializes to (how it arrives when a worker reconstructs the node), or None.

Type:

What a node’s join_policy= argument accepts

alias of JoinPolicy | dict | None

class videoflow.core.node.Leaf(*args: Any, **kwargs: Any)

Bases: Node

Node with no children.

class videoflow.core.node.ModuleNode(entry_node: Node, exit_node: Node, *args: Any, **kwargs: Any)

Bases: Node

Module node that wraps a subgraph of computation. Each node of the Module must be a ProcessorNode or a ModuleNode itself. For simplicity, a module node has exaclty one node as entry point, and exactly one node as exit point. If for some reason a ModuleNode has flag one_process set to True:

• Then any module within the subgraph must also be of that type, or an exception will be thrown.

• No process inside the module can be allocated to a gpu, or an exception will be thrown

• Arguments:

  • entry_node (Node): The node that sits at the top of the subgraph

  • exit_node (Node): The node that sits at the top of the subgraph

• Raises:

  • ValueError if:

    • There is at least one node in the sequence that is not instance of ProcessorNode or of ModuleNode

    • There is a cycle in the subgraph

    • The exit_node is not reachable from the entry_node

    • The flag one_process is set to True, and any of the following conditions is true:

      • There is a ModuleNode within the subgraph that does not have that flag set to true too.

      • There is at least one node in the sequence that has device_type GPU

get_params() → NoReturn

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

property nodes: List[ProcessorNode]

class videoflow.core.node.Node(name: str | None = None, image: str | None = None)

Bases: object

Represents a computational node in the graph. It is also a callable object. It can be call with the list of parents on which it depends.

• Arguments:

  • name (str): a unique, stable identifier for this node within a flow. Used as the node’s identity everywhere outside the process that built the graph: message broker subject names, Kubernetes resource names, and logging. If not given, one is auto-generated from the class name and a per-class construction counter; uniqueness within a flow is validated when the flow is built (see videoflow.core.graph.GraphEngine), not at construction time, since a collision can only be detected once the whole graph is known.

  • image (str): the container image ref that this node’s worker runs in on Kubernetes, e.g. ghcr.io/me/app:v1. Declare it here when a node intrinsically needs a specific environment (a GPU model image, say). If omitted, the image is taken from the deploy-time default (--image); a deploy-time --image-override beats both. Ignored by the local engine, which runs workers in the current Python environment.

add_child(child: Node) → None

Adds child to the set of childs that depend on it.

property children: Set[Node] | None

Returns a set of the child nodes

close() → None

This method is called by the task running after finishing doing all consuming, processing or producing because of and end signal receival. Should be used to close any resources that were opened by the open() method, such as files, tensorflow sessions, etc.

get_params() → Dict[str, Any]

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

property image: str | None

The container image ref declared for this node (or None to use the deploy-time default).

property name: str

The stable, unique-within-a-flow string identity of the node. This is the identifier used for broker subjects, Kubernetes resource names, and node reconstruction in worker processes.

open() → None

This method is called by the task runner before doing any consuming, processing or producing. Should be used to open any resources that will be needed during the life of the task, such as opening files, tensorflow sessions, etc.

property parents: List[Node] | None

Returns a list with the parent nodes

remove_child(child: Node) → None

class videoflow.core.node.OneTaskProcessorNode(device_type: str = 'cpu', name: str | None = None, **kwargs: Any)

Bases: ProcessorNode

Used for processes that keep internal state so they are easily parallelizable. The main use of this class if for processes that can only run one task, such as trackers and aggregators.

class videoflow.core.node.ProcessorNode(nb_tasks: int = 1, device_type: str = 'cpu', name: str | None = None, partition_by: str | None = None, join_policy: JoinPolicy | dict | None = None, gpu_count: int = 1, gpu_resource_name: str | None = None, **kwargs: Any)

Bases: Node

• Arguments:

  • nb_tasks (int): number of parallel replicas to allocate for this processor.

  • device_type (str): videoflow.core.constants.CPU or GPU.

  • partition_by (str): if set (and nb_tasks > 1), replicas partition the input by a key instead of competing for it: each message is handled by exactly one replica chosen by hash(key) % nb_tasks. The key is the message’s trace_id (the special value 'trace_id', which co-locates both halves of a join on the same replica) or a metadata field name. Required for a multi-parent (join) node with nb_tasks > 1.

  • join_policy (JoinPolicy | dict): for multi-parent nodes, how to handle a join group that never completes (timeout + missing policy). Defaults per flow type when unset.

  • gpu_count (int): GPUs each replica requests on Kubernetes (device_type=GPU only; ignored locally). Whole devices — the resource is not overcommittable.

  • gpu_resource_name (str): Kubernetes extended-resource name each replica requests, when the cluster does not expose plain nvidia.com/gpu — e.g. a MIG profile (nvidia.com/mig-1g.10gb) or a renamed time-sliced resource (nvidia.com/gpu.shared). None defers to the deploy-time default (--gpu-resource-name, else nvidia.com/gpu).

  • name (str): see Node.

change_device(device_type: str) → None

property device_type: str

Returns the preferred device type to use to run the processor’s code

property gpu_count: int

GPUs each replica requests on Kubernetes (meaningful only when device_type is GPU).

property gpu_resource_name: str | None

Extended-resource name each replica requests, or None for the deploy default.

property join_policy: JoinPolicy | None

Returns the JoinPolicy object (or None), reconstructed from the stored dict.

property nb_tasks: int

Returns the number of tasks to allocate to this processor

property partition_by: str | None

process(inp: Any) → Any

Method definition that needs to be implemented by subclasses.

• Arguments:

  • inp: object or list of objects being received for processing from parent nodes.

• Returns:

  • the output being consumed by child nodes.

class videoflow.core.node.ProducerNode(is_finite: bool = True, name: str | None = None, **kwargs: Any)

Bases: Node

The producer node does not receive input, and produces input. Each time the next() method is called, it produces a new input.

It would have been more natural to implement the ProducerNode as a generator, but a node is shipped to its worker as (class path, get_params()) and rebuilt there with type(node)(**get_params()) — a generator has no such reconstructable form, so it cannot cross into the worker process.

• Arguments:

  • is_finite (bool): True (the default) if a call to next() will eventually raise StopIteration on its own (e.g. reading a fixed video file). Set to False for producers that read from an unbounded/live source (e.g. an RTSP stream) and only stop when told to. The Kubernetes execution engine uses this to decide whether to deploy the producer as a Job (finite) or a Deployment (infinite).

  • name (str): see Node.

property is_finite: bool

next() → Any

Returns next produced element.

Raises StopIteration after the last element has been produced and a call to self.next happens.

class videoflow.core.node.TaskModuleNode(entry_node: ProcessorNode, exit_node: ProcessorNode, nb_tasks: int = 1, name: str | None = None, **kwargs: Any)

Bases: ProcessorNode

Processor node that wraps a graph of processor nodes. This has the effect that instead of allocating one task per processor node in the graph, only one task process is allocated for the entire subgraph.

• Arguments:

  • entry_node (Node): The node that sits at the top of the subgraph

  • exit_node (Node): The node that sits at the top of the subgraph

  • nb_tasks (int) The number of parallel tasks to allocate

• Raises:

  • ValueError if:

    • There is at least one node in the subgraph that is not instance of ProcessorNode

    • nb_tasks parameter is greater than one and there is at least one node in the sequence that derives from OneTaskProcessorNode.

    • There is at least one node in the sequence that has device_type GPU

    • The subgraph has less than one node.

    • There is a node of type TaskModuleNode among the nodes of the subgraph.

Note: because it wraps live references to other nodes, TaskModuleNode does not support get_params()-based reconstruction and is not yet supported by the distributed Kubernetes execution path (see videoflow.core.compiler).

get_params() → NoReturn

Returns a JSON-serializable dict of keyword arguments sufficient to reconstruct an equivalent node via type(node)(**node.get_params()). This is what lets a worker process (running in its own container, with no access to the object that built the graph) recreate the exact node it is responsible for running.

Default implementation: walks every class in this node’s MRO that declares its own __init__, inspects that __init__’s named parameters (skipping *args/**kwargs), and for each parameter name looks up self._<name> and then self.<name>. This matches the prevailing convention in this codebase of storing constructor arguments verbatim as an underscore-prefixed attribute of the same name.

Subclasses whose constructor arguments aren’t stored under a matching attribute name, or that accept non-serializable arguments (e.g. references to other nodes, as TaskModuleNode does), must override this method.

process(*inp: Any) → Any

Method definition that needs to be implemented by subclasses.

• Arguments:

  • inp: object or list of objects being received for processing from parent nodes.

• Returns:

  • the output being consumed by child nodes.

videoflow.core.policies module

Policies that tune how a multi-parent (join) node aligns its inputs.

A JoinPolicy is attached to a node and travels with it as a plain dict (so it survives get_params() serialization into a worker). It decides two things:

• How input groups are formed (mode): by lineage (trace, the default — inputs that descend from the same originating message of a single producer) or by event time (time — inputs whose event_ts fall within a tolerance of each other, which is how streams from independent producers such as multiple cameras and sensors are fused).

• What to do with a group that never completes — a real possibility when one branch drops a message (REALTIME) or stalls: timeout + missing policy, and for time-aligned joins, an optional quorum that lets a late group emit with the parents it has.

videoflow.core.policies.JOIN_TRACE = 'trace'

How input groups are formed at a multi-parent node.

class videoflow.core.policies.JoinPolicy(timeout_seconds: float | None = None, missing: str = 'drop', max_pending: int = 256, mode: str = 'trace', tolerance_ms: float | None = None, quorum: int | None = None, collect: dict | None = None)

Bases: object

• Arguments:

  • timeout_seconds: how long to wait for the rest of a join group before applying missing (or emitting a quorum group). None means no timeout (wait forever). For mode='time' this is the lateness bound: the answer to “how long after a window’s first message may a straggler still arrive”.

  • missing: one of drop / wait / error (see constants above).

  • max_pending: hard cap on buffered incomplete groups; the oldest is evicted (as drop) beyond this, protecting against unbounded memory.

  • mode: trace (default) or time. time groups inputs whose event_ts (stamped by the producers) are within tolerance_ms of each other, instead of requiring a shared upstream trace id — required to join branches that descend from different producers.

  • tolerance_ms: (time mode only, required) two messages from different parents belong to the same group when their event times differ by at most this much. Pick it below the fastest parent’s inter-message period (e.g. for 50fps cameras, < 20ms).

  • quorum: (time mode only) minimum number of synchronized parents that must be present for a timed-out group to still be emitted (missing parents are passed to process() as None). None (default) means all parents are required and a timed-out group is handled by missing. With N cameras, quorum=k gives “emit with at least k views”. Requires timeout_seconds.

  • collect: (time mode only) dict {parent_name: window_ms} marking high-rate parents (e.g. a 500Hz sensor vs 50fps cameras) that should not join 1:1: every message of that parent whose event_ts is within window_ms of the group’s time is delivered as a list in that parent’s position. Collect parents never gate completeness and don’t count toward quorum; a group holds for the largest collect window after completing so trailing samples can arrive.

classmethod default_for(flow_type: str) → JoinPolicy

BATCH waits (completeness matters; bounded by max_pending). REALTIME times out and drops (a dropped sibling frame must not stall the join forever).

classmethod from_dict(d: Dict[str, Any] | None) → JoinPolicy | None

to_dict() → Dict[str, Any]

videoflow.core.policies.MISSING_DROP = 'drop'

What to do with an incomplete join group once it times out.

videoflow.core.remote module

Remote (language-agnostic) components in a Python-authored graph.

A component(ref, params=...) node stands in for a component that runs as its own container image and is authored in any language. It behaves like a normal Producer/Processor/Consumer node for graph-building (wiring, validation, the compiler, manifests), but it carries no Python implementation — its next/process/consume run out-of-process in the vendor image, driven by that image’s own SDK speaking the wire protocol (spec/PROTOCOL.md). The Python process only ever builds and compiles the graph; it never imports the component.

The three concrete kinds subclass the existing node base classes so every isinstance dispatch in graph.py/compiler.py/flow.py keeps working unchanged; RemoteNodeMixin marks them for the compiler, which records a component_ref + descriptor instead of a Python node_class.

class videoflow.core.remote.RemoteConsumer(component_ref: str, descriptor: ComponentDescriptor, params: Dict[str, Any], metadata: bool = False, idempotent: bool = False, join_policy: JoinPolicy | dict | None = None, name: str | None = None, image: str | None = None)

Bases: RemoteNodeMixin, ConsumerNode

consume(*inputs: Any) → None

Method definition that needs to be implemented by subclasses.

• Arguments:

  • item: the item being received as input (or consumed).

class videoflow.core.remote.RemoteNodeMixin

Bases: object

Shared state/identity for the three remote node kinds.

property component_command: List[str] | None

property component_ref: str

property descriptor: ComponentDescriptor

get_params() → Dict[str, Any]

The component’s own params, delivered via VF_NODE_PARAMS_JSON. For a native component these are the only params (routing settings reach it via env); for a Python component, the reconstructing worker calls pythonClass(**params), so node-level settings the class needs (nb_tasks/device_type) are merged in — matching what a normal node’s get_params() would carry. Every videoflow node constructor accepts these via **kwargs.

property local_command: List[str] | None

class videoflow.core.remote.RemoteProcessor(component_ref: str, descriptor: ComponentDescriptor, params: Dict[str, Any], nb_tasks: int = 1, device_type: str = 'cpu', partition_by: str | None = None, join_policy: JoinPolicy | dict | None = None, name: str | None = None, image: str | None = None, gpu_count: int = 1, gpu_resource_name: str | None = None)

Bases: RemoteNodeMixin, ProcessorNode

process(*inputs: Any) → Any

Method definition that needs to be implemented by subclasses.

• Arguments:

  • inp: object or list of objects being received for processing from parent nodes.

• Returns:

  • the output being consumed by child nodes.

class videoflow.core.remote.RemoteProducer(component_ref: str, descriptor: ComponentDescriptor, params: Dict[str, Any], is_finite: bool = True, name: str | None = None, image: str | None = None)

Bases: RemoteNodeMixin, ProducerNode

next() → Any

Returns next produced element.

Raises StopIteration after the last element has been produced and a call to self.next happens.

videoflow.core.remote.component(ref: str | ComponentDescriptor, params: Dict[str, Any] | None = None, name: str | None = None, nb_tasks: int = 1, device_type: str = 'cpu', partition_by: str | None = None, join_policy: JoinPolicy | dict | None = None, image: str | None = None, is_finite: bool | None = None, metadata: bool = False, idempotent: bool = False, gpu_count: int = 1, gpu_resource_name: str | None = None) → Node

Create a graph node backed by a language-agnostic component described by ref (a path to a component.yaml or, later, an oci:// ref). Returns a Producer/Processor/Consumer node per the descriptor’s role, so it wires into a Python graph exactly like a native node:

from videoflow.core import Flow, component
tracker = component('components/sort', params={'iou_threshold': 0.3})(detector)

Validation happens here, at graph-build time, so a misconfiguration is a build error in the authoring script rather than a crash inside the vendor container: params are checked against the descriptor’s JSON Schema, device_type against the declared devices, the image against what the descriptor provides, and the join policy against what the component says it accepts.

• Arguments:

  • ref: descriptor reference (local path now; oci:// in Phase 6), or an already-loaded ComponentDescriptor.

  • params: component params dict (validated + defaulted against the descriptor).

  • device_type: ‘cpu’ or ‘gpu’ (must be in the descriptor’s device list).

  • image: explicit image ref override (else the descriptor’s image for the device).

  • is_finite: producers only; defaults to the descriptor’s finite.

  • nb_tasks/partition_by/join_policy/metadata/idempotent: as on the native nodes.

  • gpu_count/gpu_resource_name: processors only; as on ProcessorNode.

videoflow.core.task module

class videoflow.core.task.ConsumerTask(consumer: ConsumerNode, messenger: Messenger, has_children: bool, parent_names: List[str], ctx: RuntimeContext | None = None, idempotency_store: IdempotencyStore | None = None)

Bases: NodeTask

It runs forever, blocking until it receives a message from every parent node through the messenger. It consumes the message and does not publish anything back down the pipe — consumers are the leaves of the graph.

class videoflow.core.task.NodeTask(computation_node: Node, messenger: Messenger, has_children: bool, ctx: RuntimeContext | None = None)

Bases: Task

A NodeTask is a wrapper around a videoflow.core.node.Node that is able to interact with the execution environment through a messenger. Nodes receive input and/or produce output, but tasks are the ones that run in infinite loops, receiving inputs from the environment and passing them to the computation node, and taking outputs from the computation node and passing them to the environment.

• Arguments:

  • computation_node

  • messenger (Messenger): the messenger that will communicate between nodes.

  • has_children (bool): True if this node has at least one downstream child in the graph — used to skip publishing when nothing would ever consume it.

property computation_node: Node

Returns the current computation node

run() → None

Starts the task in an infinite loop. If this method is called and the set_messenger() method has not been called yet, an assertion error will happen.

class videoflow.core.task.ProcessorTask(processor: ProcessorNode, messenger: Messenger, has_children: bool, parent_names: List[str], ctx: RuntimeContext | None = None)

Bases: NodeTask

It runs forever, first blocking until it receives a message from every parent node through the messenger. Then it passes the merged inputs to the processor node and, when it gets back the output, uses the messenger to publish it down the flow. If every parent has signaled termination, it passes termination message down the flow and breaks from infinite loop.

change_device(device_type: str) → None

property device_type: str

class videoflow.core.task.ProducerTask(producer: ProducerNode, messenger: Messenger, has_children: bool, ctx: RuntimeContext | None = None)

Bases: NodeTask

It runs forever calling the next() method in the producer node. At each iteration it checks for a termination signal, and if so it sends a termination message to its child task and breaks the infinite loop.

class videoflow.core.task.Task

Bases: object

run() → None

Starts the task in an infinite loop.