Metadata-Version: 2.4
Name: recordstream
Version: 0.1.0a2
Summary: Clean, functional data pipelines for ML research and production.
Author: Gert Behiels
License-Expression: MIT
Project-URL: Homepage, https://github.com/Gearlux/recordstream
Project-URL: Repository, https://github.com/Gearlux/recordstream.git
Project-URL: Issues, https://github.com/Gearlux/recordstream/issues
Project-URL: Changelog, https://github.com/Gearlux/recordstream/blob/main/CHANGELOG.md
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: Intended Audience :: Science/Research
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.12
Classifier: Programming Language :: Python :: 3.13
Classifier: Topic :: Scientific/Engineering :: Artificial Intelligence
Classifier: Topic :: Software Development :: Libraries :: Python Modules
Classifier: Typing :: Typed
Requires-Python: >=3.12
Description-Content-Type: text/markdown
License-File: LICENSE
Requires-Dist: confluid>=0.3.0
Requires-Dist: loggair>=0.2.0
Requires-Dist: numpy
Requires-Dist: Pillow
Requires-Dist: h5py
Requires-Dist: zarr
Requires-Dist: typing-extensions
Requires-Dist: datasets
Requires-Dist: albumentations
Requires-Dist: orjson
Requires-Dist: fsspec
Requires-Dist: cloudpathlib
Requires-Dist: liquifai>=0.2.0
Provides-Extra: torch
Requires-Dist: torch; extra == "torch"
Provides-Extra: keras
Requires-Dist: keras>=3.0; extra == "keras"
Provides-Extra: dev
Requires-Dist: build>=1.0.0; extra == "dev"
Requires-Dist: twine>=5.0.0; extra == "dev"
Requires-Dist: black<25.0.0,>=24.0.0; extra == "dev"
Requires-Dist: isort<6.0.0,>=5.13.0; extra == "dev"
Requires-Dist: flake8<8.0.0,>=6.0.0; extra == "dev"
Requires-Dist: flake8-junit-report>=2.1.0; extra == "dev"
Requires-Dist: mypy<2.0.0,>=1.0.0; extra == "dev"
Requires-Dist: pytest<9.0.0,>=7.0.0; extra == "dev"
Requires-Dist: pytest-cov<7.0.0,>=4.0.0; extra == "dev"
Requires-Dist: torchvision; extra == "dev"
Requires-Dist: matplotlib; extra == "dev"
Provides-Extra: notebook
Requires-Dist: matplotlib; extra == "notebook"
Requires-Dist: datasets; extra == "notebook"
Requires-Dist: python-dotenv; extra == "notebook"
Provides-Extra: vision
Requires-Dist: scipy; extra == "vision"
Requires-Dist: torchvision; extra == "vision"
Dynamic: license-file

# RecordStream

**RecordStream** is a high-performance, functional data processing engine built for modern Machine Learning pipelines. It provides a clean, fluent API for streaming and transforming data from any source while maintaining strict compatibility with PyTorch and Hugging Face.

Part of the **Modular Quartet**: `Loggair`, `Confluid`, `Liquifai`, and `RecordStream`.

## 🚀 Key Features

