vcti-dataflow 3.0.0

vcti-flow bindings for array data: a DataNode payload binding and a FieldSet table binding (sources, transformers, reducers, sinks).

Data Flow

vcti-flow bindings for array data: a DataNode payload binding and a FieldSet table binding — sources, transformers, reducers, and sinks.

Overview

vcti-flow is a payload-agnostic framework for composing flow graphs — it never inspects the values flowing through it. This package binds that framework to two concrete payloads so you subclass ready-bound node kinds instead of writing Source[...] everywhere:

• vcti.flow.data — the vcti-datanode DataNode binding: one array plus layered attributes behind a data source. Node kinds Source / Transformer / Reducer / Sink / Observer, plus from_array (eager) and ArraySource (lazy) payload builders.
• vcti.flow.fieldset — the vcti-fieldset FieldSet binding: a table of named columns. This is where column and row reshaping lives — select / drop / rename / cast / compute, filter / sort / slice, and the N→1 combiners merge / concat. The nodes are thin wrappers over FieldSet's own copy-free project() / rows() builders.

Use the DataNode binding for single-array flows; use the FieldSet binding to reshape multi-column tables. DataNodesSource / WriteDataNodes bridge between them (a group of DataNodes ⇆ a FieldSet).

Installation

pip install vcti-dataflow

In pyproject.toml dependencies

dependencies = [
    "vcti-dataflow>=3.0.0",
]

---

Quick Start — DataNode binding

import numpy as np
from vcti.flow.data import Source, Transformer, DataNode, from_array

# A source produces a DataNode
class Stress(Source):
    def load(self) -> DataNode:
        return from_array(np.array([1.0, 2.0, 3.0]), {"units": "MPa"})

# A transformer maps one DataNode to another
class Scale(Transformer):
    def __init__(self, factor: float) -> None:
        super().__init__()
        self.factor = factor

    def transform(self, record: DataNode) -> DataNode:
        return from_array(record.load() * self.factor, record.attributes)

result = Scale(2.0).connect(Stress()).execute()
result.load()               # array([2., 4., 6.])
result.attributes["units"]  # "MPa"

Reach for the array only when you need it (record.load()); a leaf source can hand back a LazyDataSource-backed node (or use ArraySource) to defer a heavy read.

Quick Start — FieldSet binding (tables)

The vcti.flow.fieldset nodes carry a FieldSet (named columns) and reshape it. Column ops (select / drop / rename / cast) share columns by reference — no data copy; cast is a lazy per-column cast. Row ops fold to a single index.

import numpy as np
from vcti.datanode import DataNode, EagerDataSource
from vcti.flow.fieldset import (
    DataNodesSource, SelectFields, RenameFields, CastFields, ComputeFields,
    FilterRows, SortRows, MergeFields, ConcatRows, WriteDataNodes,
)

# Read a group of DataNodes as one FieldSet (lazy — nothing materialises here)
src = DataNodesSource([
    DataNode(name="id",     data_source=EagerDataSource(np.array([1, 2, 3]))),
    DataNode(name="stress", data_source=EagerDataSource(np.array([10.0, 20.0, 30.0]))),
])

# Reshape: rename, add a computed column, filter and sort rows
flow = RenameFields({"stress": "s"}).connect(src)
flow = ComputeFields("s_kpa = s * 1000").connect(flow)
flow = FilterRows("s > 10").connect(flow)
flow = SortRows("s", descending=True).connect(flow)

result = flow.execute()                 # a FieldSet
result.get_values("s_kpa")              # array([30000., 20000.])

# Combine tables — N→1 reducers own the policy (collision / schema)
wide = MergeFields().connect(flowA).connect(flowB)          # union columns
tall = ConcatRows().connect(flowA).connect(flowB)           # stack rows

# Write out: FieldSet -> DataNodes -> your writer (I/O stays in your app)
WriteDataNodes(lambda node: write_to_h5(node)).connect(flow).execute()

Column reshaping is project()-backed; row reshaping is rows()-backed. See docs/patterns.md for the full recipe set.

---

API

vcti.flow.data — DataNode binding

Symbol	Purpose
Source / Transformer / Reducer / Sink / Observer	vcti.flow kinds bound to DataNode
from_array(array, attributes=None)	Build a DataNode from an in-memory array (eager)
ArraySource(load_fn, attributes=None)	Lazy leaf source over a callable returning an array
DataNode / EagerDataSource / LazyDataSource	Re-exported from vcti-datanode

vcti.flow.fieldset — FieldSet binding

Symbol	Purpose
Source / Transformer / Reducer / Sink / Observer	vcti.flow kinds bound to FieldSet
SelectFields / DropFields / RenameFields	Keep / remove / rename columns (share sources, no copy)
CastFields / ComputeFields / FreezeFields	Cast dtypes (lazy), add expression columns, bake an expression to data
FilterRows / SortRows / HeadRows / SliceRows	Row selection (fold to one index; slice stays a view)
MergeFields / ConcatRows	N→1: union columns / stack rows (own collision & schema policy)
FieldSetSource	Leaf source over an existing FieldSet (or a callable returning one)
DataNodesSource / WriteDataNodes	Boundary: DataNodes → FieldSet / FieldSet → DataNodes
for_each_group(source, key_field, factory)	Fan out into one flow per group
from_datanodes / to_datanodes	Re-exported from vcti-fieldset

---

Dependencies

• vcti-flow (>=2.0.0) — the generic framework
• vcti-datanode (>=2.0.0) — the DataNode payload
• vcti-fieldset [datanode] (>=2.0.0) — the FieldSet table payload and its copy-free reshaping
• numpy (>=1.24)