# Data sources, ops, and pipelines (v3 paradigm)

Status: design doc for the v3 paradigm.

This document captures the locked decisions for the new abstraction layer
shipped under `cfdmod/core/` and `cfdmod/adapters/`. It is the durable
reference for "what shape is the new layer, and why". Implementation
details (file paths, signatures) live in the docstrings of those modules;
this doc is for the contract.

## 1. Goal

Replace the ad-hoc per-recipe orchestration in `cfdmod/pressure/`,
`cfdmod/inflow.py`, `cfdmod/s1/`, and the former HFPI dynamic pipeline
with a single composable paradigm: a small frozen value object
(`DataSource`) plus a library of pure transformations on it.

Phases 0-2 of the plan land the abstractions and prove on-disk parity
with every existing fixture. Phases 3-8 rewrite recipe internals on top
of the new core. The v2 pressure entry points have been removed in v3;
post-processing runs through the v3 recipes and the
`cfdmod run <template.yaml>` CLI. Geometry, S1, inflow, climate,
analytical, and IO helpers remain exported unchanged.

## 2. Data source

A `DataSource` is a frozen Pydantic value object. It holds everything
needed to describe one slab of simulation output:

- `kind`: one of `surface`, `volume`, `points`, `groups`, `modes`.
- `time`: an affine `TimeAxis(initial_time, timestep_size, n_timesteps)`.
  The full time array is *never* stored; it is reconstructed on demand
  from the three numbers. Time-aggregated outputs use `n_timesteps == 0`.
- `topology`: `Topology(cell_type, connectivity, vertices)` for surfaces
  and (later) volumes. `cell_type` is a discriminator (`triangle`,
  `point`, `cell`) so volume export is additive, not a rewrite. Points
  data sources use `cell_type="point"` and an empty connectivity array.
  `GroupsDataSource` does not own its own topology; it carries a
  *reference* to a parent surface's topology plus a per-element group
  index, so it cannot diverge from the parent mesh.
- `elements`: `ElementMeta` with positions, areas, volumes, normals as
  applicable. Optional per-element free-form metadata (e.g. station
  name) lives here.
- `groupings`: `dict[str, np.ndarray]` of int arrays of length
  `n_elements`. Each array assigns each element to exactly one group of
  the named grouping (surface name, planar/volumetric selection, S1
  separation, ...).
- `fields`: a `FieldStore`. The `FieldStore` exposes per-field arrays of
  shape `(n_elements, n_timesteps)` or `(n_elements,)` for time-aggregated
  outputs. Whether the arrays live in RAM or in an h5 file is opaque to
  the rest of the core.
- `field_meta`: `dict[str, FieldMeta]` with name, scale, units.
- `attrs`: source-level metadata (free-form, validated at construction).

The five subclasses are thin wrappers that lock the `kind` discriminator
and constrain which `ElementMeta` columns are required. They share the
same fields and methods.

## 3. Time axis is affine, not stored

The proposal is explicit about this: the time axis is reconstructable
from `(initial_time, timestep_size, n_timesteps)`. Time ops (window
selection, translation, rescale) mutate these three numbers. They never
resample. Resampling is a *field op* and goes through `FieldStore`.

## 4. Algebra: four broadcasting rules

All field algebra (Cp = `(p - p_ref) * scaling`, S1 = `profile / ref`,
etc.) goes through one module, `cfdmod/core/algebra.py`, with four
broadcasting rules dispatched on the shape pattern:

1. `[multi or single] * constant` -> uniform scaling.
2. `[multi] * [single]` with the same time axis -> column-wise (e.g.
   `p - p_ref`).
3. `[multi] * [multi]` with the same shape but only one carries
   timesteps -> row-wise (e.g. S1).
4. `[multi] * [multi]` with the same shape -> element-wise.

Recipes never reimplement broadcasting. Any new recipe (Cf, S1,
pedestrian comfort) is a `compose(...)` of existing ops + algebra calls.

## 5. Functional core, imperative shell

- Ops are pure functions: `op(ds, params) -> DataSource`. No methods on
  `DataSource` mutate state or do I/O.
- The shell (CLI, recipe runners) wires concrete adapters in via a
  frozen `Context`.
- Pipelines are `compose(*ops)` -> `Callable[[DataSource], DataSource]`,
  built once at recipe-construction time, applied at run time.

## 6. The injection seam: `FieldStore` and `Storage`

`FieldStore` is the only place the small-data (numpy in RAM) vs
large-data (XDMF + h5, 10+ GB) distinction lives. Every op calls
`fields.read(name, time_slice=..., element_slice=...)` and trusts the
adapter.

`Storage` handles whole `DataSource` round-trips: topology, time axis,
elements, groupings, plus the `FieldStore` for the data.

Two adapters land in Phase 1:

- `MemoryFieldStore` / `MemoryStorage` (in-process, dict-keyed). Used by
  every test and by notebooks. ~50 lines each.
- `H5FieldStore` / `XdmfH5Storage`. Wraps the existing
  `cfdmod.io.xdmf` writers. Preserves the on-disk layout exactly:
  `/Triangles`, `/Geometry`, `/meta/{time_steps,time_normalized,
  region_labels}`, `/{group}/t{T}`. No format change.

