vcti-dataflow 2.0.0

The DataNode binding of vcti-flow: sources, transformers, reducers, and combiners for vcti-datanode payloads.

Data Flow

The DataNode binding of vcti-flow: sources, transformers, reducers, and combiners for vcti-datanode payloads.

Overview

vcti-flow is a payload-agnostic framework for composing flow graphs — it never inspects the values flowing through it. vcti.flow.data binds that framework to the vcti-datanode DataNode payload (data plus layered attributes behind a data source), so you get familiar, ready-bound node kinds — Source, Transformer, Reducer, Sink — instead of writing Source[DataNode] everywhere, plus from_array (eager) and ArraySource (lazy) for building payloads.

These node-kind names are the vcti.flow kinds bound to DataNode: in this package Source is Source[DataNode]. It is a flow leaf node — distinct from vcti.datanode.DataSource, the array-source ABC re-exported here as EagerDataSource / LazyDataSource.

Nodes that need a structured (named-field) array — field-wise merge and row-keyed iteration — live in the vcti.flow.data.fields submodule; the base binding makes no assumption about array shape.

Installation

pip install vcti-dataflow

In pyproject.toml dependencies

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

---

Quick Start

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 to defer a heavy read.

Structured-array nodes

The vcti.flow.data.fields submodule adds nodes that assume a structured (named-field) array — building tables, naming/selecting/computing fields, merging field groups, and row-keyed iteration:

from vcti.flow.data import ArraySource          # lazy leaf source (base binding)
from vcti.flow.data.fields import (
    RowTableSource, NameFields, SelectFields, ComputeFields,
    RenameFields, DropFields, CastFields, MergeFields, for_each_field,
)

# Build a structured table from dict rows (lazy — rows read on load())
mats = RowTableSource(lambda: material_rows(reader),
                      columns={"id": "MAT_ID", "EX": "Young's Modulus"})

# Name plain columns, select/rename, compute
coords = NameFields(["X", "Y", "Z"]).connect(ArraySource(lambda: reader.coords()))
picked = SelectFields({"X": "x", "Y": "y"}).connect(coords)
mag = ComputeFields({"mag": lambda a: np.hypot(a["X"], a["Y"])}).connect(coords)

# Rename, drop, cast (rename & drop return views — no data copy)
renamed = RenameFields({"X": "x"}).connect(coords)   # rename, keep the rest
trimmed = DropFields(["Z"]).connect(coords)          # drop, keep the rest
narrow = CastFields({"X": "f4"}).connect(coords)     # change dtypes

# Merge field groups (same row count) into one structured array
combined = MergeFields().connect(ids).connect(coords).execute()

# One flow per row, keyed by a field
for case_id, flow in for_each_field(cases, build_case_flow, key_field="ID"):
    flow.execute()

---

API

Symbol	Purpose
Source / Transformer / Reducer / Sink	vcti.flow node kinds bound to DataNode
Observer	vcti.flow observer 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 (the lazy counterpart of from_array)
DataNode / EagerDataSource / LazyDataSource	Re-exported from vcti-datanode for convenience
fields.RowTableSource	Lazy leaf source — a structured table from dict rows
fields.NameFields / fields.SelectFields / fields.ComputeFields	Name plain columns, select/rename fields, append/replace computed fields
fields.RenameFields / fields.DropFields / fields.CastFields	Rename or drop fields (views, no copy), or change field dtypes
fields.MergeFields	Field-wise merge of structured arrays
fields.for_each_field / fields.field_items	Row-keyed fan-out over a structured array

---

Dependencies

• vcti-flow (>=2.0.0) — the generic framework
• vcti-datanode (>=2.0.0) — the DataNode payload
• numpy (>=1.24)