"""pops.case -- the top-level compilable assembly (Spec 5 sec.5.16 / sec.11).
:class:`Case` is the inert, typed top-level assembly a user authors before lowering:
a mesh ``layout``, one or more physics ``block`` declarations, the elliptic ``field``
problems, the runtime ``param`` declarations, the static ``aux`` inputs, the ``output``
policies and the ``time`` scheme. A :class:`Case` is an ASSEMBLY that CONTAINS
descriptors; it is NOT itself a :class:`pops.descriptors.Descriptor` (Spec 5 sec.6 table /
sec.15: "Case non -- assemblage contenant des descriptors"). It still answers the same
inspectable surface -- it declares its requirements / capabilities / options and answers
``available(context)`` / :meth:`validate` with an EXPLAINABLE status before the runtime is
ever touched -- by implementing those methods DIRECTLY, not by inheriting a descriptor base.
It computes nothing.
``pops.compile(case, time=...)`` lowers the assembly through the EXISTING codegen
(``compile_problem``) and ``pops.bind(compiled, ...)`` wires it onto the EXISTING runtime
(``System`` / ``AmrSystem`` via the unified ``install``). The :class:`Case` here owns no
codegen and no runtime of its own; the heavy ``.so`` compile + install + run path is
Kokkos-gated and validated on CI / ROMEO. Every not-yet-wired route fails LOUD (a clear
``NotImplementedError``), never silently.
"""
from pops.descriptors import Availability
from pops.fields import FieldProblem
from pops.mesh.cartesian import CartesianMesh
from pops.mesh.layouts import AMR, Uniform
# Sentinel distinguishing "no kind= passed" from "kind=None": Case.param de-strings by rejecting
# any kind= keyword (Spec 5 sec.7) with a clear error naming the typed alternative.
_CASE_NO_KIND = object()
class _AMRPolicyHandle:
"""The ``case.amr`` refinement-policy handle (only valid for an ``AMR`` layout).
A thin authoring shim that records the refinement criteria on the case's ``AMR``
layout descriptor and returns the :class:`Case` so it chains. It owns no AMR runtime;
the policies it stores (``pops.mesh.amr.Refine`` / ``TagUnion`` / ``RegridEvery`` ...) are
inert descriptors the deferred AMR route will materialise.
"""
def __init__(self, case):
self._case = case
@property
def _layout(self):
return self._case._layout
def _refine_context(self):
"""The block model the refine subject is checked against, or ``None`` when ambiguous.
Returns the single block's physics model so ``Refine.validate`` can confirm the subject
exists. Returns ``None`` when there is not exactly one block (the subject would be
ambiguous across blocks) so the subject check DEFERS rather than guesses -- no false
positive.
"""
blocks = self._case._blocks
if len(blocks) != 1:
return None
(spec,) = blocks.values()
return spec.get("physics")
def refine(self, criterion=None, *, regrid=None, nesting=None, patches=None):
"""Record the refinement criterion / regrid / nesting / patch policies (chains).
When a @p criterion is recorded, its subject (role / state component / named aux) is
validated against the case's block model HERE -- the one place the model is available
-- so a refinement on a bogus role is refused before runtime with a clear, declared-roles
message (Spec 5 sec.8.6 / criterion 31). The discipline is NO FALSE POSITIVE: the check
only runs when exactly one block model is present and it advertises its subjects.
"""
layout = self._layout
if criterion is not None:
if hasattr(criterion, "validate"):
criterion.validate(self._refine_context())
layout.refine = criterion
if regrid is not None:
layout.regrid = regrid
if nesting is not None:
layout.nesting = nesting
if patches is not None:
layout.patches = patches
return self._case
[docs]
class Case:
"""A typed, inert top-level assembly: layout + blocks + fields + params + aux + outputs.
``Case(layout=Uniform(CartesianMesh()), name="plasma")`` then chained::
case = (pops.Case(name="plasma")
.block("ne", physics=model, spatial=pops.FiniteVolume())
.field(pops.fields.FieldProblem(unknown=phi, equation=eq, solver=mg))
.time(pops.time.Program(...)))
Each assembly setter RETURNS the case so calls chain. The default ``layout`` is a
single-level :class:`~pops.mesh.layouts.Uniform` over a default
:class:`~pops.mesh.cartesian.CartesianMesh`. :meth:`validate` runs structural checks; multi-block
assemblies, named non-Poisson elliptic fields and output / checkpoint policies all lower
(C3 / C1-System / C4); ``compile`` / ``bind`` flow them onto the runtime. AMR per-level output
(``Plotfile`` / level filtering) is the one remaining deferral (ADC-511).
A Case CONTAINS descriptors (the layout, the blocks' physics, the field problems) but is
NOT itself a :class:`pops.descriptors.Descriptor` (Spec 5 sec.6 table / sec.15). It exposes
the same inspectable surface -- ``requirements`` / ``capabilities`` / ``options`` /
``available`` / ``validate`` / ``inspect`` / ``lower`` -- implemented DIRECTLY here, so it
duck-types as a route-describing object without inheriting a descriptor identity.
"""
category = "case"
#: A Case names a pure-Python assembly, not a single native C++ symbol.
native_id = None
def __init__(self, layout=None, name=None):
self._name = str(name) if name else "Case"
self._layout = layout if layout is not None else Uniform(CartesianMesh())
self._blocks = {} # name -> {"physics": <model>, "spatial": <brick or None>}
self._fields = {} # field.name -> FieldProblem
self._params = {} # param_name -> {"default": value, "kind": str}
self._aux = {} # aux_name -> value/descriptor
self._outputs = [] # output / checkpoint policies (bind() flows them onto the System; run() fires them)
self._time = None # optional pops.time.Program (attaches at compile time)
@property
def name(self):
return self._name
# --- assembly (chaining setters) ----------------------------------------
[docs]
def block(self, name, physics, spatial=None):
"""Declare a physics block ``name`` (its ``physics`` model is REQUIRED). Chains."""
key = str(name)
if physics is None:
raise ValueError("Case.block(%r): a physics model is required" % key)
if key in self._blocks:
raise ValueError("Case.block(%r): a block of that name already exists" % key)
self._blocks[key] = {"physics": physics, "spatial": spatial}
return self
[docs]
def field(self, field_problem):
"""Register an elliptic :class:`~pops.fields.FieldProblem` (keyed on its name). Chains."""
if not isinstance(field_problem, FieldProblem):
raise TypeError("Case.field: expected a pops.fields.FieldProblem; got %r"
% type(field_problem).__name__)
key = field_problem.name
if key in self._fields:
raise ValueError("Case.field: a field named %r already exists" % key)
self._fields[key] = field_problem
return self
[docs]
def param(self, name, default=None, *, kind=_CASE_NO_KIND):
"""Declare a runtime/const parameter and its default value. Chains.
The KIND is a TYPED param object (Spec 5 sec.7), not a ``kind=`` string:
``case.param(pops.physics.RuntimeParam("alpha", 1.0))`` /
``case.param(pops.physics.ConstParam("gamma", 1.4))`` / ``case.param("alpha", 1.0)``
(const shorthand). A bare ``kind="const"/"runtime"`` keyword is REJECTED -- the string
route is removed; the typed sugars lower to the SAME ``{default, kind}`` record.
"""
if kind is not _CASE_NO_KIND:
raise TypeError(
"Case.param: the kind= string is removed (Spec 5 sec.7); pass a typed param "
"object (pops.physics.RuntimeParam(name, value) or "
"pops.physics.ConstParam(name, value)) instead of kind=%r" % (kind,))
if hasattr(name, "kind") and hasattr(name, "name") and hasattr(name, "value"):
# A typed pops.physics param object (ConstParam / RuntimeParam): read its identity.
if default is not None:
raise TypeError(
"Case.param: a typed param was given; do not also pass a default (%r)"
% (default,))
self._params[str(name.name)] = {"default": name.value, "kind": str(name.kind)}
else:
# The (name, default) shorthand declares a CONST param (the default mode).
self._params[str(name)] = {"default": default, "kind": "const"}
return self
[docs]
def aux(self, name, value=None):
"""Declare a static aux input ``name`` (e.g. a background field). Chains."""
self._aux[str(name)] = value
return self
[docs]
def output(self, policy):
"""Attach an output / checkpoint policy. Chains.
The policy is stored on the Case; ``bind()`` flows it onto the bound ``System`` and
``run()`` fires it at each policy cadence through the existing write / checkpoint writers
(ADC-509). AMR per-level output (``Plotfile`` / level filtering) remains deferred (ADC-511).
"""
self._outputs.append(policy)
return self
[docs]
def time(self, program):
"""Attach the time scheme (a ``pops.time.Program``) used at compile. Chains."""
self._time = program
return self
# --- layout / amr access -------------------------------------------------
@property
def layout(self):
return self._layout
@property
def amr(self):
"""The AMR refinement-policy handle (raises ``ValueError`` if layout is not AMR)."""
if not isinstance(self._layout, AMR):
raise ValueError(
"Case.amr: only available with layout=AMR(...); this case has layout %r"
% type(self._layout).__name__)
return _AMRPolicyHandle(self)
# --- DescriptorProtocol surface (pure Python; no runtime, no codegen) ----
def options(self):
return {"name": self._name, "layout": self._layout.name,
"n_blocks": len(self._blocks), "n_fields": len(self._fields),
"n_params": len(self._params), "n_aux": len(self._aux),
"n_outputs": len(self._outputs), "has_time": self._time is not None}
def requirements(self):
req = dict(self._layout.requirements())
if self._fields:
req["elliptic_solve"] = True
req["time_scheme"] = True
return req
def capabilities(self):
caps = dict(self._layout.capabilities())
caps["blocks"] = sorted(self._blocks)
caps["fields"] = sorted(self._fields)
return caps
[docs]
def available(self, context=None):
"""An EXPLAINABLE availability status, computed from the parts (no runtime)."""
layout_status = self._layout.available(context)
if not layout_status.ok:
return layout_status
if not self._blocks:
return Availability.no("case has no block; add one with .block(name, physics)",
missing=["block"])
for name, spec in self._blocks.items():
if spec.get("physics") is None:
return Availability.no("block %r has no physics model" % name,
missing=["physics"])
for field in self._fields.values():
status = field.available(context)
if not status.ok:
return status
collisions = set(self._blocks) & set(self._fields)
if collisions:
return Availability.no(
"block and field share name(s): %s" % ", ".join(sorted(collisions)),
missing=list(collisions))
return Availability.yes()
[docs]
def validate(self, context=None):
"""Structural validation; raise LOUD on a structural error (never fake success).
Checks the layout, that there is at least one block each with a physics model, that
every field problem is itself valid, and that no block and field share a name. Multi-block
assemblies, named non-Poisson elliptic fields and output / checkpoint policies now LOWER
(C3 / C1-System / C4); each output entry is only confirmed to be a real policy descriptor
here (a non-policy object is rejected loud). AMR per-level output is deferred (ADC-511).
"""
self._layout.validate(context)
if not self._blocks:
raise ValueError(
"Case.validate: no block declared; add one with .block(name, physics)")
for name, spec in self._blocks.items():
if spec.get("physics") is None:
raise ValueError("Case.validate: block %r has no physics model" % name)
# Carry the mesh layout into each field problem's validation so its solver can refuse a
# layout it cannot serve, precisely, before any compile (Spec 6 sec.8/9: the spectral FFT
# solver requires a uniform periodic mesh, not an AMR hierarchy).
field_context = self._field_validation_context(context)
for field in self._fields.values():
field.validate(field_context)
collisions = set(self._blocks) & set(self._fields)
if collisions:
raise ValueError("Case.validate: block and field share name(s): %s"
% ", ".join(sorted(collisions)))
# Multi-block (C3) and named non-Poisson elliptic fields (C1-System) now LOWER -- the
# block-count and field-name rejects are removed. A named field that no block's model
# declares is still caught downstream at install (_install_solver checks the declared
# elliptic_field_names set).
#
# OUTPUT / CHECKPOINT policies (C4 / ADC-509) now LOWER on the Uniform / System path: the
# NotImplementedError deferral is removed. bind() flows the policies onto the bound System
# (its run() fires them at each policy cadence through the existing write()/checkpoint
# writers). validate() only confirms each entry is a real policy descriptor (its category is
# one of the two policy categories) -- a non-policy object is a typo, rejected loud here
# rather than at run time. AMR per-level writes (Plotfile / level filtering) remain ADC-511.
for policy in self._outputs:
cat = getattr(policy, "category", None)
if cat not in ("output_policy", "checkpoint_policy"):
raise TypeError(
"Case.validate: output() expects a pops.output.OutputPolicy / CheckpointPolicy; "
"got %r (category %r)" % (type(policy).__name__, cat))
return True
def _field_validation_context(self, context):
"""The validation context handed to each field problem, carrying the mesh layout.
A field problem's solver may refuse a layout it cannot serve (Spec 6 sec.8/9). The Case
owns the authoritative layout, so it is merged under the ``"layout"`` key (overriding any
caller-supplied layout); a dict @p context's other keys are preserved, and a non-dict /
absent context yields a fresh single-key dict.
"""
merged = dict(context) if isinstance(context, dict) else {}
merged["layout"] = self._layout
return merged
[docs]
def explain_routes(self):
"""Return a printable route matrix sourced from the C++ authoritative facts (sec.13.12.1).
Spec 5 sec.13.12.1 (criterion #37): a ``feature x layout x backend x platform -> status /
limitation`` matrix whose capability VALUES come from the C++ core
(``_pops.module_capabilities()``), NOT a Python-derived walk. Each transport feature
(uniform / amr / mpi / gpu / stride / named_fields / partial_imex_mask) is reported with the
native ``available`` / ``unavailable`` status the built module decides, plus the route facts
the descriptor catalog adds (the layouts this case can author). Inert: it imports ``_pops``
LAZILY and reads metadata only -- no compile, bind or run.
When ``_pops`` is absent or predates ``module_capabilities`` (old build), the matrix reports
every native status as ``unknown`` with that reason rather than fabricating a value.
"""
return build_route_matrix(self)
[docs]
def lower(self, context=None):
"""The inert lowering record for the assembly (metadata only; no computation).
Mirrors the descriptor :meth:`lower` shape so a Case duck-types as a route-describing
object, without being a :class:`pops.descriptors.Descriptor`. It runs no numeric loop and
touches no runtime; ``pops.compile`` does the real lowering through ``compile_problem``.
"""
return {"name": self._name, "category": self.category,
"native_id": self.native_id, "options": self.options()}
def inspect(self):
# The base inspect() dict, inlined (a Case is no longer a Descriptor subclass).
info = {"name": self._name, "category": self.category, "native_id": self.native_id,
"options": self.options(), "requirements": self.requirements(),
"capabilities": self.capabilities()}
info["layout"] = self._layout.inspect()
info["blocks"] = {
name: {"physics": getattr(spec["physics"], "name", repr(spec["physics"])),
"spatial": getattr(spec["spatial"], "name", spec["spatial"])}
for name, spec in self._blocks.items()}
info["fields"] = {name: fp.inspect() for name, fp in self._fields.items()}
info["params"] = dict(self._params)
info["aux"] = sorted(self._aux)
info["outputs"] = [getattr(p, "name", repr(p)) for p in self._outputs]
info["time"] = getattr(self._time, "name", None) if self._time is not None else None
return info
def __str__(self):
return ("%s [%s] layout=%s | blocks=%d | fields=%d | params=%d | aux=%d | time=%s"
% (self._name, self.category, self._layout.name, len(self._blocks),
len(self._fields), len(self._params), len(self._aux),
"set" if self._time is not None else "none"))
def __repr__(self):
return ("Case(name=%r, layout=%s, blocks=%s, fields=%s)"
% (self._name, self._layout.name, sorted(self._blocks), sorted(self._fields)))
# --- route matrix (Spec 5 sec.13.12.1, criterion #37) -----------------------------------
# The transport features the route matrix reports, paired with the platform / backend axis each
# one lives on. The capability VALUES are sourced from the C++ _pops.module_capabilities(), never a
# Python computation; this table only names the features and their human-facing axis.
_ROUTE_FEATURES = (
("supports_uniform", "layout", "uniform single-level grid"),
("supports_amr", "layout", "adaptive mesh refinement"),
("supports_mpi", "backend", "distributed MPI transport"),
("supports_gpu", "backend", "GPU device backend (Kokkos CUDA/HIP)"),
("supports_stride", "transport", "strided cell access (production route)"),
("supports_named_fields", "transport", "named aux-field transport"),
("supports_partial_imex_mask", "transport", "partial IMEX mask"),
)
class RouteRow:
"""One row of a :class:`RouteMatrix`: a feature x axis -> status / limitation (sec.13.12.1).
A plain, inert value. ``status`` is ``"available"`` / ``"unavailable"`` / ``"unknown"``;
``source`` is ``"native"`` (the C++ ``module_capabilities`` flag) or ``"unknown"`` (no _pops);
``limitation`` carries the honest reason a feature is unavailable / unknown (else empty).
"""
def __init__(self, feature, axis, status, *, source, limitation=""):
self.feature = feature
self.axis = axis
self.status = status
self.source = source
self.limitation = limitation
def to_dict(self):
return {"feature": self.feature, "axis": self.axis, "status": self.status,
"source": self.source, "limitation": self.limitation}
def __repr__(self):
return ("RouteRow(feature=%r, axis=%r, status=%r, source=%r)"
% (self.feature, self.axis, self.status, self.source))
class RouteMatrix:
"""The printable route matrix of a :class:`Case` (Spec 5 sec.13.12.1, criterion #37).
Holds the per-feature :class:`RouteRow`s (capability values sourced from the C++ core), the
case name and its authored layout. :meth:`to_dict` returns a plain nested dict and
:meth:`__str__` a short, deterministic table. It is inert -- it computes nothing.
"""
def __init__(self, case_name, layout_name, rows):
self.case_name = case_name
self.layout_name = layout_name
self.rows = list(rows)
def to_dict(self):
return {"case": self.case_name, "layout": self.layout_name,
"rows": [r.to_dict() for r in self.rows]}
def __iter__(self):
return iter(self.rows)
def __repr__(self):
return "RouteMatrix(case=%r, layout=%r, %d rows)" % (
self.case_name, self.layout_name, len(self.rows))
def __str__(self):
lines = ["route matrix for %r (layout=%s, Spec 5 sec.13.12.1):"
% (self.case_name, self.layout_name)]
for row in self.rows:
note = (" -- %s" % row.limitation) if row.limitation else ""
lines.append(" %-28s [%-9s] %-12s (source=%s)%s"
% (row.feature, row.axis, row.status, row.source, note))
return "\n".join(lines)
def _module_capabilities():
"""The C++ authoritative capability dict (``_pops.module_capabilities()``) or ``None``.
Lazily imports ``_pops`` (top-level then ``pops._pops``) so :mod:`pops.case` stays codegen /
``_pops`` free at module scope and the import graph acyclic. Returns ``None`` when ``_pops`` is
unavailable or predates ``module_capabilities`` (old build) -- the matrix then reports the native
statuses as ``unknown`` rather than fabricating a value."""
try:
import _pops as mod # noqa: PLC0415 -- lazy: keeps pops.case free of _pops at import time
except Exception:
try:
from pops import _pops as mod # noqa: PLC0415
except Exception:
return None
fn = getattr(mod, "module_capabilities", None)
if fn is None:
return None
try:
return dict(fn())
except Exception:
return None
def build_route_matrix(case):
"""Build the :class:`RouteMatrix` of @p case from the C++ authoritative facts (sec.13.12.1).
Each :data:`_ROUTE_FEATURES` entry becomes a :class:`RouteRow` whose status is read straight
from the C++ ``_pops.module_capabilities()`` flag (``source="native"``). A feature the C++ source
reports ``False`` carries an honest limitation note (e.g. ``partial_imex_mask`` has no C++ path);
when ``_pops`` is absent the row is ``unknown`` with that reason -- never a fabricated value."""
native_caps = _module_capabilities()
rows = []
for feature, axis, description in _ROUTE_FEATURES:
if native_caps is None or feature not in native_caps:
rows.append(RouteRow(
feature, axis, "unknown", source="unknown",
limitation="_pops.module_capabilities() unavailable (module not built or too old)"))
continue
if native_caps[feature]:
rows.append(RouteRow(feature, axis, "available", source="native", limitation=description))
else:
rows.append(RouteRow(
feature, axis, "unavailable", source="native",
limitation="%s: not provided by this build" % description))
return RouteMatrix(case.name, case.layout.name, rows)
__all__ = ["Case", "RouteMatrix", "RouteRow", "build_route_matrix"]