The `core` package depends on `numpy` and `pydantic` only; `h5py` lives
exclusively under `adapters/xdmf_h5/`.

## 7. Containers

`Container[K, V]` is a directional / parametric multi-case map (hashable
Pydantic key, `join_by(callback)`, `filter_by`, `map_values`). The
building dynamic-response cases (`cfdmod.dynamics.cases`, keyed by
`BuildingCaseParameters`) group and filter directional results through it
with no bespoke machinery.

Parallelism is *injected*, not built in: `Container.map_values(p,
pool=ctx.pool)` runs the pipeline over the container in parallel only
if a pool is supplied. Sequential is the default.

## 8. Inside vs outside the paradigm

Inside (becomes data sources + ops): `pressure/` (Cp, Cf, Cm, Ce),
`pressure/filters.py`, `inflow.py`, `s1/profile.py`, the former HFPI
dynamic pipeline (now `core/recipes/dynamic.py` + `cfdmod/dynamics/`),
the existing `analysis/inflow/`, `io/xdmf.py`, `io/timeseries.py`.
Pedestrian comfort joins later (it composes existing primitives + a
climate-data input).

Outside (stay standalone): `loft/`, `roughness/`, `snapshot/`,
`altimetry/`, `analytical/`, `climate/`, plotting helpers,
`io/geometry/`, `io/vtk/`. They produce inputs to or consume outputs
from data sources but never participate as filters in a pipeline.

## 9. Migration outcome

The pressure post-processing surface was cut over to the new core rather
than kept as parallel wrappers. The `cfdmod.pressure` package and its
disk-first entry points (`run_cp` / `run_cf` / `run_cm` / `run_ce` /
`apply_filters` / `MovingAverageFilter`) were removed. The supported
access paths are the v3 recipes (`cfdmod.recipes`), the op catalog
(`cfdmod.core_ops`), and version-3 YAML templates executed with
`cfdmod run <template.yaml>`. Non-pressure public helpers (geometry, S1,
inflow, climate, analytical, IO) remain exported from `cfdmod`.

## 10. What is NOT in scope

- CPU vs GPU: out of scope. If it ever matters, it is another adapter.
- `ArrayLike` protocol over `numpy.ndarray`: not the kind of swap we
  need; numpy is a hard dependency of the core.
- Abstract `BaseFilter` / `BaseRecipe` classes: composition over
  inheritance. Recipes are concrete `Pipeline`s.
- Climate, altimetry, snapshot, loft, roughness as pipeline stages:
  per the odt, these stay outside the paradigm. Pedestrian comfort
  takes climate data as a non-pipeline input.

## 11. Consuming cfdmod as a service

The v3 core is usable not just as a library you call but as a contract a
service can reflect on and drive from a UI (e.g. a node-based pipeline
editor). Four capabilities support that:

### Dependency-light import

`import cfdmod` -- and importing the template schema / op catalog under
`cfdmod.core` -- pulls none of the heavy scientific stack (h5py,
matplotlib, pandas, pyarrow, vtk, trimesh). Those load lazily, only when
a symbol that needs them is first accessed. A web/API layer can validate
and build templates without the numeric stack installed. The stable,
dependency-light surface is `cfdmod.core.pipeline_yaml`
(`PipelineTemplate`, `validate_template`, `list_ops`, `op_info`).

### Op catalog (introspection)

The op registry is populated at import, so a consumer can enumerate the
op set without first running a template:

```python
from cfdmod import list_ops, op_info

for info in list_ops():
    print(info.kind, info.family, info.arity, info.consumes, info.produces)
    # info.params_schema is the op's parameter JSON Schema (render a form)

op_info("statistics").params_schema   # single op
```

### Custom ops

A consumer registers its own op with `register_op`; it becomes a
first-class citizen -- usable in templates under its `kind`, validated,
and listed by `list_ops`:

```python
from typing import ClassVar, Literal
from cfdmod import DataSource, register_op
from cfdmod.core.ops import OpParams

class ClipParams(OpParams):
    kind: Literal["clip"] = "clip"
    field: str = "cp"
    lo: float = 0.0
    hi: float = 1.0
    op_family: ClassVar[str] = "field"   # or inferred from the module path

def clip(ds: DataSource, p: ClipParams) -> DataSource: ...

register_op("clip", clip, ClipParams, arity="unary")
```

### Static graph validation

`validate_template` runs a symbolic contract pass over the catalog:
each op declares `consumes` / `produces` kinds and
`requires_element_meta`, and `validate_template` checks the wiring
without running anything -- catching e.g. a `force_contribution` before
`mesh_attach`, or a surface-only op fed a points binding. It is strict on
kinds and metadata (deterministic) and permissive on fields when the
input's fields were not declared.

### Typed errors and object storage

`run_template` raises a typed hierarchy under `CfdmodError`
(`TemplateError`, `TemplateReferenceError`, `OpError` with `step_id` /
`op_kind`, `StorageKeyError`); each also subclasses the builtin it
replaced, so existing `except (KeyError, ValueError)` handlers keep
working. For non-filesystem output, `XdmfH5BlobStorage` runs the same
pipeline against any `BlobStore` (an object store such as S3, a DB blob
column) -- cfdmod stays free of a cloud SDK.