"""The stage registry: stage name -> ``Stage`` class (5.1a).
The single seam where *config* becomes a *running stage* -- the stage-plane
mirror of ``backends/registry.py``. A user's YAML names each stage entry by
the stage's own name (``compose``, ``structure``, ...); this module maps that
string to the concrete class and instantiates it with the entry's options.
Registration is an **explicit call**, not an import-time decorator, so there
is no hidden import-order dependency: a concrete stage module calls
``register("compose", ComposeStage)`` at the bottom of its file, and whoever
wires the tool imports the stage modules it intends to offer (the loader's
``build_stages`` imports the built-ins).
**Construction contract:** a registered class is instantiated as
``cls(**options)`` -- unlike backends (whose ``cls(name, options)`` keeps a
free-form dict because every kind wants different keys), each stage already
declares a typed keyword signature and validates every value itself, raising
``StageError`` for anything invalid. Unpacking the config options onto that
signature means an unknown option fails loudly with the stage named, instead
of riding along silently in a dict.
``FUNNEL_ORDER`` is the funnel, written down exactly once: it is the
reference against which ``build_stages`` checks a config's stage set.
``OPTIONAL_STAGES`` names the funnel members a config may omit -- the
configured stages must follow the funnel order, skipping only these.
``STAGE_OUTPUT_STATE`` is its state-plane companion: each canonical stage
name -> the lifecycle state its completion lands. The orchestrator derives
every stage's input state (the previous stage's output; the first stage
consumes ``NEW``) and its completion target from this map -- so the
stage <-> state coupling lives here by *name*, never by list position, and
omitting an optional stage from a run cannot shift its neighbours' states.
"""
from __future__ import annotations
from typing import Any
from matmelt.core.stages.base import Stage, StageError
from matmelt.core.state import MaterialState
#: The funnel, in orchestrator order. ``build_stages`` requires a config's
#: stages to follow this order, omitting only ``OPTIONAL_STAGES`` members.
FUNNEL_ORDER: tuple[str, ...] = ("compose", "structure", "screen", "calculate", "extract")
#: Funnel members a config may leave out entirely. Every other funnel stage
#: up to the furthest configured one is required -- a hole at a required
#: stage is refused by ``build_stages``, never silently bridged.
OPTIONAL_STAGES: frozenset[str] = frozenset({"calculate"})
#: Canonical stage name -> the lifecycle state its completion lands. Stage
#: names not in this map (test doubles) take the next chain state
#: positionally.
STAGE_OUTPUT_STATE: dict[str, MaterialState] = {
"compose": MaterialState.COMPOSED,
"structure": MaterialState.STRUCTURED,
"screen": MaterialState.SCREENED,
"calculate": MaterialState.CALCULATED,
"extract": MaterialState.EXTRACTED,
}
#: The live ``name`` -> class table. Module-global and intentionally mutable:
#: concrete stage modules register into it when imported. Starts empty.
_REGISTRY: dict[str, type[Stage]] = {}
[docs]
def register(name: str, cls: type[Stage]) -> None:
"""Register ``cls`` as the implementation of stage ``name``.
Called explicitly at the bottom of each concrete stage module.
Parameters
----------
name
The stage's canonical name, which is also what a config's ``stages:``
block keys on.
cls
The implementing class, which must be a ``Stage`` subclass.
Raises
------
StageError
If ``cls`` is not a ``Stage`` subclass, or if ``name`` is already
registered to a *different* class. Re-registering the same class is
idempotent, so a module re-import is a no-op, but silently shadowing an
existing stage is refused.
"""
if not (isinstance(cls, type) and issubclass(cls, Stage)):
raise StageError(f"cannot register {cls!r} for stage {name!r}: not a Stage subclass")
existing = _REGISTRY.get(name)
if existing is not None and existing is not cls:
raise StageError(
f"stage {name!r} is already registered to {existing.__name__}; "
f"refusing to shadow it with {cls.__name__}"
)
_REGISTRY[name] = cls
[docs]
def get(name: str) -> type[Stage]:
"""Return the class registered under ``name``.
Parameters
----------
name
The stage name to look up.
Returns
-------
type
The registered ``Stage`` subclass.
Raises
------
StageError
If nothing is registered under ``name``. The message names the stages
that *are* known, so a config typo produces an actionable message
rather than a ``KeyError``.
"""
try:
return _REGISTRY[name]
except KeyError:
known = ", ".join(sorted(_REGISTRY)) or "(none registered)"
raise StageError(f"unknown stage {name!r}; known stages: {known}") from None
[docs]
def create_stage(name: str, options: dict[str, Any]) -> Stage:
"""Instantiate the stage ``name`` with its config ``options`` as kwargs.
The one seam where a config block becomes a live stage.
Parameters
----------
name
Which registered stage to construct.
options
The entry's ``options:`` block, unpacked onto the stage's typed
keyword signature.
Returns
-------
Stage
The constructed, option-validated stage.
Raises
------
StageError
If ``name`` is unknown; if the stage's own ``__init__`` validation
rejects a value; or if an option the stage does not accept (or a
non-string key, which cannot be a keyword) is passed -- caught here and
reported with the stage named, instead of riding along silently.
Examples
--------
>>> from matmelt.core.stages import compose # registers "compose"
>>> stage = create_stage("compose", {"python": "python3", "threshold": 2})
>>> stage.name, stage.runner_python
('compose', 'python3')
"""
cls = get(name)
try:
return cls(**options)
except TypeError as exc:
raise StageError(f"stage {name!r}: invalid options: {exc}") from None
[docs]
def known_stages() -> list[str]:
"""List the stage names currently registered.
Returns
-------
list of str
Sorted names, used in diagnostics and error messages. Which stages are
present depends on which stage modules have been imported.
"""
return sorted(_REGISTRY)