-   **A record is a plain dict:** the [record model](https://github.com/Gearlux/recordstream/blob/main/docs/record-model.md) — a `dict` of typed values (`Image`, `Mask`, `Boxes`, `Label`, `MultiLabel`, …), each owning its own metadata, with key names carrying meaning (`"image"`, `"mask"`, `"bboxes"`). No wrapper container, no role tags.
-   **Libraries run AS-IS:** bare [albumentations and torchvision `transforms.v2`](https://github.com/Gearlux/recordstream/blob/main/docs/augmentation.md) transforms drop straight into any ops list — the engine invokes each op family natively (one call = one joint draw across image/mask/boxes). No adapter classes anywhere.
-   **Type-dispatched native ops:** a `Transform` samples its parameters once per record and applies a per-type kernel to every value it handles — teach an existing op a new value type with one `@MyOp.kernel(NewType)` registration.
-   **Graph pipelines:** readable [`flow:` documents](https://github.com/Gearlux/recordstream/blob/main/docs/graph.md) of named steps — `from:` forks, `merge_from:` merges, `bind:` feeds one step's value into another's parameter. An `ops:` list is the same engine's linear spelling; both parse to one step graph.
-   **High Performance:** Native multiprocess support via `.parallel(workers=N)` using the safe `spawn` context; [1→N expanding ops](https://github.com/Gearlux/recordstream/blob/main/docs/kinds.md#1n-expanding-ops-iterable-only-pipelines) flatten in every route.
-   **Advanced Storage:** HDF5, Zarr and Directory backends with matching read-back sources and [metadata-only querying](https://github.com/Gearlux/recordstream/blob/main/docs/storage.md#queryable-metadata-recordstreamstoragequery) — filter stored datasets without loading a single array.
-   **Passive Introspection:** ops declare the value types they [handle / consume / produce](https://github.com/Gearlux/recordstream/blob/main/docs/record-model.md) and are discoverable by category for visual editors and schema generators.
-   **Algorithms:** declare what an op is tuned by, reads and computes (`Param` / `Input` / `Output`) and write `compute()`. The [record handling, the constructor, the settings schema and the chain contract](https://github.com/Gearlux/recordstream/blob/main/docs/algorithm.md) are derived from those declarations.
-   **100% Reproducibility:** Entire pipelines are serializable via **Confluid** manifests.

## 🛠 Quick Start

One pipeline mixing a **bare albumentations Compose** (image + mask + boxes move together in one draw), a **bare torchvision v2 transform**, and a **native op** — no wrappers (mirrors [`examples/record_pipeline.py`](https://github.com/Gearlux/recordstream/blob/main/examples/record_pipeline.py)):

```python
import albumentations as A
import numpy as np
from recordstream import Stream, Image, Label, Mask, as_transform

records = [
    {
        "image": Image(rng.random((16, 20, 3)).astype(np.float32)),   # typed: knows its layout
        "mask": Mask((rng.random((16, 20)) > 0.5).astype(np.uint8)),
        "bboxes": [[2, 3, 6, 7]],                                     # albumentations vocabulary
        "labels": ["drone"],
        "class": Label("drone_x", classes=["noise", "drone_x"]),      # typed: knows its vocab
        "gain_db": -3.0,                                              # a scalar is just another key
    }
    for rng in (np.random.default_rng(i) for i in range(100))
]

stream = Stream(
    source=records,
    ops=[
        A.Compose(                                    # bare albumentations — as-is
            [A.HorizontalFlip(p=0.5)],
            bbox_params=A.BboxParams(format="pascal_voc", label_fields=["labels"]),
        ),
        A.GaussNoise(p=1.0),                          # image only (its own kwarg vocabulary)
        as_transform(lambda d: d - 0.5, handles=(Image,)),   # native: a plain function op
    ],
).parallel(workers=4)

for record in stream:
    print(record["image"].shape, record["class"].value)   # image+mask+boxes flipped together
```

The same ops list in Confluid YAML — bare library transforms are ordinary `!class:` nodes:

```yaml
ops:
  - !class:albumentations.HorizontalFlip
    p: 0.5
  - !class:albumentations.GaussNoise
    p: 1.0
  - !class:recordstream.ops.numpy.Threshold
    low_level: 0.5
```

### Toggling a branch from the CLI (`Enable`)

Wrap any stretch of an ops list in `Enable` to switch the whole chain on or off from one flag.
The toggle is the declared `enabled` parameter; `name` identifies the wrapper so several of them
toggle independently:

```yaml
ops:
  - !class:recordstream.ops.numpy.Threshold {low_level: 0.5}
  - !class:recordstream.ops.enable.Enable
    name: visualize          # ← names THIS wrapper; scopes its CLI flag
    enabled: false           # ← off by default; the chain below is skipped
    ops:
      - !class:recordstream.ops.image.ConvertToImage {}
      - !class:recordstream.ops.debug.PrintRecordOp {}
```

```bash
recordstream run pipeline.yaml --visualize.enabled true   # this wrapper only
recordstream run pipeline.yaml --visualize.enabled+       # polarity shorthand → True
recordstream run pipeline.yaml --enabled false            # broadcast: every Enable off
```

Inner ops are not materialized until the wrapper first fires, so gating an expensive chain with
`enabled: false` costs nothing at startup. In Python the same wrapper is one call —
`Enable(ops=[...], name="visualize", enabled=False)` — which is what lets a visual editor or a
generated tool schema set the toggle too (see [docs/architecture.md](https://github.com/Gearlux/recordstream/blob/main/docs/architecture.md#6-every-knob-is-a-declared-parameter--the-enable-toggle-2026-07-27)).

### Inference as an op (`ModelPredict`)

A pipeline can carry its own inference: `recordstream.ops.predict.ModelPredict` runs any
callable model wrapper on each record and stamps the prediction back as a record field —
a class `Label`, `Boxes`, an int class mask, or the restored image, by `kind`. The model's
heavy work (build the network, load `checkpoint_path`) happens in its `solidify()`, called
lazily on the first record; the op itself imports no ML framework.

```yaml
pipeline: !class:recordstream.core.stream.Stream
  source: !class:recordstream.sources.huggingface.HuggingFaceSource {path: ylecun/mnist, split: test}
  ops:
    - !class:recordstream.ops.image.ConvertToImage {width: 224, height: 224}
    - !class:recordstream.ops.predict.ModelPredict
        model: !class:<your model wrapper> {checkpoint_path: runs/checkpoints/mnist/last.ckpt}
        kind: classification        # or detection / segmentation / restoration
```

A viewer reads the stamped `predict*` fields back as layers; `recordstream run` executes
the same document offline.

### Stating a pipeline's interface (`RecordContract`)

`recordstream.ops.contract.RecordContract` is a pass-through op that asserts what each
record carries at the point in the chain where it sits — an executable, visible interface
statement. One class serves both boundary roles, decided by position: the **first** op in
a chain states what the host must feed (input contract), the **last** states what the
pipeline guarantees to deliver (output contract). `fields` maps a record key to a
registered item type name (`"*"` = present with any type, for boundaries past a
conversion that emits plain values); `name` labels the boundary in errors:

```yaml
pipeline: !class:recordstream.core.stream.Stream
  source: !class:recordstream.sources.huggingface.HuggingFaceSource {path: ylecun/mnist, split: test}
  ops:
    - !class:recordstream.ops.contract.RecordContract
        name: classification output
        fields: {image: Image, class: Label}
```

A violating record fails loudly at the boundary — naming the contract, the record
ordinal, the offending entry, and what the record does carry — instead of surfacing
later as an empty result:

```
ContractError: classification output: record #0 has no entry 'class' (expected Label); present: image[Image]
```

A consuming workspace or visual editor seeds the contract into a graph so the required
record shape is declared before the first node is wired; the exported document then
enforces the same contract when it runs offline.

### Checking that a chain holds together (`check_chain`)

`RecordContract` answers *"what must reach this point?"*. A chain of many small ops raises a
different question — *"does this chain hold together at all?"* — and getting it wrong is **silent**:
an op that reads an entry an earlier op was supposed to write does not raise, it returns the record
unchanged, so the pipeline runs to completion and answers nothing.

So an op MAY declare its interface as class attributes, and `check_chain` reads them off the op list
once, before the first record: A visual editor may draw a gate as a wire from the producer's outcome and write `requires` back into the document on export — the document form stays the one shown here.

```python
class MeasureSymbolClock:
    consumes = {"signal": "Signal", "inst_freq": "InstFreq"}   # same {key: item type} vocabulary
    produces = {"symbol_clock": "SymbolClock"}                 # as RecordContract.fields
    reports  = "clock"        # its key in an analysis report; "" marks a transform
    flags    = ()             # the boolean findings it raises

class DecodeBleAdvertising:
    def __init__(self, requires: str = "") -> None:
        self.requires = requires        # the ONE flag that gates this op
```

```python
from recordstream.ops.contract import check_chain

check_chain(ops, provided={"signal"}, where="view_bte.yaml")
# ChainContractError: view_bte.yaml: MeasureSymbolClock needs the record entry 'inst_freq',
# which nothing before it produces — MeasureInstantaneousFrequency produces it, but LATER in
# the chain — move it before MeasureSymbolClock
```

Four things are refused: an unmet `consumes`, a `requires` naming a flag nothing raises (or one
raised only later), two ops declaring one flag, and two ops reporting under one name.

Declaring is **opt-in** — an op with none of these attributes is checked for nothing, so existing
chains are unaffected. `flag_producers(ops)` gives `{flag: index of the op that raises it}`, total by
construction, so *"which op decided this branch?"* always has an answer.

### Writing an op as an algorithm (`Algorithm`)

An algorithm declares its settings, inputs and outputs, and never sees a record:

```python
import numpy as np
from recordstream import Algorithm, Image, Input, Output, Param

class BackgroundLevel(Algorithm):
    percentile: float = Param(default=25.0, doc="Percentile rank across the per-row medians.")
    image: Image = Input(doc="The image to read.")
    background: float = Output(doc="The background level.")

    def compute(self):
        per_row = np.median(np.asarray(self.image, dtype=np.float64), axis=1)
        return {"background": float(np.percentile(per_row, self.percentile))}

image = Image(np.full((4, 6), 10.0), layout="HWC")
BackgroundLevel(percentile=30.0).run(image=image)                 # {'background': 10.0}
BackgroundLevel()({"image": image, "exposure_ms": 20})            # the record + 'background'
BackgroundLevel(keys={"image": "photo"})({"photo": image})        # read from another entry
BackgroundLevel().consumes                                        # {'image': 'Image'} — for check_chain
```

Used as an op, it reads each input from the record entry of the same name and writes each output
the same way; `keys` names other entries, and `Output(replaces="boxes")` writes back where an input
was read. Inputs are found by name only, so `check_chain` knows before a run what a pipeline needs.
Full guide: [docs/algorithm.md](https://github.com/Gearlux/recordstream/blob/main/docs/algorithm.md).

### Declaring the class vocabulary (`ClassNamesOutput`, `ClassNamesScan`)

`RecordContract` states what each RECORD carries. A classification pipeline usually has
something to say about the DATASET too — the ordered class list a consumer needs to show a
label as a name rather than an integer. `ClassNamesOutput` is that statement. It holds one
thing, the list, and derives nothing:

```yaml
class_names_output: !class:recordstream.ops.contract.ClassNamesOutput
  names: [cat, dog]                           # in class-id order
```

The list gets there one of three ways, each chosen explicitly:

| route | how |
|---|---|
| state them | type `names: [cat, dog]` |
| take them from a source that knows them | wire the source's `class_names` output into `names` — `HuggingFaceSource` reads its `ClassLabel` names from the dataset metadata, no rows read |
| take them from a walker | wire a `ClassNamesScan`'s `class_names` output into `names` |

`ClassNamesScan` derives a vocabulary by walking a source's label column — for a source whose
format does not describe one (a folder reader, a CSV):

```python
from recordstream import Label
from recordstream.ops.contract import ClassNamesOutput, ClassNamesScan

records = [{"class": Label(value="dog")}, {"class": Label(value="cat")}, {"class": Label(value="dog")}]
output = ClassNamesOutput(names=ClassNamesScan(source=records).class_names)
output.class_names  # ['cat', 'dog']
```

The walk goes through [key projection](https://github.com/Gearlux/recordstream/blob/main/docs/projection.md)
so a projection-aware source is asked
only for the label column, and it reports sorted-unique stringified values — the same ordering
`LabelMap.fit` uses. It is deliberately a **separate class**: a walk costs seconds per thousand
records, so it happens because someone asked for it, never as a silent fallback inside
`ClassNamesOutput`.

**In a document, a wire is a literal.** A config document cannot reference one object's
attribute from another — `names: !ref:scan.class_names` is refused when the document loads — so
in a document the vocabulary is always the list itself. A visual editor that lets you draw the
wire evaluates it when it saves the graph and writes the resulting list into `names`.

## 📚 Documentation

| Page | Covers |
|---|---|
| [docs/record-model.md](https://github.com/Gearlux/recordstream/blob/main/docs/record-model.md) | The record data model: a plain dict of typed values, type-dispatched ops and kernels, mixing libraries as-is, custom item types, engines, storage layout |
| [docs/algorithm.md](https://github.com/Gearlux/recordstream/blob/main/docs/algorithm.md) | Writing an op as an `Algorithm`: `Param` / `Input` / `Output` slots, `compute()`, entry names (`keys`), replacing outputs, what tools derive (`algorithm_spec`, `consumes` / `produces`, the settings schema) |
| [docs/kinds.md](https://github.com/Gearlux/recordstream/blob/main/docs/kinds.md) | Writing ops (kernels, `field=`, type-changing ops), the collate registry (`collate_records`) + its read-back (`batch_values` / `batch_tensor` / `batch_metadata`), the Keras `RecordSequence` adapter, 1→N expanding ops |
| [docs/graph.md](https://github.com/Gearlux/recordstream/blob/main/docs/graph.md) | `flow:` documents + the `FlowGraph` engine, `ops:` as the linear spelling of the same step graph, expanding (1→N) steps, `Stream.from_ops_yaml` |
| [docs/sources.md](https://github.com/Gearlux/recordstream/blob/main/docs/sources.md) | `HuggingFaceSource`, `FilesSource`, `DatasetSplit` train/val/test views, `RangeSource`, `ConcatSource`, Confluid `!ref:` sharing, dataset identity (`dataset_uri` / `dataset_url`) |
| [docs/storage.md](https://github.com/Gearlux/recordstream/blob/main/docs/storage.md) | HDF5 / Zarr / Directory sinks & sources (`typedrecord-v1`), array-valued item attributes, the `SupportsMetadataScan` protocol + `MetadataFilterSource` querying |
| [docs/projection.md](https://github.com/Gearlux/recordstream/blob/main/docs/projection.md) | Key projection (`SupportsProjection`), lazy key walks (`iter_key`), one-peek `first_value`, `num_classes`, the fittable `LabelMap`, class-balance weights |
| [docs/predictions.md](https://github.com/Gearlux/recordstream/blob/main/docs/predictions.md) | The model boundary: prediction-output contracts (`ClassificationOutput` & co), `ensure_record_dataset`, the `PredictionsSink` protocol + the classification sink |
| [docs/image.md](https://github.com/Gearlux/recordstream/blob/main/docs/image.md) | Generic value→image conversion (`ConvertToImage`, `normalize_to_uint8`), mask→class-id conversion (`ConvertToMask`), channel layout / value range / element type (`ConvertMode`, `Scale`, `ToType`), array introspection helpers |
| [docs/configure.md](https://github.com/Gearlux/recordstream/blob/main/docs/configure.md) | Per-record op parameters (`ConfigureOp` and the `Capture`/`Apply` context ops) |
| [docs/runnable.md](https://github.com/Gearlux/recordstream/blob/main/docs/runnable.md) | Runnables (`run()` + `recordstream run`), the `@entrypoint` task/role markers + `run_entrypoint` dispatch with a worked example, `TorchRunner` / `ProgressReporting` |
| [docs/workflow.md](https://github.com/Gearlux/recordstream/blob/main/docs/workflow.md) | Workflow combinators (`Sequence`/`Conditional`/`Switch` + predicates): resume-safe multi-stage pipelines as ONE document |
| [docs/augmentation.md](https://github.com/Gearlux/recordstream/blob/main/docs/augmentation.md) | Augmentation via bare albumentations / torchvision `transforms.v2` — the op-family dispatch, key vocabulary, bbox recipes, seeding |
| [docs/architecture.md](https://github.com/Gearlux/recordstream/blob/main/docs/architecture.md) | Architecture decision records — the *why* behind non-obvious mechanisms (e.g. why collation is a pluggable registry) |

## 🧭 Scope: a modality-neutral engine

RecordStream deliberately contains **no domain-specific code** — every op, source and sink in this package is meaningful for any modality (arrays, tensors, images, generic metadata). Domain packages build on it and keep their own vocabulary:

- Signal/waveform items and ops (spectrograms, FFT windows, recording formats) live in the domain package, which registers its item types into the same registries.
- Task-specific trainers, collates and models live in their consuming projects.

## 🌐 Ecosystem Integration

RecordStream is designed to sit between your data catalog and your training loop, acting as the high-performance "glue" for ML pipelines:

- **Hugging Face** for community datasets and Arrow/Parquet loading — `HuggingFaceSource` turns a `datasets.Dataset` into record dicts of typed values with full metadata traceability, and [names the dataset it reads](https://github.com/Gearlux/recordstream/blob/main/docs/sources.md#identifying-a-dataset) so a run record can point at it (see [docs/sources.md](https://github.com/Gearlux/recordstream/blob/main/docs/sources.md)).
- **Confluid** for configuration: every pipeline is a YAML document, every op a `!class:` node — including bare library transforms — every run reproducible.
- **PyTorch**: `Stream` and `FlowGraph` implement the `Dataset` protocol (`__len__`/`__getitem__`/`.batch`/`.parallel`) and plug straight into a `DataLoader` with a [registry collate](https://github.com/Gearlux/recordstream/blob/main/docs/kinds.md#batching--collate_records--the-collate-registry-recordstreamcollate) (`collate_records` is the default).
- **Keras 3**: no `DataLoader` exists to do the batching, so [`RecordSequence`](https://github.com/Gearlux/recordstream/blob/main/docs/kinds.md#keras-recordsequence--the-batching-half-the-framework-leaves-to-you) is the `keras.utils.PyDataset` half — row order, slicing, per-epoch reshuffle, `collate_records` — and a `transform` callable supplies the batch shape, exactly as `collate_fn` does for torch.
- **Augmentation libraries**: [albumentations](https://albumentations.ai) and torchvision `transforms.v2` transforms run **as-is** in any ops list — the engine speaks each library's native convention (kwarg vocabulary vs dict walk), so there is nothing to wrap (see [docs/augmentation.md](https://github.com/Gearlux/recordstream/blob/main/docs/augmentation.md)).

## 🔧 Installation

RecordStream is on PyPI as a pre-release, so `pip` needs `--pre` to see it:

```bash
pip install --pre recordstream
```

The core engine is **numpy**, and installs no ML framework. A framework arrives only with the extra
that needs it:

| Extra | Provides |
|---|---|
| `torch` | The pieces that genuinely produce tensors — the `ToTensor` op, `batch_tensor`, and the `classification_output` / `segmentation_output` builders |
| `keras` | `recordstream.keras` — the `RecordSequence` `PyDataset` adapter and the `KERAS_BACKEND` ordering. Keras 3 is an API, so this names no compute engine; it runs on whichever of torch / TensorFlow / JAX you have |

```bash
pip install --pre "recordstream[torch]"
```

Everything else works without either. A `Stream` is map-style (`__len__`/`__getitem__`), so a
`DataLoader` still accepts one directly on a torch install; `batch_values`, `multi_hot` and the
class-balance statistics return numpy, so a non-torch backend converts in one line. Reaching for
`recordstream.ops.ToTensor` without the extra raises an `ImportError` naming it.

## 📄 License

MIT
