Metadata-Version: 2.4
Name: vcti-dataflow
Version: 3.0.0
Summary: vcti-flow bindings for array data: a DataNode payload binding and a FieldSet table binding (sources, transformers, reducers, sinks).
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: vcti-fieldset[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

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

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

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

### In `pyproject.toml` dependencies

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

---

## Quick Start — DataNode binding

```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 (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.

```python
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](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](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
- [vcti-fieldset](https://github.com/vcollab/vcti-python-fieldset) `[datanode]` (>=2.0.0) — the `FieldSet` table payload and its copy-free reshaping
- [numpy](https://numpy.org/) (>=1.24)
