Metadata-Version: 2.4
Name: vcti-dataflow
Version: 2.0.0
Summary: The DataNode binding of vcti-flow: sources, transformers, reducers, and combiners for vcti-datanode payloads.
Author: Visual Collaboration Technologies Inc.
Requires-Python: <3.15,>=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: vcti-flow>=2.0.0
Requires-Dist: vcti-datanode>=2.0.0
Requires-Dist: numpy>=1.24
Provides-Extra: test
Requires-Dist: pytest; extra == "test"
Requires-Dist: pytest-cov; extra == "test"
Provides-Extra: lint
Requires-Dist: ruff; extra == "lint"
Provides-Extra: typecheck
Requires-Dist: mypy; extra == "typecheck"
Dynamic: license-file

# Data Flow

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

## Overview

[vcti-flow](https://github.com/vcollab/vcti-python-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](https://github.com/vcollab/vcti-python-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

```bash
pip install vcti-dataflow
```

### In `pyproject.toml` dependencies

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

---

## Quick Start

```python
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:

```python
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](https://github.com/vcollab/vcti-python-flow) (>=2.0.0) — the generic framework
- [vcti-datanode](https://github.com/vcollab/vcti-python-datanode) (>=2.0.0) — the `DataNode` payload
- [numpy](https://numpy.org/) (>=1.24)
