diff --git a/packages/essreduce/docs/developer/adr/0001-minimal-workflow-spec.md b/packages/essreduce/docs/developer/adr/0001-minimal-workflow-spec.md new file mode 100644 index 000000000..2081a52e0 --- /dev/null +++ b/packages/essreduce/docs/developer/adr/0001-minimal-workflow-spec.md @@ -0,0 +1,250 @@ +# ADR 0001: Minimal implementation-independent workflow specifications + +- Status: proposed +- Deciders: Simon +- Date: 2026-09-14 + +## Context + +Three mechanisms currently describe workflow interfaces to users, with +overlapping purpose and no shared shape: + +- `ess.reduce.parameter` / `ess.reduce.workflow`: per-sciline-key `Parameter` + dataclasses in a global registry, with parameters discovered by walking the + pipeline graph from selected outputs. Drives the ipywidgets GUI. +- `ess.livedata.config.workflow_spec.WorkflowSpec`: one pydantic params model + and one pydantic outputs model per workflow, driving the live-data dashboard. +- `ess.nmx.configurations`: standalone pydantic models for batch reduction. + +Consolidation was analyzed at length in +[scipp/ess#653](https://github.com/scipp/ess/issues/653) and +[scipp/esslivedata#889](https://github.com/scipp/esslivedata/issues/889). Two +earlier attempts stalled, both for the same reason: scope. A universal +`ess.schemas` catalog ("all workflows for all instruments, imported by +everyone") turned shared conventions into a cross-team release-coordination +problem; a rewrite of the essreduce widget layer +([scipp/ess#689](https://github.com/scipp/ess/pull/689)) kept sciline keys, +workflow factories, and widget concerns inside the spec, so the spec could not +outlive or precede any particular implementation. + +The goal is the minimal layer that lets a *generic* user interface — ipywidgets, +a web dashboard, a command-line tool, or a GUI application — be generated from a workflow +description alone. Compute is deliberately abstracted away: the same spec must +make sense whether the workflow runs as a local sciline pipeline, behind a web +service, or as a cluster job. Compute is not part of this work, but it shapes +the design: nothing implementation-bound may appear in the spec. + +## Decision + +A new module `ess.reduce.spec` defines the spec layer. Its only dependency +beyond the standard library is pydantic (a new essreduce dependency); the +scipp-facing pieces are quarantined in one submodule. + +### The spec is pure interface: no factory, no keys, no registry + +`WorkflowSpec` holds identity (`name`, `version`), display metadata (`title`, +`description`, both mandatory), a params model, an outputs model, and an +optional `code_revision`. Nothing else. In particular it holds *no* workflow +factory and *no* sciline keys: a spec describes *what a user can configure and +what they get back*, not how it is computed. Binding a spec to an executor — +conceptually a mapping from spec identity to +`Callable[[BaseModel], BaseModel]`, or a remote service holding the same +spec — is a parallel mechanism, intentionally undefined here. This is what +keeps the spec valid across local, service, and cluster execution. + +One requirement follows for workflow packages: the module that defines a spec +must be importable without importing the workflow code. A service then loads +and validates every spec it knows without importing sciline pipelines or +instrument code, and the process that binds a spec to code is the only one +that pays for that import. esslivedata already separates the two for this +reason. + +How specs are enumerated (module-level tuples, entry points, esslivedata's +per-instrument registration) is out of scope. Any mechanism works against the +same spec type; prescribing one here would recreate the catalog problem that +sank the `ess.schemas` plan. Entry points split by role, specs in one group +and factories in another under the same name, are the natural fit for the +requirement above and are expected to become the convention once the first +service adopts this spec. + +### One params model per workflow + +Parameters are a single pydantic model class per workflow +(`params: type[BaseModel]`), not per-key entries in a registry. This enables +cross-parameter validation, gives JSON Schema for free, and removes the +implementation coupling of key-addressed parameters. The graph-derived +"select outputs, then see only relevant parameters" feature of +`ess.reduce.workflow.get_parameters` does not survive: it treats output +selection as workflow slicing, which only the sciline implementation can +express. If output-dependent parameter sets are needed, they are distinct +workflows (distinct specs). + +The field defaults to `NoParams` (a closed model with no fields), so consumers +never branch on params being absent, and sending parameters to a workflow that +takes none is a validation error rather than silently ignored. + +### Outputs are a typed model in the same vocabulary + +Outputs are likewise a pydantic model class (`outputs: type[BaseModel]`, +mandatory). Field title and description are the display metadata; a field may +be optional when the workflow does not always produce it; declaration order is +meaningful (consumers show outputs in order, primary output first). Array and +file outputs are data fields (next section); small values such as a beam +centre or a fitted scale factor are `Quantity`, a scalar or short vector with a +unit, as plain data. + +An earlier form of this decision declared outputs as a dictionary of +structural descriptions, with arrays typed by `ArraySpec` and everything else +untyped. That broke "outputs can be inputs" for exactly the values that most +often feed the next workflow. With both sides as models over one vocabulary, +chaining is a type check between an output field and a parameter field, and +where a framework stores an output (inline in a record, or in a data store) is +decided by the field's type and is not a spec concept. Output *selection* +(choosing which sciline targets to compute) is still not modeled: like +parameter slicing, it is an implementation notion. Livedata-specific output +machinery (`OutputView`, `Temporality`, windowing) stays in esslivedata. + +### Data fields: inputs are parameters + +There is no separate input section. A parameter or output that holds data +rather than a literal is a **data field**: a field of type `Ref`, annotated +with the `Format` of the bytes (raw NeXus file, scipp object, opaque file) and, +for scipp data, an `ArraySpec` describing dims, unit, coordinate units, and +whether the data is binned. The `binned` flag tells consumers which outputs are +event data that must not be plotted directly. A scalar with a unit is the 0-d +case. + +A reference is plain data naming data that exists elsewhere: an output of an +earlier run (`OutputRef`: record, output name, optionally one element of a +collection by key), or a dataset the framework did not compute (`DatasetRef`: +an identity string whose meaning, a catalogue PID or a local file's identity, +belongs to the framework). A field may be a union of a literal and a reference, +for values a user may type in or take from a previous run. Collections, +`list[...]` and `dict[str, ...]` of one declared type, are allowed on both +sides, and a reference may name one element of a collection output. + +The spec says nothing about how a workflow gets at the bytes. Whether a +reference becomes a local path or an in-memory object is decided where the +workflow is called, by the executor binding that the ADR leaves out of scope, +and the workflow asks there for the form it wants: a path for a NeXus file it +loads by component, an object for a curve it fits. A framework that adds an +in-memory fast path for chained runs therefore changes no spec and no workflow +interface. An earlier form of this decision typed a data field as a union of the +reference and the materialized value, a path or a scipp object, so that one +model served both the request and the call. That made the model wrong in both +phases, needed a validator that accepted anything not plain data, hid that +member from the JSON Schema, and put a materialization instruction into what +was meant to be pure interface. + +The format serves the consumers of the spec: a framework compares the +producer's output annotation with the consumer's parameter annotation before +chaining, a picker lists candidates of matching format, a UI selects a plotter +from the `ArraySpec`. A dataset's format is not checked at submission; a dataset +that is not what the field declares fails when the workflow reads it. Every +difference between an input and a parameter — resolution, provenance, which +widget a UI shows — is behaviour a framework selects by the field's type; the +spec only declares the type. Helpers find the data fields of a model and the +references in a plain request value, so a framework never re-derives the +annotation's meaning. The structural check of a scipp object against its +`ArraySpec` needs scipp and lives in `ess.reduce.spec.conversions`, called by +whoever runs the workflow on the outputs it returns. + +### Two forms, one-way projection + +`WorkflowSpec` is the in-process form: it holds the params and outputs model +*classes*, so same-process consumers get full pydantic validation including +custom validators. `spec.serialize()` projects onto `SerializedWorkflowSpec`, +a plain-data pydantic model with both models as JSON Schema +(`model_json_schema()`), which round-trips through JSON and is what a service +announces to remote consumers. Data fields appear in the schema under a +`dataField` key with their format and array structure; the schema is the +entire cross-process surface, sufficient to +render a form, offer a picker for data fields, and select a plotter for an +output. + +There is deliberately no inverse. Validators do not survive JSON Schema, so a +deserialized spec would be a lie about its own validation. Instead, validation +authority sits with the process owning the model classes: in-process UIs +validate directly; remote UIs validate optimistically against the schema and +the owning service accepts or rejects authoritatively. This matches the +announcement-as-contract design adopted for esslivedata in +[scipp/esslivedata#889](https://github.com/scipp/esslivedata/issues/889): the +serialized spec is the entire cross-process surface, and where a model class is +*defined* is invisible to consumers. + +### Identity is `name` + `version`; scoping is the enumerator's problem + +No `instrument` field and no `WorkflowId` class at this level. Instrument is +meaningless for technique-level batch workflows, and a spec cannot guarantee +global uniqueness of anything — only the context that enumerates or deploys +specs can. esslivedata keeps keying workflows by `(instrument, name, version)`, +supplying the instrument from its registration context. Data-provenance +identity (which spec, params, and input datasets produced a dataset) similarly +composes spec identity with deployment context; the spec's contribution is +being serializable and versioned. + +`code_revision` is provenance, not identity: an optional git commit or package +version of the code the spec describes, so that a record made from a +development branch is honest about what ran. The interface version stays +`version`. + +### Shared parameter vocabulary, scipp-free + +`ess.reduce.spec.parameters` provides constrained unit enums, range/edges +models with cross-field validation (`stop > start`, log-scale positivity) — +the models previously duplicated between esslivedata and package-specific +code — and `Quantity`. They contain no scipp: conversion of validated values +into scipp objects (`edges_to_variable`, `range_to_variables`) lives in +`ess.reduce.spec.conversions`, imported by workflow implementations only. This +keeps the vocabulary JSON-Schema-clean and the spec layer importable without +touching scipp. Value defaults (start/stop/bin counts) are set by workflow +authors at the use site, not by the vocabulary — sensible values are a +workflow/instrument decision, and a generic default is a wrong default. + +### Convergence with esslivedata + +Explicit goal: `ess.livedata.config.workflow_spec.WorkflowSpec` eventually +inherits from this spec, adding its live-data fields (`instrument`, `group`, +`source_names`, `aux_sources`, reset flags). The base spec's field names and +semantics (`name`, `version`, `title`, `description`, `params`, `outputs`) are +a strict subset of esslivedata's today for exactly this reason, and +esslivedata already declares outputs as a model class with title and +description as field metadata. The remaining difference is field types: +esslivedata's outputs are `sc.DataArray` fields with default-factory templates +used for plotter selection; here they are data fields constrained by +`ArraySpec`, which serializes. The migration (already planned independently in +scipp/esslivedata#889) changes field types only; esslivedata's `Temporality` +annotation coexists with the data-field annotation in the same `Annotated`. +The import edge is free — the esslivedata backend already depends on +essreduce, and its dashboard is decoupled via the serialized-spec announcement, +not via imports. + +## Consequences + +- Generic UIs (including a command-line interface) can be generated from + `WorkflowSpec` alone, and from `SerializedWorkflowSpec` across process + boundaries, with no knowledge of the workflow implementation. +- A framework that chains workflows validates a reference by looking up the + producer's output field and comparing its data-field annotation with the + consumer's; how strict that comparison is (format only, or full `ArraySpec` + compatibility) is the framework's rule. +- A workflow receives references and resolves them through whatever runs it; + the contract for that resolution belongs to the executor binding, not to the + spec. A runner calls `check_array` on array outputs at completion. +- essreduce gains a pydantic dependency. +- `ess.reduce.parameter`, `ess.reduce.workflow`, and the widgets built on them + are superseded and will be removed in a later hard break; they are untouched + for now. The graph-derived parameter discovery they provide is dropped, not + ported. +- `ess.nmx.configurations` and esslivedata migrate to the shared vocabulary + and spec incrementally, per package, with no coordination requirement — a + package that never migrates costs the others nothing. +- The executor binding and spec enumeration remain to be designed when a + concrete consumer needs them; the spec layer does not constrain either + beyond being addressable by `(name, version)` and importable without + workflow code. + +Foreseen extensions, each one optional spec field or one field-level +annotation, deliberately not added until a consumer exists: declared failure +reasons, an intermediate flag for retention, and declared keys of a collection +output. diff --git a/packages/essreduce/docs/developer/adr/index.md b/packages/essreduce/docs/developer/adr/index.md new file mode 100644 index 000000000..da16b0c2f --- /dev/null +++ b/packages/essreduce/docs/developer/adr/index.md @@ -0,0 +1,17 @@ +# Architecture Decision Records + +Lightweight records of load-bearing design decisions and their rationale. Each +ADR captures one decision. Accepted text is not rewritten: corrections and +extensions land as a dated amendment section at the bottom, flagged in the +status line, so the original stays readable as the reasoning of its time. +Reversing or replacing a decision gets a new ADR that links back. Format +follows [scipp's ADR convention](https://github.com/scipp/scipp/tree/main/docs/development/adr). + +```{toctree} +--- +maxdepth: 1 +glob: true +--- + +0* +``` diff --git a/packages/essreduce/docs/developer/index.md b/packages/essreduce/docs/developer/index.md index e47b24cf0..64d3cffa4 100644 --- a/packages/essreduce/docs/developer/index.md +++ b/packages/essreduce/docs/developer/index.md @@ -14,4 +14,5 @@ getting-started coding-conventions dependency-management gui +adr/index ``` diff --git a/packages/essreduce/pyproject.toml b/packages/essreduce/pyproject.toml index cacd35069..e6f10ec66 100644 --- a/packages/essreduce/pyproject.toml +++ b/packages/essreduce/pyproject.toml @@ -32,6 +32,7 @@ dynamic = ["version"] dependencies = [ "dask>=2022.1.0", "graphviz>=0.20", + "pydantic>=2.5", "sciline>=25.11.0", "scipp>=26.3.1", "scippneutron>=26.6.0", diff --git a/packages/essreduce/src/ess/reduce/spec/__init__.py b/packages/essreduce/src/ess/reduce/spec/__init__.py new file mode 100644 index 000000000..6d2aec14f --- /dev/null +++ b/packages/essreduce/src/ess/reduce/spec/__init__.py @@ -0,0 +1,46 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Scipp contributors (https://github.com/scipp) +""" +Implementation-independent workflow specifications for UI generation. + +See :mod:`ess.reduce.spec._workflow_spec` for the design, :mod:`~.data` for +data fields, and ADR 0001 (docs/developer/adr) for the rationale. +""" + +from ._workflow_spec import NoParams, SerializedWorkflowSpec, WorkflowSpec +from .data import ( + Array, + ArraySpec, + DataField, + DatasetRef, + Format, + NexusFile, + OpaqueFile, + OutputRef, + Ref, + as_ref, + data_fields, + ref_fields, + walk_refs, +) +from .parameters import Quantity + +__all__ = [ + 'Array', + 'ArraySpec', + 'DataField', + 'DatasetRef', + 'Format', + 'NexusFile', + 'NoParams', + 'OpaqueFile', + 'OutputRef', + 'Quantity', + 'Ref', + 'SerializedWorkflowSpec', + 'WorkflowSpec', + 'as_ref', + 'data_fields', + 'ref_fields', + 'walk_refs', +] diff --git a/packages/essreduce/src/ess/reduce/spec/_workflow_spec.py b/packages/essreduce/src/ess/reduce/spec/_workflow_spec.py new file mode 100644 index 000000000..421543e0b --- /dev/null +++ b/packages/essreduce/src/ess/reduce/spec/_workflow_spec.py @@ -0,0 +1,148 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Scipp contributors (https://github.com/scipp) +""" +Workflow specifications: implementation-independent workflow metadata. + +A :class:`WorkflowSpec` describes a workflow's user-facing interface — identity, +display metadata, parameters, and outputs — without reference to how or where +the workflow is computed. User interfaces (widgets, dashboards, command-line +tools) are generated from the spec alone; the binding from a spec to an +executor is a separate, parallel mechanism deliberately not defined here. + +Parameters and outputs are both pydantic model classes over one vocabulary +(:mod:`ess.reduce.spec.parameters` for literals, :mod:`ess.reduce.spec.data` +for files and arrays), so an output field of one workflow can feed a parameter +field of another when their types match. + +Two forms exist, related by a one-way projection: + +* :class:`WorkflowSpec` is the in-process form. It holds the params and outputs + model *classes*, so consumers in the same process get full pydantic + validation, including cross-field validators. +* :class:`SerializedWorkflowSpec` is the plain-data form produced by + :meth:`WorkflowSpec.serialize`, with both models as JSON Schema. It is what a + service announces to remote consumers, which can render forms and validate + optimistically against the schema. There is intentionally no inverse: + validators do not round-trip through JSON Schema, and authoritative + validation always happens in the process owning the model classes. +""" + +from __future__ import annotations + +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field + + +class NoParams(BaseModel): + """ + Params model for workflows that take no configuration. + + Workflows always have a params model, so consumers never branch on its + absence; "takes no parameters" is expressed as a model with no fields. + Extra fields are rejected so that sending params to such a workflow is an + error rather than silently ignored. + """ + + model_config = ConfigDict(extra='forbid') + + +class _SpecFields(BaseModel, frozen=True): + """Metadata fields shared by both forms of the workflow spec.""" + + name: str = Field( + min_length=1, + description=( + "Machine-readable workflow identifier. Unique within the context " + "that enumerates the spec; global uniqueness is the enumerator's " + "responsibility, not the spec's." + ), + ) + version: int = Field( + ge=1, + description=( + "Version of the workflow interface. Increment on any change a " + "consumer could observe: params model, outputs, or semantics." + ), + ) + title: str = Field(min_length=1, description="Display title of the workflow.") + description: str = Field( + min_length=1, description="Description of what the workflow computes." + ) + code_revision: str | None = Field( + default=None, + description=( + "Git commit or package version of the workflow code this spec " + "describes, so that a record made from a development branch is " + "honest about what ran. Provenance, not identity: the interface " + "version is ``version``." + ), + ) + + +class WorkflowSpec(_SpecFields, frozen=True): + """ + Implementation-independent specification of a workflow's user interface. + + Holds identity and display metadata and the pydantic model classes defining + the workflow's parameters and outputs. Contains no factory, no executor, and + no reference to any workflow implementation; pairing a spec with something + that computes it is a separate mechanism. The module defining a spec must + therefore be importable without importing the workflow code, so that a + service can load and validate every spec it knows without that code. + """ + + params: type[BaseModel] = Field( + default=NoParams, + description=( + "Pydantic model class defining the workflow parameters. Defaults " + "to :class:`NoParams` for workflows that take no configuration." + ), + ) + outputs: type[BaseModel] = Field( + description=( + "Pydantic model class defining the workflow outputs. Field title " + "and description are the display metadata; array and file outputs " + "are data fields (see :mod:`ess.reduce.spec.data`); a field may be " + "optional when the workflow does not always produce it. Order is " + "meaningful: consumers present outputs in this order and may " + "auto-select the first, so put the primary output first." + ), + ) + + def serialize(self) -> SerializedWorkflowSpec: + """ + Project to the plain-data form with params and outputs as JSON Schema. + + The projection is one-way: pydantic validators do not survive it, so + a consumer of the serialized form can validate only optimistically. + Authoritative validation happens where the model classes live. + """ + return SerializedWorkflowSpec( + name=self.name, + version=self.version, + title=self.title, + description=self.description, + code_revision=self.code_revision, + params_schema=self.params.model_json_schema(), + outputs_schema=self.outputs.model_json_schema(), + ) + + +class SerializedWorkflowSpec(_SpecFields, frozen=True): + """ + Plain-data form of a workflow spec, safe to send across process boundaries. + + Produced by :meth:`WorkflowSpec.serialize`; round-trips through JSON. Params + and outputs are represented as JSON Schema, sufficient for form generation + and optimistic validation but not for authoritative validation — that + remains with the process owning the model classes. Data fields carry a + ``dataField`` key with their format and array structure. + """ + + params_schema: dict[str, Any] = Field( + description="JSON Schema of the workflow's params model." + ) + outputs_schema: dict[str, Any] = Field( + description="JSON Schema of the workflow's outputs model." + ) diff --git a/packages/essreduce/src/ess/reduce/spec/conversions.py b/packages/essreduce/src/ess/reduce/spec/conversions.py new file mode 100644 index 000000000..685bb53c5 --- /dev/null +++ b/packages/essreduce/src/ess/reduce/spec/conversions.py @@ -0,0 +1,61 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Scipp contributors (https://github.com/scipp) +""" +The scipp side of the spec vocabulary. + +Conversions from validated parameter models to scipp objects, and the +structural check of a scipp object against an :class:`ArraySpec`. Consumed by +workflow implementations and runners only; kept out of the rest of +:mod:`ess.reduce.spec` so the vocabulary itself stays free of scipp and +serializes cleanly to JSON Schema. +""" + +import scipp as sc + +from .data import ArraySpec +from .parameters import EdgesModel, RangeModel, Scale + + +def check_array(value: sc.Variable | sc.DataArray, spec: ArraySpec) -> None: + """ + Raise ``ValueError`` unless ``value`` has the structure ``spec`` declares. + + Pydantic cannot inspect a scipp object, so a runner calls this on array + outputs at completion, and may on the arrays it resolves for a workflow. + """ + problems = [] + if tuple(value.dims) != spec.dims: + problems.append(f'dims {value.dims} != {spec.dims}') + unit = None if spec.unit is None else sc.Unit(spec.unit) + if value.unit != unit: + problems.append(f'unit {value.unit} != {unit}') + if (value.bins is not None) != spec.binned: + problems.append(f'binned={value.bins is not None} != {spec.binned}') + coords = value.coords if isinstance(value, sc.DataArray) else {} + for name, coord_unit in spec.coords.items(): + if name not in coords: + problems.append(f'missing coord {name!r}') + continue + expected = None if coord_unit is None else sc.Unit(coord_unit) + if coords[name].unit != expected: + problems.append(f'coord {name!r} unit {coords[name].unit} != {expected}') + if problems: + raise ValueError('array does not match its spec: ' + '; '.join(problems)) + + +def edges_to_variable(edges: EdgesModel, dim: str) -> sc.Variable: + """Return the bin edges described by the model as a scipp variable.""" + op = {Scale.LINEAR: sc.linspace, Scale.LOG: sc.geomspace}[edges.scale] + return op( + dim=dim, + start=edges.start, + stop=edges.stop, + num=edges.num_bins + 1, + unit=str(edges.unit), + ) + + +def range_to_variables(range_: RangeModel) -> tuple[sc.Variable, sc.Variable]: + """Return the range bounds as a pair of scipp scalars.""" + unit = str(range_.unit) + return sc.scalar(range_.start, unit=unit), sc.scalar(range_.stop, unit=unit) diff --git a/packages/essreduce/src/ess/reduce/spec/data.py b/packages/essreduce/src/ess/reduce/spec/data.py new file mode 100644 index 000000000..090b1c3cb --- /dev/null +++ b/packages/essreduce/src/ess/reduce/spec/data.py @@ -0,0 +1,213 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Scipp contributors (https://github.com/scipp) +""" +Data fields: parameters and outputs that hold data rather than literals. + +A data field is a parameter or output field whose value is a file or an array. +Its type is a :data:`Ref`, a reference to data that exists elsewhere: an output +of an earlier run, or a dataset the framework did not compute. A +:class:`DataField` annotation on the field says what the bytes are, its +:class:`Format`, and for scipp data its :class:`ArraySpec`, so that an output +field of one spec can feed a parameter field of another when the two agree, and +so that a consumer can offer candidates for a field or select a plotter for an +output. + +The spec says nothing about how a workflow gets at the bytes. A reference is +what a request names and what a record keeps; turning it into a local path or an +in-memory object is the business of whatever runs the workflow, and the workflow +asks for the form it wants. Collections of data fields, ``list[...]`` and +``dict[str, ...]`` of one declared type, are allowed and a reference may name one +element of a collection output. This module imports no scipp; the structural +check of a scipp object against its :class:`ArraySpec` lives in +:mod:`ess.reduce.spec.conversions`. +""" + +from __future__ import annotations + +from collections.abc import Iterator +from dataclasses import dataclass +from enum import StrEnum +from types import UnionType +from typing import Annotated, Any, Union, get_args, get_origin + +from pydantic import BaseModel, Field + + +class Format(StrEnum): + """What the bytes of a data field are.""" + + NEXUS = 'nexus' + """A raw NeXus file.""" + SCIPP = 'scipp' + """A scipp object, held in memory or as scipp HDF5; structure by ArraySpec.""" + OPAQUE = 'opaque' + """A file of a format the framework does not read, such as CIF or ORSO.""" + + +class ArraySpec(BaseModel, frozen=True): + """ + Structural description of an array: dimensions, units, and whether binned. + + Shape-independent, so a consumer can prepare for the data, e.g., select a + plotter, before any has been computed. A scalar with a unit is the 0-d case, + ``ArraySpec(dims=(), unit='counts')``. + """ + + dims: tuple[str, ...] = Field(description="Dimension names, outermost first.") + unit: str | None = Field( + default=None, + description=( + "Unit of the array values. None means no unit at all, as for " + "strings or datetimes; a dimensionless quantity is 'dimensionless'." + ), + ) + coords: dict[str, str | None] = Field( + default_factory=dict, + description="Coordinate names mapped to their units, as for ``unit``.", + ) + binned: bool = Field( + default=False, + description="Event data in bins; never plotted directly.", + ) + + +class OutputRef(BaseModel, frozen=True): + """Output ``output`` of record ``record``, or one element ``key`` of it.""" + + record: str = Field(min_length=1) + output: str = Field(min_length=1) + key: str | None = None + + def __str__(self) -> str: + key = f'[{self.key}]' if self.key is not None else '' + return f'{self.record}.{self.output}{key}' + + +class DatasetRef(BaseModel, frozen=True): + """ + Data the framework did not compute, named by an identity the framework owns. + + What the identity means, a catalogue PID or a local file's identity, is not + the spec's concern, and neither is the dataset's format: a dataset that is + not what the field declares fails when the workflow reads it. + """ + + dataset: str = Field(min_length=1) + + def __str__(self) -> str: + return self.dataset + + +Ref = OutputRef | DatasetRef +"""A reference: the value of a data field.""" + + +@dataclass(frozen=True) +class DataField: + """ + Field annotation marking a data field, with its format and array structure. + + Serialized into JSON Schema under the ``dataField`` key so that remote + consumers can tell data fields from literals and know their structure. + """ + + format: Format + array: ArraySpec | None = None + + def __get_pydantic_json_schema__(self, core_schema: Any, handler: Any) -> Any: + schema = handler(core_schema) + schema['dataField'] = {'format': self.format.value} + if self.array is not None: + schema['dataField']['array'] = self.array.model_dump(mode='json') + return schema + + +NexusFile = Annotated[Ref, DataField(format=Format.NEXUS)] +"""A raw NeXus file.""" +OpaqueFile = Annotated[Ref, DataField(format=Format.OPAQUE)] +"""A file the framework does not read.""" + + +def Array(spec: ArraySpec | None = None) -> Any: + """Type of a field holding a scipp object, constrained by ``spec`` if given.""" + return Annotated[Ref, DataField(format=Format.SCIPP, array=spec)] + + +def _members(annotation: Any) -> Iterator[Any]: + """The annotation and, through unions, optionals, and collections, its parts.""" + yield annotation + origin = get_origin(annotation) + if origin is Annotated: + yield from _members(get_args(annotation)[0]) + elif origin in (Union, UnionType): + for arg in get_args(annotation): + yield from _members(arg) + elif origin is list: + yield from _members(get_args(annotation)[0]) + elif origin is dict: + yield from _members(get_args(annotation)[1]) + + +def _data_field(annotation: Any) -> DataField | None: + for member in _members(annotation): + if get_origin(member) is Annotated: + for metadata in get_args(member)[1:]: + if isinstance(metadata, DataField): + return metadata + return None + + +def data_fields(model: type[BaseModel]) -> dict[str, DataField]: + """ + Data fields of a params or outputs model, by name. + + Optional fields and collections count; every element of a collection shares + the annotation. + """ + fields = {} + for name, field in model.model_fields.items(): + found = next((m for m in field.metadata if isinstance(m, DataField)), None) + if found is None: + found = _data_field(field.annotation) + if found is not None: + fields[name] = found + return fields + + +def ref_fields(model: type[BaseModel]) -> set[str]: + """Fields that may hold a reference: data fields and literal-or-reference unions.""" + data = data_fields(model) + return { + name + for name, field in model.model_fields.items() + if name in data + or any(m in (OutputRef, DatasetRef) for m in _members(field.annotation)) + } + + +_OUTPUT_REF_KEYS = frozenset(OutputRef.model_fields) + + +def as_ref(value: Any) -> Ref | None: + """The reference a plain value denotes, if it is one.""" + if isinstance(value, OutputRef | DatasetRef): + return value + if isinstance(value, dict): + keys = set(value) + if {'record', 'output'} <= keys <= _OUTPUT_REF_KEYS: + return OutputRef.model_validate(value) + if keys == {'dataset'}: + return DatasetRef.model_validate(value) + return None + + +def walk_refs(value: Any, path: str = '') -> Iterator[tuple[str, Ref]]: + """Yield every reference in a plain (JSON-shaped) value, with its path.""" + if (ref := as_ref(value)) is not None: + yield path, ref + elif isinstance(value, dict): + for k, v in value.items(): + yield from walk_refs(v, f'{path}.{k}' if path else str(k)) + elif isinstance(value, list): + for i, v in enumerate(value): + yield from walk_refs(v, f'{path}[{i}]') diff --git a/packages/essreduce/src/ess/reduce/spec/parameters.py b/packages/essreduce/src/ess/reduce/spec/parameters.py new file mode 100644 index 000000000..4b9124d54 --- /dev/null +++ b/packages/essreduce/src/ess/reduce/spec/parameters.py @@ -0,0 +1,208 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Scipp contributors (https://github.com/scipp) +""" +Shared vocabulary of workflow parameter models. + +Common building blocks for workflow params models: constrained unit choices and +range/edges models with cross-field validation. Purely declarative — no scipp; +converting validated values into scipp objects is the workflow implementation's +concern (see :mod:`ess.reduce.spec.conversions`). + +Value defaults (start, stop, number of bins) are deliberately not provided +here: sensible values depend on the workflow and instrument, so workflow +authors set them at the use site, e.g.:: + + class MyParams(pydantic.BaseModel): + wavelength: WavelengthEdges = WavelengthEdges( + start=1.0, stop=10.0, num_bins=200 + ) +""" + +from __future__ import annotations + +from abc import ABC +from enum import StrEnum + +from pydantic import BaseModel, Field, field_validator, model_validator + + +class Scale(StrEnum): + """Spacing of generated bin edges.""" + + LINEAR = 'linear' + LOG = 'log' + + +class TimeUnit(StrEnum): + """Allowed units for time.""" + + NS = 'ns' + US = 'us' + MICROSECOND = 'µs' + MS = 'ms' + S = 's' + + +class WavelengthUnit(StrEnum): + """Allowed units for wavelength.""" + + ANGSTROM = 'Å' + NANOMETER = 'nm' + + +class DspacingUnit(StrEnum): + """Allowed units for d-spacing.""" + + ANGSTROM = 'Å' + NANOMETER = 'nm' + + +class LengthUnit(StrEnum): + """Allowed units for length.""" + + METER = 'm' + CENTIMETER = 'cm' + MILLIMETER = 'mm' + + +class AngleUnit(StrEnum): + """Allowed units for angles.""" + + DEGREE = 'deg' + RADIAN = 'rad' + + +class QUnit(StrEnum): + """Allowed units for momentum transfer Q.""" + + INVERSE_ANGSTROM = '1/Å' + INVERSE_NANOMETER = '1/nm' + + +class EnergyUnit(StrEnum): + """Allowed units for energy transfer.""" + + MILLI_EV = 'meV' + MICRO_EV = 'µeV' + + +class Quantity(BaseModel, frozen=True): + """ + A scalar or short vector with a unit, as plain data. + + The small-value type for parameters and outputs such as a beam centre or a + fitted scale factor, which are typed in or chained between workflows + without going through the data store. + """ + + value: float | tuple[float, ...] = Field(description="The value(s).") + unit: str | None = Field(default=None, description="Unit, if any.") + + +class RangeModel(BaseModel, ABC): + """Base model for a value range. Subclasses constrain the unit.""" + + start: float = Field(description="Start of the range.") + stop: float = Field(description="Stop of the range.") + unit: str + + @field_validator('stop') + @classmethod + def stop_must_be_greater_than_start(cls, v: float, info) -> float: + start = info.data.get('start') + if start is not None and v <= start: + raise ValueError('stop must be greater than start') + return v + + +class EdgesModel(BaseModel, ABC): + """Base model for bin edges. Subclasses constrain the unit.""" + + start: float = Field(description="First bin edge.") + stop: float = Field(description="Last bin edge.") + num_bins: int = Field(ge=1, le=10000, description="Number of bins.") + scale: Scale = Field( + default=Scale.LINEAR, + description="Spacing of the edges, either 'linear' or 'log'.", + ) + unit: str + + @field_validator('stop') + @classmethod + def stop_must_be_greater_than_start(cls, v: float, info) -> float: + start = info.data.get('start') + if start is not None and v <= start: + raise ValueError('stop must be greater than start') + return v + + @model_validator(mode='after') + def start_must_be_positive_if_log(self) -> EdgesModel: + if self.scale == Scale.LOG and self.start <= 0: + raise ValueError("start must be positive when scale is 'log'") + return self + + +class TOARange(RangeModel): + """Time-of-arrival range.""" + + unit: TimeUnit = Field( + default=TimeUnit.MICROSECOND, description="Unit of the range bounds." + ) + + +class WavelengthRange(RangeModel): + """Wavelength range.""" + + unit: WavelengthUnit = Field( + default=WavelengthUnit.ANGSTROM, description="Unit of the range bounds." + ) + + +class TOAEdges(EdgesModel): + """Time-of-arrival bin edges.""" + + unit: TimeUnit = Field(default=TimeUnit.MS, description="Unit of the edges.") + + +class WavelengthEdges(EdgesModel): + """Wavelength bin edges.""" + + unit: WavelengthUnit = Field( + default=WavelengthUnit.ANGSTROM, description="Unit of the edges." + ) + + +class DspacingEdges(EdgesModel): + """D-spacing bin edges.""" + + unit: DspacingUnit = Field( + default=DspacingUnit.ANGSTROM, description="Unit of the edges." + ) + + +class TwoThetaEdges(EdgesModel): + """Scattering angle (two-theta) bin edges.""" + + unit: AngleUnit = Field(default=AngleUnit.DEGREE, description="Unit of the edges.") + + +class ThetaEdges(EdgesModel): + """Theta bin edges.""" + + unit: AngleUnit = Field(default=AngleUnit.DEGREE, description="Unit of the edges.") + + +class QEdges(EdgesModel): + """Momentum transfer (Q) bin edges.""" + + unit: QUnit = Field( + default=QUnit.INVERSE_ANGSTROM, description="Unit of the edges." + ) + + +class EnergyEdges(EdgesModel): + """Energy transfer bin edges.""" + + unit: EnergyUnit = Field( + default=EnergyUnit.MILLI_EV, description="Unit of the edges." + ) diff --git a/packages/essreduce/tests/spec/conversions_test.py b/packages/essreduce/tests/spec/conversions_test.py new file mode 100644 index 000000000..ffdc13723 --- /dev/null +++ b/packages/essreduce/tests/spec/conversions_test.py @@ -0,0 +1,79 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Scipp contributors (https://github.com/scipp) +import pytest +import scipp as sc + +from ess.reduce.spec import ArraySpec +from ess.reduce.spec.conversions import ( + check_array, + edges_to_variable, + range_to_variables, +) +from ess.reduce.spec.parameters import ( + Scale, + TOARange, + WavelengthEdges, +) + + +class TestCheckArray: + @pytest.fixture + def iofq(self) -> sc.DataArray: + return sc.DataArray( + sc.ones(dims=['Q'], shape=[3], unit='counts'), + coords={'Q': sc.linspace('Q', 0.0, 1.0, 4, unit='1/Å')}, + ) + + def test_matching_array_passes(self, iofq: sc.DataArray) -> None: + check_array(iofq, ArraySpec(dims=('Q',), unit='counts', coords={'Q': '1/Å'})) + + def test_variable_matches_spec_without_coords(self) -> None: + check_array(sc.scalar(1.0, unit='m'), ArraySpec(dims=(), unit='m')) + + def test_no_unit_is_distinct_from_dimensionless(self) -> None: + check_array(sc.scalar('a', unit=None), ArraySpec(dims=())) + check_array(sc.scalar(1.0), ArraySpec(dims=(), unit='dimensionless')) + with pytest.raises(ValueError, match='unit'): + check_array(sc.scalar(1.0), ArraySpec(dims=())) + + @pytest.mark.parametrize( + 'spec', + [ + ArraySpec(dims=('x',), unit='counts'), + ArraySpec(dims=('Q',), unit='m'), + ArraySpec(dims=('Q',), unit='counts', coords={'wavelength': 'Å'}), + ArraySpec(dims=('Q',), unit='counts', coords={'Q': 'nm'}), + ArraySpec(dims=('Q',), unit='counts', binned=True), + ], + ) + def test_mismatch_raises(self, iofq: sc.DataArray, spec: ArraySpec) -> None: + with pytest.raises(ValueError, match='does not match'): + check_array(iofq, spec) + + def test_binned_data_matches_binned_spec(self) -> None: + events = sc.data.binned_x(nevent=10, nbin=2) + check_array(events, ArraySpec(dims=('x',), unit='K', binned=True)) + with pytest.raises(ValueError, match='binned'): + check_array(events, ArraySpec(dims=('x',), unit='K')) + + +def test_linear_edges() -> None: + edges = WavelengthEdges(start=1.0, stop=10.0, num_bins=9) + var = edges_to_variable(edges, dim='wavelength') + assert sc.identical( + var, sc.linspace('wavelength', start=1.0, stop=10.0, num=10, unit='Å') + ) + + +def test_log_edges() -> None: + edges = WavelengthEdges(start=1.0, stop=100.0, num_bins=2, scale=Scale.LOG) + var = edges_to_variable(edges, dim='wavelength') + assert sc.identical( + var, sc.geomspace('wavelength', start=1.0, stop=100.0, num=3, unit='Å') + ) + + +def test_range_to_variables() -> None: + low, high = range_to_variables(TOARange(start=10.0, stop=20.0)) + assert sc.identical(low, sc.scalar(10.0, unit='µs')) + assert sc.identical(high, sc.scalar(20.0, unit='µs')) diff --git a/packages/essreduce/tests/spec/data_test.py b/packages/essreduce/tests/spec/data_test.py new file mode 100644 index 000000000..492f2742c --- /dev/null +++ b/packages/essreduce/tests/spec/data_test.py @@ -0,0 +1,131 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Scipp contributors (https://github.com/scipp) +from pathlib import Path + +import pytest +import scipp as sc +from pydantic import BaseModel, ValidationError + +from ess.reduce.spec import ( + Array, + ArraySpec, + DatasetRef, + Format, + NexusFile, + OpaqueFile, + OutputRef, + Quantity, + as_ref, + data_fields, + ref_fields, + walk_refs, +) + + +class Params(BaseModel): + data: Array() + background: Array(ArraySpec(dims=('x',))) | None = None + runs: list[NexusFile] = [] + banks: dict[str, Array(ArraySpec(dims=('tof',)))] = {} + centre: Quantity | OutputRef | None = None + label: str = '' + + +OUTPUT_REF = {'record': 'r1', 'output': 'data'} +DATASET_REF = {'dataset': 'pid-1'} + + +class TestFieldIntrospection: + def test_data_fields_include_optionals_and_collections(self) -> None: + fields = data_fields(Params) + assert set(fields) == {'data', 'background', 'runs', 'banks'} + assert fields['data'].format is Format.SCIPP + assert fields['data'].array is None + assert fields['background'].array == ArraySpec(dims=('x',)) + assert fields['banks'].array == ArraySpec(dims=('tof',)) + assert fields['runs'].format is Format.NEXUS + + def test_ref_fields_include_literal_or_reference_unions(self) -> None: + assert ref_fields(Params) == {'data', 'background', 'runs', 'banks', 'centre'} + + +class TestValidation: + def test_data_field_accepts_either_reference_form(self) -> None: + assert Params(data=OUTPUT_REF).data == OutputRef(record='r1', output='data') + assert Params(data=DATASET_REF).data == DatasetRef(dataset='pid-1') + + @pytest.mark.parametrize( + 'bad', + ['a path', Path('/data/run.nxs'), 3, [1, 2], {'x': 1}, None, sc.scalar(1.0)], + ) + def test_data_field_rejects_anything_but_a_reference(self, bad: object) -> None: + with pytest.raises(ValidationError): + Params(data=bad) + + def test_file_field_holds_references_like_any_data_field(self) -> None: + params = Params(data=OUTPUT_REF, runs=[DATASET_REF, OUTPUT_REF]) + assert params.runs == [ + DatasetRef(dataset='pid-1'), + OutputRef(record='r1', output='data'), + ] + with pytest.raises(ValidationError): + Params(data=OUTPUT_REF, runs=['/data/run.nxs']) + + def test_literal_or_reference_union_accepts_both(self) -> None: + params = Params(data=OUTPUT_REF, centre={'value': (0.1, 0.2), 'unit': 'm'}) + assert params.centre == Quantity(value=(0.1, 0.2), unit='m') + params = Params(data=OUTPUT_REF, centre={'record': 'r0', 'output': 'centre'}) + assert params.centre == OutputRef(record='r0', output='centre') + + +class TestJsonSchema: + def test_marks_data_fields_with_format_and_structure(self) -> None: + schema = Params.model_json_schema()['properties'] + assert schema['data']['dataField'] == {'format': 'scipp'} + assert schema['runs']['items']['dataField'] == {'format': 'nexus'} + assert schema['background']['anyOf'][0]['dataField']['array'] == { + 'dims': ['x'], + 'unit': None, + 'coords': {}, + 'binned': False, + } + assert 'dataField' not in schema['label'] + + @pytest.mark.parametrize('annotation', [Array(), NexusFile, OpaqueFile]) + def test_data_field_schema_is_the_two_reference_forms( + self, annotation: object + ) -> None: + class P(BaseModel): + field: annotation + + forms = { + c['$ref'] for c in P.model_json_schema()['properties']['field']['anyOf'] + } + assert forms == {'#/$defs/OutputRef', '#/$defs/DatasetRef'} + + +class TestReferences: + def test_walk_refs_finds_references_at_any_depth(self) -> None: + params = { + 'data': OUTPUT_REF, + 'runs': [DATASET_REF, {'record': 'f2', 'output': 'file'}], + 'banks': {'a': {'record': 'r2', 'output': 'banks', 'key': 'a'}}, + 'centre': {'value': 1.0, 'unit': 'm'}, + } + assert [(p, str(r)) for p, r in walk_refs(params)] == [ + ('data', 'r1.data'), + ('runs[0]', 'pid-1'), + ('runs[1]', 'f2.file'), + ('banks.a', 'r2.banks[a]'), + ] + + def test_as_ref_decides_what_a_reference_is(self) -> None: + assert as_ref(OutputRef(record='r', output='o')) == OutputRef( + record='r', output='o' + ) + assert as_ref(OUTPUT_REF) == OutputRef(record='r1', output='data') + assert as_ref(DATASET_REF) == DatasetRef(dataset='pid-1') + assert as_ref({'record': 'r1', 'output': 'o', 'extra': 1}) is None + assert as_ref({'dataset': 'pid', 'extra': 1}) is None + assert as_ref({'value': 1.0}) is None + assert as_ref('r1.data') is None diff --git a/packages/essreduce/tests/spec/parameters_test.py b/packages/essreduce/tests/spec/parameters_test.py new file mode 100644 index 000000000..78fc487c6 --- /dev/null +++ b/packages/essreduce/tests/spec/parameters_test.py @@ -0,0 +1,78 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Scipp contributors (https://github.com/scipp) +import pydantic +import pytest + +from ess.reduce.spec.parameters import ( + Quantity, + Scale, + WavelengthEdges, + WavelengthRange, + WavelengthUnit, +) + + +class TestQuantity: + def test_scalar_and_vector(self) -> None: + assert Quantity(value=1.0, unit='m').value == 1.0 + assert Quantity(value=(0.1, 0.2), unit='m').value == (0.1, 0.2) + assert Quantity(value=2.0).unit is None + + def test_roundtrips_through_json(self) -> None: + q = Quantity(value=(0.1, 0.2), unit='m') + assert Quantity.model_validate_json(q.model_dump_json()) == q + + +class TestRangeModel: + def test_valid_range(self) -> None: + r = WavelengthRange(start=1.0, stop=2.0) + assert r.unit == WavelengthUnit.ANGSTROM + + def test_stop_must_exceed_start(self) -> None: + with pytest.raises(pydantic.ValidationError): + WavelengthRange(start=2.0, stop=1.0) + with pytest.raises(pydantic.ValidationError): + WavelengthRange(start=1.0, stop=1.0) + + def test_bounds_are_required(self) -> None: + with pytest.raises(pydantic.ValidationError): + WavelengthRange(stop=2.0) + + def test_unit_is_constrained(self) -> None: + with pytest.raises(pydantic.ValidationError): + WavelengthRange(start=1.0, stop=2.0, unit='m') + + +class TestEdgesModel: + def test_valid_edges(self) -> None: + edges = WavelengthEdges(start=1.0, stop=10.0, num_bins=100) + assert edges.scale == Scale.LINEAR + + def test_stop_must_exceed_start(self) -> None: + with pytest.raises(pydantic.ValidationError): + WavelengthEdges(start=10.0, stop=1.0, num_bins=100) + + def test_log_scale_requires_positive_start(self) -> None: + with pytest.raises(pydantic.ValidationError): + WavelengthEdges(start=0.0, stop=10.0, num_bins=100, scale=Scale.LOG) + WavelengthEdges(start=0.1, stop=10.0, num_bins=100, scale=Scale.LOG) + + def test_num_bins_bounds(self) -> None: + with pytest.raises(pydantic.ValidationError): + WavelengthEdges(start=1.0, stop=10.0, num_bins=0) + with pytest.raises(pydantic.ValidationError): + WavelengthEdges(start=1.0, stop=10.0, num_bins=10001) + + +class TestJsonSchema: + def test_unit_choices_appear_as_enum(self) -> None: + schema = WavelengthEdges.model_json_schema() + unit_ref = schema['properties']['unit'] + enum = schema['$defs']['WavelengthUnit']['enum'] + assert set(enum) == {'Å', 'nm'} + assert unit_ref is not None + + def test_validated_model_roundtrips_through_json(self) -> None: + edges = WavelengthEdges(start=1.0, stop=10.0, num_bins=100, scale=Scale.LOG) + restored = WavelengthEdges.model_validate_json(edges.model_dump_json()) + assert restored == edges diff --git a/packages/essreduce/tests/spec/workflow_spec_test.py b/packages/essreduce/tests/spec/workflow_spec_test.py new file mode 100644 index 000000000..a67d167b1 --- /dev/null +++ b/packages/essreduce/tests/spec/workflow_spec_test.py @@ -0,0 +1,143 @@ +# SPDX-License-Identifier: BSD-3-Clause +# Copyright (c) 2026 Scipp contributors (https://github.com/scipp) +import pydantic +import pytest +from pydantic import Field + +from ess.reduce.spec import ( + Array, + ArraySpec, + NexusFile, + NoParams, + Quantity, + SerializedWorkflowSpec, + WorkflowSpec, +) + + +class Params(pydantic.BaseModel): + sample: NexusFile + lower: float + upper: float + + @pydantic.model_validator(mode='after') + def upper_greater_than_lower(self) -> 'Params': + if self.upper <= self.lower: + raise ValueError('upper must be greater than lower') + return self + + +IOFQ = ArraySpec(dims=('Q',), unit='counts', coords={'Q': '1/Å'}) + + +class Outputs(pydantic.BaseModel): + iofq: Array(IOFQ) = Field(title='I(Q)', description='Scattering intensity.') + beam_centre: Quantity = Field(title='Beam centre') + transmission: Array() | None = Field(default=None, title='Transmission') + + +class Result(pydantic.BaseModel): + result: Array() + + +@pytest.fixture +def spec() -> WorkflowSpec: + return WorkflowSpec( + name='my-workflow', + version=1, + title='My workflow', + description='Computes things.', + params=Params, + outputs=Outputs, + ) + + +class TestWorkflowSpec: + def test_minimal_spec_defaults_to_no_params(self) -> None: + spec = WorkflowSpec( + name='wf', version=1, title='Workflow', description='D', outputs=Result + ) + assert spec.params is NoParams + assert spec.code_revision is None + + def test_outputs_are_required(self) -> None: + with pytest.raises(pydantic.ValidationError): + WorkflowSpec(name='wf', version=1, title='Workflow', description='D') + + @pytest.mark.parametrize('field', ['name', 'title', 'description']) + def test_empty_metadata_field_rejected(self, field: str) -> None: + fields = { + 'name': 'wf', + 'version': 1, + 'title': 'Workflow', + 'description': 'Does things.', + 'outputs': Result, + } + with pytest.raises(pydantic.ValidationError): + WorkflowSpec(**{**fields, field: ''}) + + def test_version_must_be_positive(self) -> None: + with pytest.raises(pydantic.ValidationError): + WorkflowSpec( + name='wf', version=0, title='W', description='D', outputs=Result + ) + + def test_spec_is_frozen(self, spec: WorkflowSpec) -> None: + with pytest.raises(pydantic.ValidationError): + spec.title = 'Other' + + def test_no_params_rejects_any_input(self) -> None: + with pytest.raises(pydantic.ValidationError): + NoParams(anything=1) + + def test_params_model_validates_in_process(self, spec: WorkflowSpec) -> None: + with pytest.raises(pydantic.ValidationError): + spec.params(sample={'dataset': 'pid'}, lower=2.0, upper=1.0) + + def test_output_metadata_is_field_metadata(self, spec: WorkflowSpec) -> None: + fields = spec.outputs.model_fields + assert list(fields) == ['iofq', 'beam_centre', 'transmission'] + assert fields['iofq'].title == 'I(Q)' + assert fields['iofq'].description == 'Scattering intensity.' + assert fields['transmission'].is_required() is False + + +class TestSerialization: + def test_serialize_projects_models_to_json_schema(self, spec: WorkflowSpec) -> None: + serialized = spec.serialize() + assert serialized.params_schema == Params.model_json_schema() + assert serialized.outputs_schema == Outputs.model_json_schema() + + def test_serialize_preserves_metadata(self) -> None: + spec = WorkflowSpec( + name='wf', + version=2, + title='W', + description='D', + code_revision='abc123', + outputs=Result, + ) + serialized = spec.serialize() + assert (serialized.name, serialized.version) == ('wf', 2) + assert (serialized.title, serialized.description) == ('W', 'D') + assert serialized.code_revision == 'abc123' + + def test_output_order_and_metadata_preserved(self, spec: WorkflowSpec) -> None: + properties = spec.serialize().outputs_schema['properties'] + assert list(properties) == ['iofq', 'beam_centre', 'transmission'] + assert properties['iofq']['title'] == 'I(Q)' + assert properties['iofq']['description'] == 'Scattering intensity.' + + def test_serialized_spec_roundtrips_through_json(self, spec: WorkflowSpec) -> None: + serialized = spec.serialize() + restored = SerializedWorkflowSpec.model_validate_json( + serialized.model_dump_json() + ) + assert restored == serialized + + def test_array_structure_survives_json_roundtrip(self, spec: WorkflowSpec) -> None: + restored = SerializedWorkflowSpec.model_validate_json( + spec.serialize().model_dump_json() + ) + iofq = restored.outputs_schema['properties']['iofq'] + assert ArraySpec.model_validate(iofq['dataField']['array']) == IOFQ