omnibias.fields
Structured scalar, vector, and tensor fields with derivative caches, calculus operators, and coordinate conventions.
import omnibias.fieldsChoose fields when you want to express calculus in terms of structured field objects rather than manually assembling every activation derivative. FieldState carries the model representation; backend views expose operators and cache derivative quantities for that representation.
A scalar field, vector field, and tensor field have different component axes. Gradient, Jacobian, divergence, curl, and Hessian are not interchangeable layouts. Respect the coordinate order and component count described by each operator; curl and related vector identities have dimensional assumptions. Weak-form operations additionally depend on the supplied quadrature or test-function representation.
Cached quantities are tied to the state that produced them. Reconstruct or invalidate the relevant view when inputs or parameters change according to the API contract. Validate both an analytic field identity and retained parameter gradients on the operator path used in your loss. The field-calculus architecture guide explains the representation and the dedicated module pages document the operators.
Start with FieldState and the backend view for your model. For a scalar field, evaluate its value and gradient before adding a Laplacian to a loss. Check a small analytic function independently and verify parameter gradients before reusing cached quantities in training. The first-Laplacian tutorial and Poisson notebook demonstrate an independently checked direct operator; those array examples are not a substitute for the structured FieldState contract.
Module contract & mathematical conventions
Backend-neutral field schemas, derivative caching, and reusable calculus.
FieldState exposes gradient, divergence, curl, Laplacian, Hessian, integration, and norms through PyTorch/JAX operators. Backends are imported lazily. Use matching dtypes (JAX_ENABLE_X64=1 for float64) for cross-backend comparisons.
Shared formulas do not imply universal floating-point identity. Verify dtype, shape, derivative-order support, and the stated numerical assumptions for the path you use.
Imports & aliases
These names share a definition with another public module. Follow a name for its full contract, parameters, and examples.
| Import from this module | Canonical reference |
|---|---|
AffineSet | omnibias.core.AffineSet |
EqualitySystem | omnibias.core.EqualitySystem |
NewtonResult | omnibias.core.NewtonResult |
UnitTerm | omnibias.core.UnitTerm |
affine_locus | omnibias.core.affine_locus |
certify_locus_point | omnibias.core.certify_locus_point |
DOMAINS
omnibias.fields.DOMAINSSource ↗DOMAINS: tuple[str, ...] = ('calculus', 'vector_calculus', 'conservation', 'fluids', 'mechanics', 'complex', 'integral')Ordered domain tags, coarse maths/physics areas, used to group the catalog.
Regression example · every operator has a known domain
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_every_operator_has_a_known_domain():
for op in list_operators():
assert op.domain in DOMAINS, oppython -m pytest packages/omnibias-fields/tests/test_catalog.py -k test_every_operator_has_a_known_domain -qRegression example · every domain is populated
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_every_domain_is_populated():
for domain in DOMAINS:
assert list_operators(domain=domain), f"domain {domain!r} has no operators"python -m pytest packages/omnibias-fields/tests/test_catalog.py -k test_every_domain_is_populated -qRead the implementation
DOMAINS: tuple[str, ...] = (
"calculus",
"vector_calculus",
"conservation",
"fluids",
"mechanics",
"complex",
"integral",
)OperatorInfo
omnibias.fields.OperatorInfoSource ↗class OperatorInfoDataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.
Metadata for one dispatchable field operator.
Parameters
name The state.ops.<name> entry point (also the DSL / ops module name).
domain One of DOMAINS.
formula A compact ASCII signature of what the operator computes.
closed_form True when the operator is the closed-form sigma tower or an exact composition of it (no autodiff, no grid approximation).
backends Backends providing a mathematically equivalent implementation.
Backend scope: this identifies the corresponding mathematical construction. It does not guarantee bitwise-equal floating-point results across devices, compilers, dtypes, or backend versions. Validate the selected path with an appropriate absolute and relative tolerance.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namedataclass field | strRequired | The state.ops.<name> entry point (also the DSL / ops module name). |
domaindataclass field | strRequired | One of DOMAINS. |
formuladataclass field | strRequired | A compact ASCII signature of what the operator computes. |
closed_formdataclass field | boolDefault: True | True when the operator is the closed-form sigma tower or an exact composition of it (no autodiff, no grid approximation). |
backendsdataclass field | tuple[str, ...]Default: ('torch', 'jax') | Backends providing a mathematically equivalent implementation. |
Fields & methods
closed_formfield
closed_form: bool = TrueTrue when the operator is the closed-form sigma tower or an exact composition of it (no autodiff, no grid approximation).
backendsfield
backends: tuple[str, ...] = ('torch', 'jax')Backends providing a mathematically equivalent implementation.
View source ↗Read the implementation
class OperatorInfo:
"""Metadata for one dispatchable field operator.
Parameters
----------
name
The ``state.ops.<name>`` entry point (also the DSL / ``ops`` module name).
domain
One of :data:`DOMAINS`.
formula
A compact ASCII signature of what the operator computes.
closed_form
``True`` when the operator is the closed-form sigma tower or an exact
composition of it (no autodiff, no grid approximation).
backends
Backends providing a bit-identical implementation.
"""
name: str
domain: str
formula: str
closed_form: bool = True
backends: tuple[str, ...] = ("torch", "jax")get_operator
omnibias.fields.get_operatorSource ↗get_operator(name: str) -> OperatorInfoReturn the OperatorInfo for name (raises KeyError if absent).
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
OperatorInfoExecuted regression · get operator roundtrips and raises
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_get_operator_roundtrips_and_raises():
info = get_operator("gradient")
assert info.domain == "calculus"
assert info.formula
with pytest.raises(KeyError):
get_operator("does_not_exist")python -m pytest "packages/omnibias-fields/tests/test_catalog.py::test_get_operator_roundtrips_and_raises" -qRegression example · aliases are catalogued with their primaries
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_aliases_are_catalogued_with_their_primaries():
names = operator_names()
for alias, primary in (("rot", "curl"), ("div", "divergence"), ("wave_operator", "dalembertian")):
assert alias in names and primary in names
assert get_operator(alias).domain == get_operator(primary).domainpython -m pytest packages/omnibias-fields/tests/test_catalog.py -k test_aliases_are_catalogued_with_their_primaries -qRead the implementation
def get_operator(name: str) -> OperatorInfo:
"""Return the :class:`OperatorInfo` for ``name`` (raises ``KeyError`` if absent)."""
try:
return _BY_NAME[name]
except KeyError:
raise KeyError(f"no catalogued operator named {name!r}") from Nonelist_operators
omnibias.fields.list_operatorsSource ↗list_operators(domain: str | None=None) -> tuple[OperatorInfo, ...]Return catalogued operators, optionally filtered by domain.
Parameters
domain One of DOMAINS. None (default) returns every operator, ordered by domain then name.
Examples
>>> from omnibias.fields import list_operators
>>> [op.name for op in list_operators(domain="complex")][:3]
['dz', 'dzbar']Parameters
| Name | Type & default | Contract |
|---|---|---|
domainpositional-or-keyword | str | NoneDefault: None | One of DOMAINS. None (default) returns every operator, ordered by domain then name. |
tuple[OperatorInfo, ...]Explicit validation & failure cases
- When
domain is not None and domain not in DOMAINS, raisesValueError(f'unknown domain {domain!r}; choose from {DOMAINS!r}').
These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.
Executed regression · every operator has a known domain
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_every_operator_has_a_known_domain():
for op in list_operators():
assert op.domain in DOMAINS, oppython -m pytest "packages/omnibias-fields/tests/test_catalog.py::test_every_operator_has_a_known_domain" -qRegression example · every domain is populated
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_every_domain_is_populated():
for domain in DOMAINS:
assert list_operators(domain=domain), f"domain {domain!r} has no operators"python -m pytest packages/omnibias-fields/tests/test_catalog.py -k test_every_domain_is_populated -qRead the implementation
def list_operators(domain: str | None = None) -> tuple[OperatorInfo, ...]:
"""Return catalogued operators, optionally filtered by ``domain``.
Parameters
----------
domain
One of :data:`DOMAINS`. ``None`` (default) returns every operator,
ordered by domain then name.
Examples
--------
>>> from omnibias.fields import list_operators
>>> [op.name for op in list_operators(domain="complex")][:3]
['dz', 'dzbar']
"""
if domain is not None and domain not in DOMAINS:
raise ValueError(f"unknown domain {domain!r}; choose from {DOMAINS!r}")
ops = _CATALOG if domain is None else tuple(op for op in _CATALOG if op.domain == domain)
return tuple(sorted(ops, key=lambda op: (DOMAINS.index(op.domain), op.name)))operator_names
omnibias.fields.operator_namesSource ↗operator_names() -> frozenset[str]The set of all catalogued operator names.
frozenset[str]Executed regression · catalog matches jax dispatch surface exactly
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_catalog_matches_jax_dispatch_surface_exactly():
assert operator_names() == _dispatch_names("jax")python -m pytest "packages/omnibias-fields/tests/test_catalog.py::test_catalog_matches_jax_dispatch_surface_exactly" -qRegression example · catalog matches torch dispatch surface exactly
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_catalog_matches_torch_dispatch_surface_exactly():
assert operator_names() == _dispatch_names("torch")python -m pytest packages/omnibias-fields/tests/test_catalog.py -k test_catalog_matches_torch_dispatch_surface_exactly -qRead the implementation
def operator_names() -> frozenset[str]:
"""The set of all catalogued operator names."""
return frozenset(_BY_NAME)ComponentSpec
omnibias.fields.ComponentSpecSource ↗class ComponentSpecFrozen ordered list of scalar component names plus named groups.
Parameters
names Ordered names of the scalar components produced by the field. Must be non-empty, all distinct, all non-empty strings. groups Optional mapping from group name to a tuple of component names. Each group name must not collide with any component name and each member must be a known component.
Examples
>>> spec = ComponentSpec(
... names=("u", "v", "w", "p"),
... groups={"velocity": ("u", "v", "w")},
... )
>>> spec.names
('u', 'v', 'w', 'p')
>>> spec.groups["velocity"]
('u', 'v', 'w')Parameters
| Name | Type & default | Contract |
|---|---|---|
namespositional-or-keyword | Sequence[str]Required | Ordered names of the scalar components produced by the field. Must be non-empty, all distinct, all non-empty strings. |
groupskeyword-only | Mapping[str, Sequence[str]] | NoneDefault: None | Optional mapping from group name to a tuple of component names. Each group name must not collide with any component name and each member must be a known component. |
Executed regression · specs construct
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_specs_construct() -> None:
coords = CoordinateSpec(axes=("t", "x"))
comps = ComponentSpec(names=("u",))
assert coords.ndim == 2
assert comps.is_component("u")
assert coords.axis_index("x") == 1python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_specs_construct" -qFields & methods
namesfield
names: tuple[str, ...]Ordered names of the scalar components produced by the field. Must be non-empty, all distinct, all non-empty strings.
View source ↗groupsfield
groups: tuple[tuple[str, tuple[str, ...]], ...]Optional mapping from group name to a tuple of component names. Each group name must not collide with any component name and each member must be a known component.
View source ↗__init__method
__init__(self, names: Sequence[str], *, groups: Mapping[str, Sequence[str]] | None=None) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namespositional-or-keyword | Sequence[str]Required | See the operation contract and implementation below. |
groupskeyword-only | Mapping[str, Sequence[str]] | NoneDefault: None | See the operation contract and implementation below. |
Returns: None
Validation & failure cases
- When
not norm_names, raisesValueError('ComponentSpec.names must be non-empty').
Executed regression · specs construct
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_specs_construct() -> None:
coords = CoordinateSpec(axes=("t", "x"))
comps = ComponentSpec(names=("u",))
assert coords.ndim == 2
assert comps.is_component("u")
assert coords.axis_index("x") == 1python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_specs_construct" -qMethod implementation
def __init__(
self,
names: Sequence[str],
*,
groups: Mapping[str, Sequence[str]] | None = None,
) -> None:
norm_names = _normalise_names(names)
if not norm_names:
raise ValueError("ComponentSpec.names must be non-empty")
norm_groups = _normalise_groups(groups, norm_names)
groups_t = tuple((k, v) for k, v in norm_groups.items())
object.__setattr__(self, "names", norm_names)
object.__setattr__(self, "groups", groups_t)n_componentsproperty
n_components(self) -> intSee the class contract above and the exact method implementation below.
Returns: int
Method implementation
def n_components(self) -> int:
return len(self.names)indexmethod
index(self, name: str) -> intResolve a component name to its integer position in names.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
Returns: int
Validation & failure cases
- When
name not in self.names, raisesKeyError(f'component {name!r} not in {self.names!r}').
Method implementation
def index(self, name: str) -> int:
"""Resolve a component name to its integer position in ``names``."""
if name not in self.names:
raise KeyError(
f"component {name!r} not in {self.names!r}"
)
return self.names.index(name)hasmethod
has(self, name: str) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
Returns: bool
Method implementation
def has(self, name: str) -> bool:
return name in self.names or any(g == name for g, _ in self.groups)is_componentmethod
is_component(self, name: str) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
Returns: bool
Executed regression · specs construct
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_specs_construct() -> None:
coords = CoordinateSpec(axes=("t", "x"))
comps = ComponentSpec(names=("u",))
assert coords.ndim == 2
assert comps.is_component("u")
assert coords.axis_index("x") == 1python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_specs_construct" -qMethod implementation
def is_component(self, name: str) -> bool:
return name in self.namesis_groupmethod
is_group(self, name: str) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
Returns: bool
Executed regression · dsl views match dispatch
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_views_match_dispatch():
nodes = _nodes(3)
st = _state("torch", "3d", nodes)
assert np.allclose(
_np(st.vel.grad_div), _np(st.ops.gradient_of_divergence(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.curl_curl), _np(st.ops.curl_of_curl(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.rate_of_rotation),
_np(st.ops.rate_of_rotation_tensor(st, ("u", "v", "w"))),
)
assert np.allclose(_np(st.c.grad_sq), _np(st.ops.grad_squared_norm(st, "c")))python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_views_match_dispatch" -qMethod implementation
def is_group(self, name: str) -> bool:
return any(g == name for g, _ in self.groups)group_membersmethod
group_members(self, name: str) -> tuple[str, ...]See the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
Returns: tuple[str, ...]
Executed regression · dsl views match dispatch
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_views_match_dispatch():
nodes = _nodes(3)
st = _state("torch", "3d", nodes)
assert np.allclose(
_np(st.vel.grad_div), _np(st.ops.gradient_of_divergence(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.curl_curl), _np(st.ops.curl_of_curl(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.rate_of_rotation),
_np(st.ops.rate_of_rotation_tensor(st, ("u", "v", "w"))),
)
assert np.allclose(_np(st.c.grad_sq), _np(st.ops.grad_squared_norm(st, "c")))python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_views_match_dispatch" -qMethod implementation
def group_members(self, name: str) -> tuple[str, ...]:
for g, members in self.groups:
if g == name:
return members
raise KeyError(f"group {name!r} not in {[g for g, _ in self.groups]!r}")all_known_namesmethod
all_known_names(self) -> tuple[str, ...]See the class contract above and the exact method implementation below.
Returns: tuple[str, ...]
Method implementation
def all_known_names(self) -> tuple[str, ...]:
return self.names + tuple(g for g, _ in self.groups)__len__method
__len__(self) -> intSee the class contract above and the exact method implementation below.
Returns: int
Method implementation
def __len__(self) -> int:
return len(self.names)__contains__method
__contains__(self, name: object) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | objectRequired | See the operation contract and implementation below. |
Returns: bool
Method implementation
def __contains__(self, name: object) -> bool:
if not isinstance(name, str):
return False
return self.has(name)__iter__method
__iter__(self)See the class contract above and the exact method implementation below.
Method implementation
def __iter__(self):
return iter(self.names)__repr__method
__repr__(self) -> strSee the class contract above and the exact method implementation below.
Returns: str
Method implementation
def __repr__(self) -> str:
bits = [f"names={self.names!r}"]
if self.groups:
gd = {g: list(m) for g, m in self.groups}
bits.append(f"groups={gd!r}")
return f"ComponentSpec({', '.join(bits)})"asdictmethod
asdict(self) -> dictSee the class contract above and the exact method implementation below.
Returns: dict
Method implementation
def asdict(self) -> dict:
return {
"names": list(self.names),
"groups": {g: list(m) for g, m in self.groups},
}Read the implementation
class ComponentSpec:
"""Frozen ordered list of scalar component names plus named groups.
Parameters
----------
names
Ordered names of the scalar components produced by the field. Must
be non-empty, all distinct, all non-empty strings.
groups
Optional mapping from group name to a tuple of component names.
Each group name must not collide with any component name and each
member must be a known component.
Examples
--------
>>> spec = ComponentSpec(
... names=("u", "v", "w", "p"),
... groups={"velocity": ("u", "v", "w")},
... )
>>> spec.names
('u', 'v', 'w', 'p')
>>> spec.groups["velocity"]
('u', 'v', 'w')
"""
names: tuple[str, ...]
groups: tuple[tuple[str, tuple[str, ...]], ...]
def __init__(
self,
names: Sequence[str],
*,
groups: Mapping[str, Sequence[str]] | None = None,
) -> None:
norm_names = _normalise_names(names)
if not norm_names:
raise ValueError("ComponentSpec.names must be non-empty")
norm_groups = _normalise_groups(groups, norm_names)
groups_t = tuple((k, v) for k, v in norm_groups.items())
object.__setattr__(self, "names", norm_names)
object.__setattr__(self, "groups", groups_t)
@property
def n_components(self) -> int:
return len(self.names)
def index(self, name: str) -> int:
"""Resolve a component name to its integer position in ``names``."""
if name not in self.names:
raise KeyError(
f"component {name!r} not in {self.names!r}"
)
return self.names.index(name)
def has(self, name: str) -> bool:
return name in self.names or any(g == name for g, _ in self.groups)
def is_component(self, name: str) -> bool:
return name in self.names
def is_group(self, name: str) -> bool:
return any(g == name for g, _ in self.groups)
def group_members(self, name: str) -> tuple[str, ...]:
for g, members in self.groups:
if g == name:
return members
raise KeyError(f"group {name!r} not in {[g for g, _ in self.groups]!r}")
def all_known_names(self) -> tuple[str, ...]:
return self.names + tuple(g for g, _ in self.groups)
def __len__(self) -> int:
return len(self.names)
def __contains__(self, name: object) -> bool:
if not isinstance(name, str):
return False
return self.has(name)
def __iter__(self):
return iter(self.names)
def __repr__(self) -> str:
bits = [f"names={self.names!r}"]
if self.groups:
gd = {g: list(m) for g, m in self.groups}
bits.append(f"groups={gd!r}")
return f"ComponentSpec({', '.join(bits)})"
def asdict(self) -> dict:
return {
"names": list(self.names),
"groups": {g: list(m) for g, m in self.groups},
}CoordinateSpec
omnibias.fields.CoordinateSpecSource ↗class CoordinateSpecFrozen descriptor of a PINN field's input axes.
Parameters
axes Names of the input axes in evaluation order. The convention is spatial axes first, then time last (e.g. ("x", "y", "z", "t")).
periodicity Per-axis booleans indicating whether the axis is periodic. None is the default and means *no* axis is periodic. A single bool is broadcast over all axes.
domain Optional (lo, hi) pair per axis. None means the domain is not pinned by the spec (caller manages it). When given, hi > lo is required per axis.
time_axis The name of the time axis. None means the field is steady (no explicit time axis). Defaults to "t" if present in axes, otherwise None.
Parameters
| Name | Type & default | Contract |
|---|---|---|
axespositional-or-keyword | Sequence[str]Required | Names of the input axes in evaluation order. The convention is spatial axes first, then time last (e.g. ("x", "y", "z", "t")). |
periodicitykeyword-only | Sequence[bool] | bool | NoneDefault: None | Per-axis booleans indicating whether the axis is periodic. None is the default and means *no* axis is periodic. A single bool is broadcast over all axes. |
domainkeyword-only | Sequence[tuple[float, float]] | NoneDefault: None | Optional (lo, hi) pair per axis. None means the domain is not pinned by the spec (caller manages it). When given, hi > lo is required per axis. |
time_axiskeyword-only | str | None | objectDefault: ... | The name of the time axis. None means the field is steady (no explicit time axis). Defaults to "t" if present in axes, otherwise None. |
Executed regression · specs construct
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_specs_construct() -> None:
coords = CoordinateSpec(axes=("t", "x"))
comps = ComponentSpec(names=("u",))
assert coords.ndim == 2
assert comps.is_component("u")
assert coords.axis_index("x") == 1python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_specs_construct" -qFields & methods
axesfield
axes: tuple[str, ...]Names of the input axes in evaluation order. The convention is spatial axes first, then time last (e.g. ("x", "y", "z", "t")).
periodicityfield
periodicity: tuple[bool, ...] = field(default=())Per-axis booleans indicating whether the axis is periodic. None is the default and means *no* axis is periodic. A single bool is broadcast over all axes.
domainfield
domain: tuple[tuple[float, float], ...] | None = field(default=None)Optional (lo, hi) pair per axis. None means the domain is not pinned by the spec (caller manages it). When given, hi > lo is required per axis.
time_axisfield
time_axis: str | None = NoneThe name of the time axis. None means the field is steady (no explicit time axis). Defaults to "t" if present in axes, otherwise None.
__init__method
__init__(self, axes: Sequence[str], *, periodicity: Sequence[bool] | bool | None=None, domain: Sequence[tuple[float, float]] | None=None, time_axis: str | None | object=...) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
axespositional-or-keyword | Sequence[str]Required | See the operation contract and implementation below. |
periodicitykeyword-only | Sequence[bool] | bool | NoneDefault: None | See the operation contract and implementation below. |
domainkeyword-only | Sequence[tuple[float, float]] | NoneDefault: None | See the operation contract and implementation below. |
time_axiskeyword-only | str | None | objectDefault: ... | See the operation contract and implementation below. |
Returns: None
Validation & failure cases
- When
ta is not None and ta not in norm_axes, raisesValueError(f'time_axis {ta!r} not in axes {norm_axes!r}').
Executed regression · specs construct
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_specs_construct() -> None:
coords = CoordinateSpec(axes=("t", "x"))
comps = ComponentSpec(names=("u",))
assert coords.ndim == 2
assert comps.is_component("u")
assert coords.axis_index("x") == 1python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_specs_construct" -qMethod implementation
def __init__(
self,
axes: Sequence[str],
*,
periodicity: Sequence[bool] | bool | None = None,
domain: Sequence[tuple[float, float]] | None = None,
time_axis: str | None | object = ...,
) -> None:
norm_axes = _normalise_axes(axes)
norm_period = _coerce_periodicity(periodicity, len(norm_axes))
norm_domain = _coerce_domain(domain, len(norm_axes))
if time_axis is ...:
ta: str | None = "t" if "t" in norm_axes else None
else:
ta = time_axis # type: ignore[assignment]
if ta is not None and ta not in norm_axes:
raise ValueError(
f"time_axis {ta!r} not in axes {norm_axes!r}"
)
object.__setattr__(self, "axes", norm_axes)
object.__setattr__(self, "periodicity", norm_period)
object.__setattr__(self, "domain", norm_domain)
object.__setattr__(self, "time_axis", ta)ndimproperty
ndim(self) -> intNumber of axes.
Returns: int
Executed regression · specs construct
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_specs_construct() -> None:
coords = CoordinateSpec(axes=("t", "x"))
comps = ComponentSpec(names=("u",))
assert coords.ndim == 2
assert comps.is_component("u")
assert coords.axis_index("x") == 1python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_specs_construct" -qMethod implementation
def ndim(self) -> int:
"""Number of axes."""
return len(self.axes)spatial_axesproperty
spatial_axes(self) -> tuple[str, ...]All axes except the time axis.
Returns: tuple[str, ...]
Executed regression · cross backend bit parity
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_cross_backend_bit_parity(which, opfn):
nodes = _nodes(2 if which == "2d" else 3, seed=7)
ts = _state("torch", which, nodes)
js = _state("jax", which, nodes)
t = _np(opfn(ts.ops, ts))
j = _np(opfn(js.ops, js))
assert np.allclose(t, j, rtol=1e-12, atol=1e-12)python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_cross_backend_bit_parity[2d-<lambda>]" -qMethod implementation
def spatial_axes(self) -> tuple[str, ...]:
"""All axes except the time axis."""
if self.time_axis is None:
return self.axes
return tuple(a for a in self.axes if a != self.time_axis)n_spatialproperty
n_spatial(self) -> intNumber of spatial axes.
Returns: int
Method implementation
def n_spatial(self) -> int:
"""Number of spatial axes."""
return len(self.spatial_axes)axis_indexmethod
axis_index(self, axis: int | str) -> intResolve an axis name or integer to its integer position.
Parameters
| Name | Type & default | Contract |
|---|---|---|
axispositional-or-keyword | int | strRequired | See the operation contract and implementation below. |
Returns: int
Validation & failure cases
- When
not -self.ndim <= axis < self.ndim, raisesIndexError(f'axis index {axis} out of range for {self.ndim}-dim spec'). - When
axis not in self.axes, raisesKeyError(f'axis {axis!r} not in spec axes {self.axes!r}').
Executed regression · specs construct
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_specs_construct() -> None:
coords = CoordinateSpec(axes=("t", "x"))
comps = ComponentSpec(names=("u",))
assert coords.ndim == 2
assert comps.is_component("u")
assert coords.axis_index("x") == 1python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_specs_construct" -qMethod implementation
def axis_index(self, axis: int | str) -> int:
"""Resolve an axis name or integer to its integer position."""
if isinstance(axis, int):
if not (-self.ndim <= axis < self.ndim):
raise IndexError(
f"axis index {axis} out of range for {self.ndim}-dim spec"
)
return axis % self.ndim
if isinstance(axis, str):
if axis not in self.axes:
raise KeyError(
f"axis {axis!r} not in spec axes {self.axes!r}"
)
return self.axes.index(axis)
raise TypeError(
f"axis must be int or str, got {type(axis).__name__}: {axis!r}"
)is_periodicmethod
is_periodic(self, axis: int | str) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
axispositional-or-keyword | int | strRequired | See the operation contract and implementation below. |
Returns: bool
Method implementation
def is_periodic(self, axis: int | str) -> bool:
return self.periodicity[self.axis_index(axis)]is_spatialmethod
is_spatial(self, axis: int | str) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
axispositional-or-keyword | int | strRequired | See the operation contract and implementation below. |
Returns: bool
Method implementation
def is_spatial(self, axis: int | str) -> bool:
return self.axes[self.axis_index(axis)] != self.time_axisis_timemethod
is_time(self, axis: int | str) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
axispositional-or-keyword | int | strRequired | See the operation contract and implementation below. |
Returns: bool
Method implementation
def is_time(self, axis: int | str) -> bool:
return self.axes[self.axis_index(axis)] == self.time_axis__repr__method
__repr__(self) -> strSee the class contract above and the exact method implementation below.
Returns: str
Method implementation
def __repr__(self) -> str:
bits = [f"axes={self.axes!r}"]
if any(self.periodicity):
bits.append(f"periodicity={self.periodicity!r}")
if self.domain is not None:
bits.append(f"domain={self.domain!r}")
if self.time_axis is not None:
bits.append(f"time_axis={self.time_axis!r}")
return f"CoordinateSpec({', '.join(bits)})"asdictmethod
asdict(self) -> dict[str, Any]Return a plain-dict view (handy for JSON serialisation).
Returns: dict[str, Any]
Method implementation
def asdict(self) -> dict[str, Any]:
"""Return a plain-dict view (handy for JSON serialisation)."""
return {
"axes": list(self.axes),
"periodicity": list(self.periodicity),
"domain": [list(d) for d in self.domain] if self.domain is not None else None,
"time_axis": self.time_axis,
}Read the implementation
class CoordinateSpec:
"""Frozen descriptor of a PINN field's input axes.
Parameters
----------
axes
Names of the input axes in evaluation order. The convention is
spatial axes first, then time last (e.g. ``("x", "y", "z", "t")``).
periodicity
Per-axis booleans indicating whether the axis is periodic. ``None``
is the default and means *no* axis is periodic. A single bool is
broadcast over all axes.
domain
Optional ``(lo, hi)`` pair per axis. ``None`` means the domain is
not pinned by the spec (caller manages it). When given,
``hi > lo`` is required per axis.
time_axis
The name of the time axis. ``None`` means the field is steady (no
explicit time axis). Defaults to ``"t"`` if present in ``axes``,
otherwise ``None``.
"""
axes: tuple[str, ...]
periodicity: tuple[bool, ...] = field(default=())
domain: tuple[tuple[float, float], ...] | None = field(default=None)
time_axis: str | None = None
def __init__(
self,
axes: Sequence[str],
*,
periodicity: Sequence[bool] | bool | None = None,
domain: Sequence[tuple[float, float]] | None = None,
time_axis: str | None | object = ...,
) -> None:
norm_axes = _normalise_axes(axes)
norm_period = _coerce_periodicity(periodicity, len(norm_axes))
norm_domain = _coerce_domain(domain, len(norm_axes))
if time_axis is ...:
ta: str | None = "t" if "t" in norm_axes else None
else:
ta = time_axis # type: ignore[assignment]
if ta is not None and ta not in norm_axes:
raise ValueError(
f"time_axis {ta!r} not in axes {norm_axes!r}"
)
object.__setattr__(self, "axes", norm_axes)
object.__setattr__(self, "periodicity", norm_period)
object.__setattr__(self, "domain", norm_domain)
object.__setattr__(self, "time_axis", ta)
@property
def ndim(self) -> int:
"""Number of axes."""
return len(self.axes)
@property
def spatial_axes(self) -> tuple[str, ...]:
"""All axes except the time axis."""
if self.time_axis is None:
return self.axes
return tuple(a for a in self.axes if a != self.time_axis)
@property
def n_spatial(self) -> int:
"""Number of spatial axes."""
return len(self.spatial_axes)
def axis_index(self, axis: int | str) -> int:
"""Resolve an axis name or integer to its integer position."""
if isinstance(axis, int):
if not (-self.ndim <= axis < self.ndim):
raise IndexError(
f"axis index {axis} out of range for {self.ndim}-dim spec"
)
return axis % self.ndim
if isinstance(axis, str):
if axis not in self.axes:
raise KeyError(
f"axis {axis!r} not in spec axes {self.axes!r}"
)
return self.axes.index(axis)
raise TypeError(
f"axis must be int or str, got {type(axis).__name__}: {axis!r}"
)
def is_periodic(self, axis: int | str) -> bool:
return self.periodicity[self.axis_index(axis)]
def is_spatial(self, axis: int | str) -> bool:
return self.axes[self.axis_index(axis)] != self.time_axis
def is_time(self, axis: int | str) -> bool:
return self.axes[self.axis_index(axis)] == self.time_axis
def __repr__(self) -> str:
bits = [f"axes={self.axes!r}"]
if any(self.periodicity):
bits.append(f"periodicity={self.periodicity!r}")
if self.domain is not None:
bits.append(f"domain={self.domain!r}")
if self.time_axis is not None:
bits.append(f"time_axis={self.time_axis!r}")
return f"CoordinateSpec({', '.join(bits)})"
def asdict(self) -> dict[str, Any]:
"""Return a plain-dict view (handy for JSON serialisation)."""
return {
"axes": list(self.axes),
"periodicity": list(self.periodicity),
"domain": [list(d) for d in self.domain] if self.domain is not None else None,
"time_axis": self.time_axis,
}DISPATCH_ATTR
omnibias.fields.DISPATCH_ATTRSource ↗DISPATCH_ATTR = '_omnibias_dispatch'Name of the class attribute a concrete field sets to select the backend
ops dispatch path (e.g. "one_layer", "spectral", "chebyshev",
"cage"). The backend ops read this marker via getattr so the
foundational omnibias-fields package never has to import concrete field
classes from a downstream package (which would create a dependency cycle).
Regression example · dispatch attr name is stable
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dispatch_attr_name_is_stable() -> None:
assert DISPATCH_ATTR == "_omnibias_dispatch"python -m pytest packages/omnibias-fields/tests/_core/test_substrate.py -k test_dispatch_attr_name_is_stable -qRead the implementation
DISPATCH_ATTR = "_omnibias_dispatch"FieldBase
omnibias.fields.FieldBaseSource ↗class FieldBase(Protocol)Structural protocol every typed field must satisfy.
A concrete field additionally sets the class attribute named by
DISPATCH_ATTR to one of the dispatch tags understood by the
backend ops ("one_layer" selects the closed-form sigma-tower
reduction; other tags select the state-method path).
Fields that participate in the frozen-feature linear solver also set the
class attribute named by READOUT_INDEPENDENT_ATTR to True when
their per-state caches are independent of the readout parameters.
Fields & methods
coordinate_specproperty
coordinate_spec(self) -> CoordinateSpecSee the class contract above and the exact method implementation below.
Returns: CoordinateSpec
Method implementation
def coordinate_spec(self) -> CoordinateSpec: ...componentsproperty
components(self) -> ComponentSpecSee the class contract above and the exact method implementation below.
Returns: ComponentSpec
Method implementation
def components(self) -> ComponentSpec: ...evaluatemethod
evaluate(self, coords: Any) -> FieldStateSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
coordspositional-or-keyword | AnyRequired | See the operation contract and implementation below. |
Returns: FieldState
Method implementation
def evaluate(self, coords: Any) -> FieldState: ...__call__method
__call__(self, coords: Any) -> FieldStateSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
coordspositional-or-keyword | AnyRequired | See the operation contract and implementation below. |
Returns: FieldState
Method implementation
def __call__(self, coords: Any) -> FieldState: ...Read the implementation
class FieldBase(Protocol):
"""Structural protocol every typed field must satisfy.
A concrete field additionally sets the class attribute named by
:data:`DISPATCH_ATTR` to one of the dispatch tags understood by the
backend ops (``"one_layer"`` selects the closed-form sigma-tower
reduction; other tags select the state-method path).
Fields that participate in the frozen-feature linear solver also set the
class attribute named by :data:`READOUT_INDEPENDENT_ATTR` to ``True`` when
their per-state caches are independent of the readout parameters.
"""
@property
def coordinate_spec(self) -> CoordinateSpec: ...
@property
def components(self) -> ComponentSpec: ...
def evaluate(self, coords: Any) -> FieldState: ...
def __call__(self, coords: Any) -> FieldState: ...READOUT_INDEPENDENT_ATTR
omnibias.fields.READOUT_INDEPENDENT_ATTRSource ↗READOUT_INDEPENDENT_ATTR = '_omnibias_readout_independent'Name of the class attribute a concrete field sets to True when every
quantity cached on a produced ~omnibias.fields._core.state.FieldState
(SigmaCache and extra) is independent of the field's *readout*
parameters, and every op re-reads those parameters live.
The frozen-feature linear solver builds a collocation plan once and sweeps
the readout against the same cached states; that is sound only for fields
that declare this marker. Fields that are nonlinear in the readout (e.g. an
integral-conservation cage) must leave it unset / False so the driver
refuses them with a named error rather than silently miscomputing.
Regression example · readout independent attr name is stable
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_readout_independent_attr_name_is_stable() -> None:
assert READOUT_INDEPENDENT_ATTR == "_omnibias_readout_independent"python -m pytest packages/omnibias-fields/tests/_core/test_substrate.py -k test_readout_independent_attr_name_is_stable -qRead the implementation
READOUT_INDEPENDENT_ATTR = "_omnibias_readout_independent"SigmaCache
omnibias.fields.SigmaCacheSource ↗class SigmaCache(Generic[T])Order-keyed cache for the activation derivative tower.
Parameters
z The pre-activation tensor z = W x + beta. Stored as the z attribute and reused by every get_or_compute call.
Notes
The cache is intentionally mutable so that ops can fill it lazily.
The enclosing FieldState is itself frozen, so the *handle*
to the cache cannot be reassigned, but its contents can grow during
a single field(coords) evaluation. This matches the pattern of a
functools.cached_property but with explicit type-safe accessors.
**Immutability contract.** Every cached value is a pure function of z
captured at construction. z (and the coords it is derived from)
must therefore be treated as **immutable** for the lifetime of the cache:
mutating it in place silently invalidates every cached sigma^(n)(z).
For torch tensors this is enforced cheaply via Tensor._version -- a
read after an in-place mutation of z raises RuntimeError
rather than returning stale derivatives. Rebuild the state from the new
coords instead of mutating in place.
Parameters
| Name | Type & default | Contract |
|---|---|---|
zpositional-or-keyword | TRequired | The pre-activation tensor z = W x + beta. Stored as the z attribute and reused by every get_or_compute call. |
Executed regression · sigma cache rejects negative order
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_sigma_cache_rejects_negative_order() -> None:
cache: SigmaCache[str] = SigmaCache(z="z")
with pytest.raises(ValueError):
cache.get_or_compute(-1, lambda n: "x")python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_sigma_cache_rejects_negative_order" -qRegression example · sigma cache noop guard for non tensor z
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_sigma_cache_noop_guard_for_non_tensor_z() -> None:
cache: SigmaCache[str] = SigmaCache(z="z") # no _version -> guard inert
assert cache.get_or_compute(0, lambda n: "s0") == "s0"
assert cache.get(0) == "s0"python -m pytest packages/omnibias-fields/tests/_core/test_sigma_cache_mutation.py -k test_sigma_cache_noop_guard_for_non_tensor_z -qFields & methods
__init__method
__init__(self, z: T) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
zpositional-or-keyword | TRequired | See the operation contract and implementation below. |
Returns: None
Executed regression · sigma cache rejects negative order
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_sigma_cache_rejects_negative_order() -> None:
cache: SigmaCache[str] = SigmaCache(z="z")
with pytest.raises(ValueError):
cache.get_or_compute(-1, lambda n: "x")python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_sigma_cache_rejects_negative_order" -qMethod implementation
def __init__(self, z: T) -> None:
self.z: T = z
self._store: dict[int, T] = {}
self._z_version: int | None = _tensor_version(z)hasmethod
has(self, order: int) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
orderpositional-or-keyword | intRequired | See the operation contract and implementation below. |
Returns: bool
Method implementation
def has(self, order: int) -> bool:
return order in self._storeget_or_computemethod
get_or_compute(self, order: int, build: Callable[[int], T]) -> TReturn sigma^(order)(z), computing it on first access.
Parameters
| Name | Type & default | Contract |
|---|---|---|
orderpositional-or-keyword | intRequired | See the operation contract and implementation below. |
buildpositional-or-keyword | Callable[[int], T]Required | See the operation contract and implementation below. |
Returns: T
Validation & failure cases
- When
order < 0, raisesValueError(f'derivative order must be >= 0, got {order}').
Executed regression · sigma cache rejects negative order
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_sigma_cache_rejects_negative_order() -> None:
cache: SigmaCache[str] = SigmaCache(z="z")
with pytest.raises(ValueError):
cache.get_or_compute(-1, lambda n: "x")python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_sigma_cache_rejects_negative_order" -qMethod implementation
def get_or_compute(self, order: int, build: Callable[[int], T]) -> T:
"""Return ``sigma^(order)(z)``, computing it on first access."""
if order < 0:
raise ValueError(f"derivative order must be >= 0, got {order}")
self._assert_z_fresh()
cached = self._store.get(order)
if cached is None:
cached = build(order)
self._store[order] = cached
return cachedputmethod
put(self, order: int, value: T) -> NoneManual insert; useful when an op already knows the value.
Parameters
| Name | Type & default | Contract |
|---|---|---|
orderpositional-or-keyword | intRequired | See the operation contract and implementation below. |
valuepositional-or-keyword | TRequired | See the operation contract and implementation below. |
Returns: None
Validation & failure cases
- When
order < 0, raisesValueError(f'derivative order must be >= 0, got {order}').
Executed regression · sigma cache detects inplace mutation on get
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_sigma_cache_detects_inplace_mutation_on_get() -> None:
z = torch.ones(4)
cache: SigmaCache[torch.Tensor] = SigmaCache(z=z)
cache.put(0, z.clone())
z.mul_(2.0)
with pytest.raises(RuntimeError, match="mutated in place"):
cache.get(0)python -m pytest "packages/omnibias-fields/tests/_core/test_sigma_cache_mutation.py::test_sigma_cache_detects_inplace_mutation_on_get" -qMethod implementation
def put(self, order: int, value: T) -> None:
"""Manual insert; useful when an op already knows the value."""
if order < 0:
raise ValueError(f"derivative order must be >= 0, got {order}")
self._assert_z_fresh()
self._store[order] = valuegetmethod
get(self, order: int) -> TGet a cached value, raising KeyError if absent.
Parameters
| Name | Type & default | Contract |
|---|---|---|
orderpositional-or-keyword | intRequired | See the operation contract and implementation below. |
Returns: T
Validation & failure cases
- When
order not in self._store, raisesKeyError(f'order {order} not in sigma cache; available orders: {sorted(self._store)}').
Executed regression · sigma cache noop guard for non tensor z
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_sigma_cache_noop_guard_for_non_tensor_z() -> None:
cache: SigmaCache[str] = SigmaCache(z="z") # no _version -> guard inert
assert cache.get_or_compute(0, lambda n: "s0") == "s0"
assert cache.get(0) == "s0"python -m pytest "packages/omnibias-fields/tests/_core/test_sigma_cache_mutation.py::test_sigma_cache_noop_guard_for_non_tensor_z" -qMethod implementation
def get(self, order: int) -> T:
"""Get a cached value, raising :class:`KeyError` if absent."""
self._assert_z_fresh()
if order not in self._store:
raise KeyError(
f"order {order} not in sigma cache; "
f"available orders: {sorted(self._store)}"
)
return self._store[order]ordersmethod
orders(self) -> tuple[int, ...]Tuple of derivative orders currently cached.
Returns: tuple[int, ...]
Executed regression · sigma cache single evaluation
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_sigma_cache_single_evaluation() -> None:
calls: list[int] = []
def build(n: int) -> str:
calls.append(n)
return f"sigma^{n}"
cache: SigmaCache[str] = SigmaCache(z="z")
assert cache.get_or_compute(2, build) == "sigma^2"
assert cache.get_or_compute(2, build) == "sigma^2"
assert cache.get_or_compute(0, build) == "sigma^0"
# Order 2 evaluated once despite two accesses.
assert calls == [2, 0]
assert cache.orders() == (0, 2)python -m pytest "packages/omnibias-fields/tests/_core/test_substrate.py::test_sigma_cache_single_evaluation" -qMethod implementation
def orders(self) -> tuple[int, ...]:
"""Tuple of derivative orders currently cached."""
return tuple(sorted(self._store))clearmethod
clear(self) -> NoneSee the class contract above and the exact method implementation below.
Returns: None
Method implementation
def clear(self) -> None:
self._store.clear()__contains__method
__contains__(self, order: object) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
orderpositional-or-keyword | objectRequired | See the operation contract and implementation below. |
Returns: bool
Method implementation
def __contains__(self, order: object) -> bool:
return isinstance(order, int) and order in self._store__len__method
__len__(self) -> intSee the class contract above and the exact method implementation below.
Returns: int
Method implementation
def __len__(self) -> int:
return len(self._store)__repr__method
__repr__(self) -> strSee the class contract above and the exact method implementation below.
Returns: str
Method implementation
def __repr__(self) -> str:
return f"SigmaCache(orders={self.orders()})"zattribute
z: T = zThe pre-activation tensor z = W x + beta. Stored as the z attribute and reused by every get_or_compute call.
Read the implementation
class SigmaCache(Generic[T]):
"""Order-keyed cache for the activation derivative tower.
Parameters
----------
z
The pre-activation tensor ``z = W x + beta``. Stored as the
``z`` attribute and reused by every :meth:`get_or_compute` call.
Notes
-----
The cache is intentionally mutable so that ops can fill it lazily.
The enclosing :class:`FieldState` is itself frozen, so the *handle*
to the cache cannot be reassigned, but its contents can grow during
a single ``field(coords)`` evaluation. This matches the pattern of a
``functools.cached_property`` but with explicit type-safe accessors.
**Immutability contract.** Every cached value is a pure function of ``z``
captured at construction. ``z`` (and the ``coords`` it is derived from)
must therefore be treated as **immutable** for the lifetime of the cache:
mutating it in place silently invalidates every cached ``sigma^(n)(z)``.
For torch tensors this is enforced cheaply via ``Tensor._version`` -- a
read after an in-place mutation of ``z`` raises :class:`RuntimeError`
rather than returning stale derivatives. Rebuild the state from the new
coords instead of mutating in place.
"""
__slots__ = ("z", "_store", "_z_version")
def __init__(self, z: T) -> None:
self.z: T = z
self._store: dict[int, T] = {}
self._z_version: int | None = _tensor_version(z)
def _assert_z_fresh(self) -> None:
"""Raise if ``z`` was mutated in place since construction (torch only)."""
if self._z_version is None:
return
current = _tensor_version(self.z)
if current != self._z_version:
raise RuntimeError(
"SigmaCache.z was mutated in place after construction "
f"(version {self._z_version} -> {current}); cached "
"sigma^(n)(z) values are now stale. coords / z must be treated "
"as immutable for the lifetime of a FieldState -- rebuild the "
"state from the new coords instead of mutating in place."
)
def has(self, order: int) -> bool:
return order in self._store
def get_or_compute(self, order: int, build: Callable[[int], T]) -> T:
"""Return ``sigma^(order)(z)``, computing it on first access."""
if order < 0:
raise ValueError(f"derivative order must be >= 0, got {order}")
self._assert_z_fresh()
cached = self._store.get(order)
if cached is None:
cached = build(order)
self._store[order] = cached
return cached
def put(self, order: int, value: T) -> None:
"""Manual insert; useful when an op already knows the value."""
if order < 0:
raise ValueError(f"derivative order must be >= 0, got {order}")
self._assert_z_fresh()
self._store[order] = value
def get(self, order: int) -> T:
"""Get a cached value, raising :class:`KeyError` if absent."""
self._assert_z_fresh()
if order not in self._store:
raise KeyError(
f"order {order} not in sigma cache; "
f"available orders: {sorted(self._store)}"
)
return self._store[order]
def orders(self) -> tuple[int, ...]:
"""Tuple of derivative orders currently cached."""
return tuple(sorted(self._store))
def clear(self) -> None:
self._store.clear()
def __contains__(self, order: object) -> bool:
return isinstance(order, int) and order in self._store
def __len__(self) -> int:
return len(self._store)
def __repr__(self) -> str:
return f"SigmaCache(orders={self.orders()})"FieldState
omnibias.fields.FieldStateSource ↗class FieldState(Generic[T])Frozen evaluation snapshot of a typed PINN field.
Parameters
coords Input tensor of shape (B, D). Owned by the caller of field(coords); the state stores a reference (no copy). It must be treated as **immutable** for the lifetime of the state: the SigmaCache memoises sigma^(n)(z) against the pre-activation z derived from these coords, so mutating coords (or z) in place silently desynchronises the cached derivatives from freshly recomputed values. In-place mutation of a torch z is caught by the cache (raises on the next read); rebuild the state from new coords rather than mutating in place.
field The FieldBase instance that produced this state. Ops read field.spec (activation), field.params, etc.
components ComponentSpec describing the output channels. Frozen.
coordinate_spec CoordinateSpec describing the input axes. Frozen.
ops Backend-specific ops module (omnibias.fields.torch._ops_dispatch or its jax twin). Looked up by attribute access in the views.
sigma_cache SigmaCache keyed by derivative order. Filled lazily by whichever ops touch it first.
extra Free-form per-state cache: ops can stash intermediate tensors (like a precomputed gradient) keyed by string. Mutable but intended for op-internal use.
**Readout-independence contract.** A FieldState is a coherent snapshot of one evaluation. Entries written by the field or its ops may depend only on coords and *frozen feature* parameters (hidden weights, temporal heads, geometry factors) -- never on the *readout* parameters the frozen-feature linear solver sweeps (c / b on a one-layer field, V / b_t on a spectral / Chebyshev field, the final affine layer of a jet MLP). Fields that honour this contract declare it via the class attribute named by ~omnibias.fields._core.field_base.READOUT_INDEPENDENT_ATTR.
Caller-installed entries are the caller's responsibility. The lim_along extension, for example, stashes user-supplied closures in extra["lim_along"]; those closures are typically readout-dependent and unverifiable by the library, so installing them disqualifies the field from declaring readout-independence for any subsequent frozen-feature solve against that state.
Notes
The class behaves frozen-ish: the dataclass attributes are set in
__init__ and not changed afterwards. The SigmaCache
contents and the extra dict are intentionally mutable so ops can
fill them lazily during a single residual evaluation.
Parameters
| Name | Type & default | Contract |
|---|---|---|
coordskeyword-only | TRequired | Input tensor of shape (B, D). Owned by the caller of field(coords); the state stores a reference (no copy). It must be treated as **immutable** for the lifetime of the state: the SigmaCache memoises sigma^(n)(z) against the pre-activation z derived from these coords, so mutating coords (or z) in place silently desynchronises the cached derivatives from freshly recomputed values. In-place mutation of a torch z is caught by the cache (raises on the next read); rebuild the state from new coords rather than mutating in place. |
fieldkeyword-only | FieldBaseRequired | The FieldBase instance that produced this state. Ops read field.spec (activation), field.params, etc. |
componentskeyword-only | ComponentSpecRequired | ComponentSpec describing the output channels. Frozen. |
coordinate_speckeyword-only | CoordinateSpecRequired | CoordinateSpec describing the input axes. Frozen. |
opskeyword-only | ModuleTypeRequired | Backend-specific ops module (omnibias.fields.torch._ops_dispatch or its jax twin). Looked up by attribute access in the views. |
sigma_cachekeyword-only | SigmaCache[T]Required | SigmaCache keyed by derivative order. Filled lazily by whichever ops touch it first. |
extrakeyword-only | dict[str, Any] | NoneDefault: None | Free-form per-state cache: ops can stash intermediate tensors (like a precomputed gradient) keyed by string. Mutable but intended for op-internal use. |
Executed regression · cross backend bit parity
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_cross_backend_bit_parity(which, opfn):
nodes = _nodes(2 if which == "2d" else 3, seed=7)
ts = _state("torch", which, nodes)
js = _state("jax", which, nodes)
t = _np(opfn(ts.ops, ts))
j = _np(opfn(js.ops, js))
assert np.allclose(t, j, rtol=1e-12, atol=1e-12)python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_cross_backend_bit_parity[2d-<lambda>]" -qFields & methods
__init__method
__init__(self, *, coords: T, field: FieldBase, components: ComponentSpec, coordinate_spec: CoordinateSpec, ops: ModuleType, sigma_cache: SigmaCache[T], extra: dict[str, Any] | None=None) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
coordskeyword-only | TRequired | See the operation contract and implementation below. |
fieldkeyword-only | FieldBaseRequired | See the operation contract and implementation below. |
componentskeyword-only | ComponentSpecRequired | See the operation contract and implementation below. |
coordinate_speckeyword-only | CoordinateSpecRequired | See the operation contract and implementation below. |
opskeyword-only | ModuleTypeRequired | See the operation contract and implementation below. |
sigma_cachekeyword-only | SigmaCache[T]Required | See the operation contract and implementation below. |
extrakeyword-only | dict[str, Any] | NoneDefault: None | See the operation contract and implementation below. |
Returns: None
Validation & failure cases
- When
not isinstance(components, ComponentSpec), raisesTypeError(f'components must be ComponentSpec, got {type(components).__name__}'). - When
not isinstance(coordinate_spec, CoordinateSpec), raisesTypeError(f'coordinate_spec must be CoordinateSpec, got {type(coordinate_spec).__name__}'). - When
not isinstance(sigma_cache, SigmaCache), raisesTypeError(f'sigma_cache must be SigmaCache, got {type(sigma_cache).__name__}').
Executed regression · cross backend bit parity
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_cross_backend_bit_parity(which, opfn):
nodes = _nodes(2 if which == "2d" else 3, seed=7)
ts = _state("torch", which, nodes)
js = _state("jax", which, nodes)
t = _np(opfn(ts.ops, ts))
j = _np(opfn(js.ops, js))
assert np.allclose(t, j, rtol=1e-12, atol=1e-12)python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_cross_backend_bit_parity[2d-<lambda>]" -qMethod implementation
def __init__(
self,
*,
coords: T,
field: FieldBase,
components: ComponentSpec,
coordinate_spec: CoordinateSpec,
ops: ModuleType,
sigma_cache: SigmaCache[T],
extra: dict[str, Any] | None = None,
) -> None:
if not isinstance(components, ComponentSpec):
raise TypeError(
f"components must be ComponentSpec, got {type(components).__name__}"
)
if not isinstance(coordinate_spec, CoordinateSpec):
raise TypeError(
f"coordinate_spec must be CoordinateSpec, got {type(coordinate_spec).__name__}"
)
if not isinstance(sigma_cache, SigmaCache):
raise TypeError(
f"sigma_cache must be SigmaCache, got {type(sigma_cache).__name__}"
)
object.__setattr__(self, "coords", coords)
object.__setattr__(self, "field", field)
object.__setattr__(self, "components", components)
object.__setattr__(self, "coordinate_spec", coordinate_spec)
object.__setattr__(self, "ops", ops)
object.__setattr__(self, "sigma_cache", sigma_cache)
object.__setattr__(self, "extra", {} if extra is None else dict(extra))__setattr__method
__setattr__(self, name: str, value: Any) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
valuepositional-or-keyword | AnyRequired | See the operation contract and implementation below. |
Returns: None
Validation & failure cases
- When
name in self.__slots__, raisesAttributeError(f'FieldState is frozen; cannot reassign {name!r}. Use .extra for per-state op caches.').
Method implementation
def __setattr__(self, name: str, value: Any) -> None: # pragma: no cover
if name in self.__slots__:
raise AttributeError(
f"FieldState is frozen; cannot reassign {name!r}. Use .extra "
"for per-state op caches."
)
raise AttributeError(f"FieldState has no attribute {name!r}")__getattr__method
__getattr__(self, name: str) -> ComponentView | VectorViewSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
Returns: ComponentView | VectorView
Executed regression · integrate regression pin
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_integrate_regression_pin(make_torch_field): # type: ignore[no-untyped-def]
rule = _rule()
state = make_torch_field()(torch.as_tensor(rule.nodes, dtype=torch.float64))
# int_0^1 (1+2x+x^2) dx * int_-1^1 (1-y) dy = (1+1+1/3) * (2) = 14/3.
assert np.isclose(float(_np(state.u.integrate(rule=rule))), 14.0 / 3.0,
rtol=1e-12, atol=1e-14)python -m pytest "packages/omnibias-fields/tests/test_integration_norms.py::test_integrate_regression_pin" -qMethod implementation
def __getattr__(self, name: str) -> ComponentView | VectorView:
# __getattr__ is called only when ordinary lookup misses; the
# __slots__ attributes (coords, field, ...) are handled by the
# default descriptor protocol.
comps = self.components
if comps.is_component(name):
return ComponentView(self, name)
if comps.is_group(name):
return VectorView(self, comps.group_members(name))
# Provide a helpful message on typos.
raise AttributeError(did_you_mean(name, comps))__getitem__method
__getitem__(self, name: str) -> ComponentView | VectorViewSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
Returns: ComponentView | VectorView
Executed regression · integrate cross backend
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_integrate_cross_backend(make_torch_field, make_jax_field): # type: ignore[no-untyped-def]
rule = _rule()
tf, jf = make_torch_field(), make_jax_field()
ts = tf(torch.as_tensor(rule.nodes, dtype=torch.float64))
js = jf(jnp.asarray(rule.nodes, dtype=jnp.float64))
for name in ("u", "v"):
t = _np(ts[name].integrate(rule=rule))
j = _np(js[name].integrate(rule=rule))
assert np.allclose(t, j, rtol=1e-12, atol=1e-12), namepython -m pytest "packages/omnibias-fields/tests/test_integration_norms.py::test_integrate_cross_backend" -qMethod implementation
def __getitem__(self, name: str) -> ComponentView | VectorView:
return self.__getattr__(name)__contains__method
__contains__(self, name: object) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | objectRequired | See the operation contract and implementation below. |
Returns: bool
Method implementation
def __contains__(self, name: object) -> bool:
if not isinstance(name, str):
return False
return self.components.has(name)valuesproperty
values(self) -> dict[str, T]One-shot dict snapshot of every scalar component value.
Returns: dict[str, T]
Method implementation
def values(self) -> dict[str, T]:
"""One-shot dict snapshot of every scalar component value."""
return {n: self.ops.value(self, n) for n in self.components.names}axesproperty
axes(self) -> tuple[str, ...]See the class contract above and the exact method implementation below.
Returns: tuple[str, ...]
Method implementation
def axes(self) -> tuple[str, ...]:
return self.coordinate_spec.axesn_componentsproperty
n_components(self) -> intSee the class contract above and the exact method implementation below.
Returns: int
Method implementation
def n_components(self) -> int:
return self.components.n_components__repr__method
__repr__(self) -> strSee the class contract above and the exact method implementation below.
Returns: str
Method implementation
def __repr__(self) -> str:
cached = self.sigma_cache.orders()
try:
shape: Sequence[int] | str = tuple(self.coords.shape) # type: ignore[attr-defined]
except Exception: # pragma: no cover -- non-tensor coords
shape = "?"
return (
f"FieldState(field={type(self.field).__name__}, "
f"components={self.components.names!r}, "
f"coords_shape={shape}, "
f"cached_sigma_orders={cached})"
)Read the implementation
class FieldState(Generic[T]):
"""Frozen evaluation snapshot of a typed PINN field.
Parameters
----------
coords
Input tensor of shape ``(B, D)``. Owned by the caller of
``field(coords)``; the state stores a reference (no copy). It must be
treated as **immutable** for the lifetime of the state: the
:class:`SigmaCache` memoises ``sigma^(n)(z)`` against the pre-activation
``z`` derived from these coords, so mutating ``coords`` (or ``z``) in
place silently desynchronises the cached derivatives from freshly
recomputed values. In-place mutation of a torch ``z`` is caught by the
cache (raises on the next read); rebuild the state from new coords
rather than mutating in place.
field
The :class:`FieldBase` instance that produced this state. Ops
read ``field.spec`` (activation), ``field.params``, etc.
components
:class:`ComponentSpec` describing the output channels. Frozen.
coordinate_spec
:class:`CoordinateSpec` describing the input axes. Frozen.
ops
Backend-specific ``ops`` module (``omnibias.fields.torch._ops_dispatch``
or its jax twin). Looked up by attribute access in the views.
sigma_cache
:class:`SigmaCache` keyed by derivative order. Filled lazily by
whichever ops touch it first.
extra
Free-form per-state cache: ops can stash intermediate tensors
(like a precomputed gradient) keyed by string. Mutable but
intended for op-internal use.
**Readout-independence contract.** A :class:`FieldState` is a
coherent snapshot of one evaluation. Entries written by the field
or its ops may depend only on ``coords`` and *frozen feature*
parameters (hidden weights, temporal heads, geometry factors) --
never on the *readout* parameters the frozen-feature linear
solver sweeps (``c`` / ``b`` on a one-layer field, ``V`` / ``b_t``
on a spectral / Chebyshev field, the final affine layer of a jet
MLP). Fields that honour this contract declare it via the class
attribute named by
:data:`~omnibias.fields._core.field_base.READOUT_INDEPENDENT_ATTR`.
Caller-installed entries are the caller's responsibility. The
``lim_along`` extension, for example, stashes user-supplied
closures in ``extra["lim_along"]``; those closures are typically
readout-dependent and unverifiable by the library, so installing
them disqualifies the field from declaring readout-independence
for any subsequent frozen-feature solve against that state.
Notes
-----
The class behaves frozen-ish: the dataclass attributes are set in
``__init__`` and not changed afterwards. The :class:`SigmaCache`
contents and the ``extra`` dict are intentionally mutable so ops can
fill them lazily during a single residual evaluation.
"""
__slots__ = (
"coords",
"field",
"components",
"coordinate_spec",
"ops",
"sigma_cache",
"extra",
)
def __init__(
self,
*,
coords: T,
field: FieldBase,
components: ComponentSpec,
coordinate_spec: CoordinateSpec,
ops: ModuleType,
sigma_cache: SigmaCache[T],
extra: dict[str, Any] | None = None,
) -> None:
if not isinstance(components, ComponentSpec):
raise TypeError(
f"components must be ComponentSpec, got {type(components).__name__}"
)
if not isinstance(coordinate_spec, CoordinateSpec):
raise TypeError(
f"coordinate_spec must be CoordinateSpec, got {type(coordinate_spec).__name__}"
)
if not isinstance(sigma_cache, SigmaCache):
raise TypeError(
f"sigma_cache must be SigmaCache, got {type(sigma_cache).__name__}"
)
object.__setattr__(self, "coords", coords)
object.__setattr__(self, "field", field)
object.__setattr__(self, "components", components)
object.__setattr__(self, "coordinate_spec", coordinate_spec)
object.__setattr__(self, "ops", ops)
object.__setattr__(self, "sigma_cache", sigma_cache)
object.__setattr__(self, "extra", {} if extra is None else dict(extra))
def __setattr__(self, name: str, value: Any) -> None: # pragma: no cover
if name in self.__slots__:
raise AttributeError(
f"FieldState is frozen; cannot reassign {name!r}. Use .extra "
"for per-state op caches."
)
raise AttributeError(f"FieldState has no attribute {name!r}")
# -- attribute DSL ------------------------------------------------
def __getattr__(self, name: str) -> ComponentView | VectorView:
# __getattr__ is called only when ordinary lookup misses; the
# __slots__ attributes (coords, field, ...) are handled by the
# default descriptor protocol.
comps = self.components
if comps.is_component(name):
return ComponentView(self, name)
if comps.is_group(name):
return VectorView(self, comps.group_members(name))
# Provide a helpful message on typos.
raise AttributeError(did_you_mean(name, comps))
def __getitem__(self, name: str) -> ComponentView | VectorView:
return self.__getattr__(name)
def __contains__(self, name: object) -> bool:
if not isinstance(name, str):
return False
return self.components.has(name)
@property
def values(self) -> dict[str, T]:
"""One-shot dict snapshot of every scalar component value."""
return {n: self.ops.value(self, n) for n in self.components.names}
@property
def axes(self) -> tuple[str, ...]:
return self.coordinate_spec.axes
@property
def n_components(self) -> int:
return self.components.n_components
def __repr__(self) -> str:
cached = self.sigma_cache.orders()
try:
shape: Sequence[int] | str = tuple(self.coords.shape) # type: ignore[attr-defined]
except Exception: # pragma: no cover -- non-tensor coords
shape = "?"
return (
f"FieldState(field={type(self.field).__name__}, "
f"components={self.components.names!r}, "
f"coords_shape={shape}, "
f"cached_sigma_orders={cached})"
)ComponentView
omnibias.fields.ComponentViewSource ↗class ComponentView(Generic[T])Single-component view: state.u.
Each attribute forwards to a one-line kernel call. The view is frozen
by construction (we use __slots__) so it can be cached freely; in
practice it is just a (state, name) pair.
Parameters
| Name | Type & default | Contract |
|---|---|---|
statepositional-or-keyword | FieldStateRequired | See the operation contract and implementation below. |
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
Executed regression · integrate regression pin
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_integrate_regression_pin(make_torch_field): # type: ignore[no-untyped-def]
rule = _rule()
state = make_torch_field()(torch.as_tensor(rule.nodes, dtype=torch.float64))
# int_0^1 (1+2x+x^2) dx * int_-1^1 (1-y) dy = (1+1+1/3) * (2) = 14/3.
assert np.isclose(float(_np(state.u.integrate(rule=rule))), 14.0 / 3.0,
rtol=1e-12, atol=1e-14)python -m pytest "packages/omnibias-fields/tests/test_integration_norms.py::test_integrate_regression_pin" -qFields & methods
__init__method
__init__(self, state: FieldState, name: str) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
statepositional-or-keyword | FieldStateRequired | See the operation contract and implementation below. |
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
Returns: None
Validation & failure cases
- When
not state.components.is_component(name), raisesKeyError(f'{name!r} is not a scalar component of {state.components.names!r}').
Executed regression · integrate regression pin
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_integrate_regression_pin(make_torch_field): # type: ignore[no-untyped-def]
rule = _rule()
state = make_torch_field()(torch.as_tensor(rule.nodes, dtype=torch.float64))
# int_0^1 (1+2x+x^2) dx * int_-1^1 (1-y) dy = (1+1+1/3) * (2) = 14/3.
assert np.isclose(float(_np(state.u.integrate(rule=rule))), 14.0 / 3.0,
rtol=1e-12, atol=1e-14)python -m pytest "packages/omnibias-fields/tests/test_integration_norms.py::test_integrate_regression_pin" -qMethod implementation
def __init__(self, state: FieldState, name: str) -> None:
if not state.components.is_component(name):
raise KeyError(
f"{name!r} is not a scalar component of "
f"{state.components.names!r}"
)
object.__setattr__(self, "_state", state)
object.__setattr__(self, "_name", name)__setattr__method
__setattr__(self, key: str, value: Any) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
keypositional-or-keyword | strRequired | See the operation contract and implementation below. |
valuepositional-or-keyword | AnyRequired | See the operation contract and implementation below. |
Returns: None
Method implementation
def __setattr__(self, key: str, value: Any) -> None: # pragma: no cover
raise AttributeError("ComponentView is frozen")stateproperty
state(self) -> FieldStateSee the class contract above and the exact method implementation below.
Returns: FieldState
Method implementation
def state(self) -> FieldState:
return self._statenameproperty
name(self) -> strSee the class contract above and the exact method implementation below.
Returns: str
Method implementation
def name(self) -> str:
return self._namevalueproperty
value(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Executed regression · gradient theorem jax
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_gradient_theorem_jax(make_jax_field): # type: ignore[no-untyped-def]
field = make_jax_field()
curve = _jax_linear(_P0, _P1)
rule = gauss_legendre([(0.0, 1.0)], 8)
state = _jax_curve_points(field, curve, rule)
got = float(_np(state.u.line_integral(curve, rule=rule)))
exp = _endpoint_difference(field, _P0, _P1, "jax")
assert np.isclose(got, exp, rtol=1e-12, atol=1e-12)python -m pytest "packages/omnibias-fields/tests/test_line_integral.py::test_gradient_theorem_jax" -qMethod implementation
def value(self) -> T:
return self._state.ops.value(self._state, self._name)dtproperty
dt(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Validation & failure cases
- When
time_axis is None, raisesAttributeError(f'dt requires a time axis on the coordinate spec (got {self._state.coordinate_spec!r})').
Executed regression · dsl dt dx use coordinate spec axis names
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_dt_dx_use_coordinate_spec_axis_names():
"""``.dt`` / ``.dx`` route through ``coordinate_spec``, not hardcoded ``t``/``x``."""
axes = ("xi", "eta", "time")
polys = {
"u": (Poly((1.0, 1.0)), Poly((1.0, 0.0)), Poly((1.0,))),
"v": (Poly((0.0, 1.0)), Poly((1.0, -1.0)), Poly((1.0,))),
"c": (Poly((1.0, 2.0)), Poly((1.0, -1.0)), Poly((0.5, 0.0, 1.0))),
}
f = make_field("torch", axes, polys, time_axis="time")
rng = np.random.default_rng(0)
nodes = rng.standard_normal((5, 3))
st = f(torch.as_tensor(nodes, dtype=torch.float64))
assert np.allclose(
_np(st.c.dt),
_np(st.ops.derivative(st, "c", axis="time")),
rtol=0.0,
atol=0.0,
)
assert np.allclose(
_np(st.c.dx),
_np(st.ops.derivative(st, "c", axis="xi")),
rtol=0.0,
atol=0.0,
)
assert np.allclose(
_np(st.c.dy),
_np(st.ops.derivative(st, "c", axis="eta")),
rtol=0.0,
atol=0.0,
)
md = st.ops.material_derivative(st, velocity=("u", "v"), scalar="c")
assert md.shape == (5,)python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_dt_dx_use_coordinate_spec_axis_names" -qMethod implementation
def dt(self) -> T:
time_axis = self._state.coordinate_spec.time_axis
if time_axis is None:
raise AttributeError(
"dt requires a time axis on the coordinate spec "
f"(got {self._state.coordinate_spec!r})"
)
return self._state.ops.derivative(
self._state, self._name, axis=time_axis
)dxproperty
dx(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Executed regression · dsl dt dx use coordinate spec axis names
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_dt_dx_use_coordinate_spec_axis_names():
"""``.dt`` / ``.dx`` route through ``coordinate_spec``, not hardcoded ``t``/``x``."""
axes = ("xi", "eta", "time")
polys = {
"u": (Poly((1.0, 1.0)), Poly((1.0, 0.0)), Poly((1.0,))),
"v": (Poly((0.0, 1.0)), Poly((1.0, -1.0)), Poly((1.0,))),
"c": (Poly((1.0, 2.0)), Poly((1.0, -1.0)), Poly((0.5, 0.0, 1.0))),
}
f = make_field("torch", axes, polys, time_axis="time")
rng = np.random.default_rng(0)
nodes = rng.standard_normal((5, 3))
st = f(torch.as_tensor(nodes, dtype=torch.float64))
assert np.allclose(
_np(st.c.dt),
_np(st.ops.derivative(st, "c", axis="time")),
rtol=0.0,
atol=0.0,
)
assert np.allclose(
_np(st.c.dx),
_np(st.ops.derivative(st, "c", axis="xi")),
rtol=0.0,
atol=0.0,
)
assert np.allclose(
_np(st.c.dy),
_np(st.ops.derivative(st, "c", axis="eta")),
rtol=0.0,
atol=0.0,
)
md = st.ops.material_derivative(st, velocity=("u", "v"), scalar="c")
assert md.shape == (5,)python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_dt_dx_use_coordinate_spec_axis_names" -qMethod implementation
def dx(self) -> T:
return self._state.ops.derivative(
self._state, self._name, axis=self._spatial_axis(0, "dx")
)dyproperty
dy(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Executed regression · dsl dt dx use coordinate spec axis names
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_dt_dx_use_coordinate_spec_axis_names():
"""``.dt`` / ``.dx`` route through ``coordinate_spec``, not hardcoded ``t``/``x``."""
axes = ("xi", "eta", "time")
polys = {
"u": (Poly((1.0, 1.0)), Poly((1.0, 0.0)), Poly((1.0,))),
"v": (Poly((0.0, 1.0)), Poly((1.0, -1.0)), Poly((1.0,))),
"c": (Poly((1.0, 2.0)), Poly((1.0, -1.0)), Poly((0.5, 0.0, 1.0))),
}
f = make_field("torch", axes, polys, time_axis="time")
rng = np.random.default_rng(0)
nodes = rng.standard_normal((5, 3))
st = f(torch.as_tensor(nodes, dtype=torch.float64))
assert np.allclose(
_np(st.c.dt),
_np(st.ops.derivative(st, "c", axis="time")),
rtol=0.0,
atol=0.0,
)
assert np.allclose(
_np(st.c.dx),
_np(st.ops.derivative(st, "c", axis="xi")),
rtol=0.0,
atol=0.0,
)
assert np.allclose(
_np(st.c.dy),
_np(st.ops.derivative(st, "c", axis="eta")),
rtol=0.0,
atol=0.0,
)
md = st.ops.material_derivative(st, velocity=("u", "v"), scalar="c")
assert md.shape == (5,)python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_dt_dx_use_coordinate_spec_axis_names" -qMethod implementation
def dy(self) -> T:
return self._state.ops.derivative(
self._state, self._name, axis=self._spatial_axis(1, "dy")
)dzproperty
dz(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Method implementation
def dz(self) -> T:
return self._state.ops.derivative(
self._state, self._name, axis=self._spatial_axis(2, "dz")
)gradproperty
grad(self) -> TSpatial gradient grad u of shape (B, n_spatial).
**Spatial only**: the time axis (if any) is excluded, consistent with
lap and hess_spatial -- but *not* with hess.
Returns: T
Method implementation
def grad(self) -> T:
"""Spatial gradient ``grad u`` of shape ``(B, n_spatial)``.
**Spatial only**: the time axis (if any) is excluded, consistent with
:attr:`lap` and :attr:`hess_spatial` -- but *not* with :attr:`hess`.
"""
return self._state.ops.gradient(self._state, self._name)lapproperty
lap(self) -> TSpatial Laplacian sum_i d^2u/dx_i^2 over the spatial axes only
(the time axis is excluded).
Returns: T
Method implementation
def lap(self) -> T:
"""Spatial Laplacian ``sum_i d^2u/dx_i^2`` over the spatial axes only
(the time axis is excluded)."""
return self._state.ops.laplacian(self._state, self._name)hessproperty
hess(self) -> T**Full** Hessian over *all* coordinate axes, shape (B, D, D).
warning: Unlike grad and lap (which are spatial-only), this includes the **time axis** when the field has one, so on a spacetime field hess is (B, D, D) with the d^2/dt^2 / mixed d^2/dx dt entries present. Use hess_spatial for the spatial-only (B, n_spatial, n_spatial) Hessian that matches grad / lap.
Returns: T
Executed regression · dsl hess is full spacetime while hess spatial excludes time
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_hess_is_full_spacetime_while_hess_spatial_excludes_time():
"""``.hess`` spans all axes (incl. time); ``.hess_spatial`` excludes time.
Documents/guards the DSL footgun that ``.grad``/``.lap`` are spatial-only
but ``.hess`` is not."""
nodes = _nodes(3)
st = _state("torch", "st", nodes) # axes (x, y, t), time_axis="t"
hess = _np(st.rho.hess) # full (B, 3, 3)
hess_sp = _np(st.rho.hess_spatial) # spatial (B, 2, 2)
b = nodes.shape[0]
assert hess.shape == (b, 3, 3)
assert hess_sp.shape == (b, 2, 2)
# The (x, y) block of the full Hessian is exactly the spatial Hessian.
np.testing.assert_allclose(hess[:, :2, :2], hess_sp, rtol=1e-12, atol=1e-12)
# rho has a degree-2 time factor, so d^2/dt^2 is nonzero -> the full
# Hessian carries time information that hess_spatial drops.
assert np.max(np.abs(hess[:, 2, 2])) > 1e-6python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_hess_is_full_spacetime_while_hess_spatial_excludes_time" -qMethod implementation
def hess(self) -> T:
"""**Full** Hessian over *all* coordinate axes, shape ``(B, D, D)``.
.. warning::
Unlike :attr:`grad` and :attr:`lap` (which are spatial-only), this
includes the **time axis** when the field has one, so on a
spacetime field ``hess`` is ``(B, D, D)`` with the
``d^2/dt^2`` / mixed ``d^2/dx dt`` entries present. Use
:attr:`hess_spatial` for the spatial-only ``(B, n_spatial,
n_spatial)`` Hessian that matches :attr:`grad` / :attr:`lap`.
"""
return self._state.ops.hessian(self._state, self._name)hess_spatialproperty
hess_spatial(self) -> TSpatial-only Hessian, shape (B, n_spatial, n_spatial).
Excludes the time axis; this is the spatial counterpart to the
full-spacetime hess, and is axis-consistent with grad
and lap.
Returns: T
Executed regression · dsl hess is full spacetime while hess spatial excludes time
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_hess_is_full_spacetime_while_hess_spatial_excludes_time():
"""``.hess`` spans all axes (incl. time); ``.hess_spatial`` excludes time.
Documents/guards the DSL footgun that ``.grad``/``.lap`` are spatial-only
but ``.hess`` is not."""
nodes = _nodes(3)
st = _state("torch", "st", nodes) # axes (x, y, t), time_axis="t"
hess = _np(st.rho.hess) # full (B, 3, 3)
hess_sp = _np(st.rho.hess_spatial) # spatial (B, 2, 2)
b = nodes.shape[0]
assert hess.shape == (b, 3, 3)
assert hess_sp.shape == (b, 2, 2)
# The (x, y) block of the full Hessian is exactly the spatial Hessian.
np.testing.assert_allclose(hess[:, :2, :2], hess_sp, rtol=1e-12, atol=1e-12)
# rho has a degree-2 time factor, so d^2/dt^2 is nonzero -> the full
# Hessian carries time information that hess_spatial drops.
assert np.max(np.abs(hess[:, 2, 2])) > 1e-6python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_hess_is_full_spacetime_while_hess_spatial_excludes_time" -qMethod implementation
def hess_spatial(self) -> T:
"""Spatial-only Hessian, shape ``(B, n_spatial, n_spatial)``.
Excludes the time axis; this is the spatial counterpart to the
full-spacetime :attr:`hess`, and is axis-consistent with :attr:`grad`
and :attr:`lap`.
"""
return self._state.ops.spatial_hessian(self._state, self._name)biharmproperty
biharm(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Method implementation
def biharm(self) -> T:
return self._state.ops.biharmonic(self._state, self._name)grad_sqproperty
grad_sq(self) -> T|grad u|^2 (spatial).
Returns: T
Executed regression · dsl views match dispatch
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_views_match_dispatch():
nodes = _nodes(3)
st = _state("torch", "3d", nodes)
assert np.allclose(
_np(st.vel.grad_div), _np(st.ops.gradient_of_divergence(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.curl_curl), _np(st.ops.curl_of_curl(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.rate_of_rotation),
_np(st.ops.rate_of_rotation_tensor(st, ("u", "v", "w"))),
)
assert np.allclose(_np(st.c.grad_sq), _np(st.ops.grad_squared_norm(st, "c")))python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_views_match_dispatch" -qMethod implementation
def grad_sq(self) -> T:
"""``|grad u|^2`` (spatial)."""
return self._state.ops.grad_squared_norm(self._state, self._name)dalembertianmethod
dalembertian(self, *, c: float=1.0, signature: str='mostly_plus') -> Td'Alembert / wave operator box u (requires a time axis).
Parameters
| Name | Type & default | Contract |
|---|---|---|
ckeyword-only | floatDefault: 1.0 | See the operation contract and implementation below. |
signaturekeyword-only | strDefault: 'mostly_plus' | See the operation contract and implementation below. |
Returns: T
Method implementation
def dalembertian(self, *, c: float = 1.0, signature: str = "mostly_plus") -> T:
"""d'Alembert / wave operator ``box u`` (requires a time axis)."""
return self._state.ops.dalembertian(
self._state, self._name, c=c, signature=signature,
)diffusive_fluxmethod
diffusive_flux(self, *, diffusivity: float | str=1.0) -> TFickian flux F = -D grad u (shape (B, n_spatial)).
Parameters
| Name | Type & default | Contract |
|---|---|---|
diffusivitykeyword-only | float | strDefault: 1.0 | See the operation contract and implementation below. |
Returns: T
Method implementation
def diffusive_flux(self, *, diffusivity: float | str = 1.0) -> T:
"""Fickian flux ``F = -D grad u`` (shape ``(B, n_spatial)``)."""
return self._state.ops.diffusive_flux(
self._state, self._name, diffusivity=diffusivity,
)variable_diffusionmethod
variable_diffusion(self, *, diffusivity: float | str=1.0) -> Tdiv(D grad u) with constant or scalar-field D.
Parameters
| Name | Type & default | Contract |
|---|---|---|
diffusivitykeyword-only | float | strDefault: 1.0 | See the operation contract and implementation below. |
Returns: T
Method implementation
def variable_diffusion(self, *, diffusivity: float | str = 1.0) -> T:
"""``div(D grad u)`` with constant or scalar-field ``D``."""
return self._state.ops.variable_coefficient_diffusion(
self._state, self._name, diffusivity=diffusivity,
)dmethod
d(self, axis: int | str, order: int=1) -> TGeneric single-axis derivative escape hatch.
Parameters
| Name | Type & default | Contract |
|---|---|---|
axispositional-or-keyword | int | strRequired | See the operation contract and implementation below. |
orderpositional-or-keyword | intDefault: 1 | See the operation contract and implementation below. |
Returns: T
Method implementation
def d(self, axis: int | str, order: int = 1) -> T:
"""Generic single-axis derivative escape hatch."""
return self._state.ops.derivative(
self._state, self._name, axis=axis, order=order,
)dnmethod
dn(self, axes: Sequence[int | str], orders: Sequence[int]) -> TMixed partial: d^|orders| u / dx_{axes[0]}^{orders[0]} ....
Parameters
| Name | Type & default | Contract |
|---|---|---|
axespositional-or-keyword | Sequence[int | str]Required | See the operation contract and implementation below. |
orderspositional-or-keyword | Sequence[int]Required | See the operation contract and implementation below. |
Returns: T
Method implementation
def dn(
self,
axes: Sequence[int | str],
orders: Sequence[int],
) -> T:
"""Mixed partial: ``d^|orders| u / dx_{axes[0]}^{orders[0]} ...``."""
return self._state.ops.mixed_partial(
self._state, self._name, tuple(axes), tuple(orders),
)polylapmethod
polylap(self, k: int) -> TDelta^k u via the polylaplacian kernel.
Parameters
| Name | Type & default | Contract |
|---|---|---|
kpositional-or-keyword | intRequired | See the operation contract and implementation below. |
Returns: T
Method implementation
def polylap(self, k: int) -> T:
"""``Delta^k u`` via the polylaplacian kernel."""
return self._state.ops.polylaplacian(self._state, self._name, k=k)p_lapmethod
p_lap(self, p: float) -> Tdiv(|grad u|^{p-2} grad u) via the p-Laplacian op.
Parameters
| Name | Type & default | Contract |
|---|---|---|
ppositional-or-keyword | floatRequired | See the operation contract and implementation below. |
Returns: T
Method implementation
def p_lap(self, p: float) -> T:
"""``div(|grad u|^{p-2} grad u)`` via the p-Laplacian op."""
return self._state.ops.p_laplacian(self._state, self._name, p=p)directionalmethod
directional(self, direction: T) -> Td_direction u = grad u . direction.
Parameters
| Name | Type & default | Contract |
|---|---|---|
directionpositional-or-keyword | TRequired | See the operation contract and implementation below. |
Returns: T
Method implementation
def directional(self, direction: T) -> T:
"""``d_direction u = grad u . direction``."""
return self._state.ops.directional_derivative(
self._state, self._name, direction=direction,
)grad_of_dmethod
grad_of_d(self, axis: int | str) -> Tgrad(d u / dx_axis) over all coordinate axes.
Parameters
| Name | Type & default | Contract |
|---|---|---|
axispositional-or-keyword | int | strRequired | See the operation contract and implementation below. |
Returns: T
Method implementation
def grad_of_d(self, axis: int | str) -> T:
"""``grad(d u / dx_axis)`` over all coordinate axes."""
return self._state.ops.gradient_of_derivative(
self._state, self._name, axis=axis,
)integratemethod
integrate(self, *, rule: Any) -> TDefinite integral over the domain implied by rule (quadrature).
Parameters
| Name | Type & default | Contract |
|---|---|---|
rulekeyword-only | AnyRequired | See the operation contract and implementation below. |
Returns: T
Executed regression · integrate regression pin
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_integrate_regression_pin(make_torch_field): # type: ignore[no-untyped-def]
rule = _rule()
state = make_torch_field()(torch.as_tensor(rule.nodes, dtype=torch.float64))
# int_0^1 (1+2x+x^2) dx * int_-1^1 (1-y) dy = (1+1+1/3) * (2) = 14/3.
assert np.isclose(float(_np(state.u.integrate(rule=rule))), 14.0 / 3.0,
rtol=1e-12, atol=1e-14)python -m pytest "packages/omnibias-fields/tests/test_integration_norms.py::test_integrate_regression_pin" -qMethod implementation
def integrate(self, *, rule: Any) -> T:
"""Definite integral over the domain implied by ``rule`` (quadrature)."""
return self._state.ops.integrate(self._state, self._name, rule=rule)line_integralmethod
line_integral(self, curve: Any, *, rule: Any) -> TGradient-theorem line integral int_C grad u . dr along curve.
Equals u(curve(t1)) - u(curve(t0)) by the multivariate FTC; state
must be pre-evaluated at curve(quadrature_nodes(rule)).
Parameters
| Name | Type & default | Contract |
|---|---|---|
curvepositional-or-keyword | AnyRequired | See the operation contract and implementation below. |
rulekeyword-only | AnyRequired | See the operation contract and implementation below. |
Returns: T
Executed regression · gradient theorem jax
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_gradient_theorem_jax(make_jax_field): # type: ignore[no-untyped-def]
field = make_jax_field()
curve = _jax_linear(_P0, _P1)
rule = gauss_legendre([(0.0, 1.0)], 8)
state = _jax_curve_points(field, curve, rule)
got = float(_np(state.u.line_integral(curve, rule=rule)))
exp = _endpoint_difference(field, _P0, _P1, "jax")
assert np.isclose(got, exp, rtol=1e-12, atol=1e-12)python -m pytest "packages/omnibias-fields/tests/test_line_integral.py::test_gradient_theorem_jax" -qMethod implementation
def line_integral(self, curve: Any, *, rule: Any) -> T:
"""Gradient-theorem line integral ``int_C grad u . dr`` along ``curve``.
Equals ``u(curve(t1)) - u(curve(t0))`` by the multivariate FTC; ``state``
must be pre-evaluated at ``curve(quadrature_nodes(rule))``.
"""
return self._state.ops.line_integral(self._state, self._name, curve, rule=rule)l2_normmethod
l2_norm(self, *, rule: Any) -> T||u||_{L2} over the quadrature domain.
Parameters
| Name | Type & default | Contract |
|---|---|---|
rulekeyword-only | AnyRequired | See the operation contract and implementation below. |
Returns: T
Executed regression · l2 norm matches analytic
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_l2_norm_matches_analytic(make_torch_field): # type: ignore[no-untyped-def]
rule = gauss_legendre(BOUNDS, (8, 8)) # u^2 is degree 4 -> need >=3 nodes
field = make_torch_field()
state = field(torch.as_tensor(rule.nodes, dtype=torch.float64))
got = float(_np(state.u.l2_norm(rule=rule)))
# ||u||^2 = (int_0^1 (1+2x+x^2)^2 dx)(int_-1^1 (1-y)^2 dy).
# (1+2x+x^2)^2 = (x+1)^4 -> int_0^1 = (2^5-1)/5 = 31/5.
# (1-y)^2 -> int_-1^1 = 8/3.
exp = np.sqrt((31.0 / 5.0) * (8.0 / 3.0))
assert np.isclose(got, exp, rtol=1e-12, atol=1e-12)python -m pytest "packages/omnibias-fields/tests/test_integration_norms.py::test_l2_norm_matches_analytic" -qMethod implementation
def l2_norm(self, *, rule: Any) -> T:
"""``||u||_{L2}`` over the quadrature domain."""
return self._state.ops.l2_norm(self._state, self._name, rule=rule)sobolev_normmethod
sobolev_norm(self, *, rule: Any, k: int=1, weights: tuple[float, ...] | None=None) -> T||u||_{H^k} over the quadrature domain (k in {0, 1, 2}).
Parameters
| Name | Type & default | Contract |
|---|---|---|
rulekeyword-only | AnyRequired | See the operation contract and implementation below. |
kkeyword-only | intDefault: 1 | See the operation contract and implementation below. |
weightskeyword-only | tuple[float, ...] | NoneDefault: None | See the operation contract and implementation below. |
Returns: T
Executed regression · sobolev norm cross backend
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_sobolev_norm_cross_backend(make_torch_field, make_jax_field): # type: ignore[no-untyped-def]
rule = gauss_legendre(BOUNDS, (8, 8))
ts = make_torch_field()(torch.as_tensor(rule.nodes, dtype=torch.float64))
js = make_jax_field()(jnp.asarray(rule.nodes, dtype=jnp.float64))
for k in (0, 1, 2):
t = _np(ts.u.sobolev_norm(rule=rule, k=k))
j = _np(js.u.sobolev_norm(rule=rule, k=k))
assert np.allclose(t, j, rtol=1e-12, atol=1e-12), kpython -m pytest "packages/omnibias-fields/tests/test_integration_norms.py::test_sobolev_norm_cross_backend" -qMethod implementation
def sobolev_norm(
self, *, rule: Any, k: int = 1, weights: tuple[float, ...] | None = None,
) -> T:
"""``||u||_{H^k}`` over the quadrature domain (``k`` in ``{0, 1, 2}``)."""
return self._state.ops.sobolev_norm(
self._state, self._name, rule=rule, k=k, weights=weights,
)__getattr__method
__getattr__(self, attr: str) -> AnySee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
attrpositional-or-keyword | strRequired | See the operation contract and implementation below. |
Returns: Any
Validation & failure cases
- When
op is None, raisesAttributeError(f"ComponentView has no attribute {attr!r} and no op registered under that name. Available registered ops: {(tuple(self._state.ops.list_ops()) if hasattr(self._state.ops, 'list_ops') else ())}").
Method implementation
def __getattr__(self, attr: str) -> Any:
# Lookup in ops registry.
from omnibias.fields._core.ops_registry import lookup
op = lookup(attr)
if op is None:
raise AttributeError(
f"ComponentView has no attribute {attr!r} and no op "
f"registered under that name. Available registered ops: "
f"{tuple(self._state.ops.list_ops()) if hasattr(self._state.ops, 'list_ops') else ()}"
)
# Bind: registered ops are callables of (state, name, **kwargs);
# accessed as a property they should evaluate immediately, so we
# call op(state, name) and return the result.
return op(self._state, self._name)__repr__method
__repr__(self) -> strSee the class contract above and the exact method implementation below.
Returns: str
Method implementation
def __repr__(self) -> str:
return (
f"ComponentView({self._name!r}, "
f"field={type(self._state.field).__name__})"
)__eq__method
__eq__(self, other: object) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
otherpositional-or-keyword | objectRequired | See the operation contract and implementation below. |
Returns: bool
Method implementation
def __eq__(self, other: object) -> bool:
return (
isinstance(other, ComponentView)
and other._state is self._state
and other._name == self._name
)__hash__method
__hash__(self) -> intSee the class contract above and the exact method implementation below.
Returns: int
Method implementation
def __hash__(self) -> int:
return hash((id(self._state), self._name))Read the implementation
class ComponentView(Generic[T]):
"""Single-component view: ``state.u``.
Each attribute forwards to a one-line kernel call. The view is frozen
by construction (we use ``__slots__``) so it can be cached freely; in
practice it is just a (state, name) pair.
"""
__slots__ = ("_state", "_name")
def __init__(self, state: FieldState, name: str) -> None:
if not state.components.is_component(name):
raise KeyError(
f"{name!r} is not a scalar component of "
f"{state.components.names!r}"
)
object.__setattr__(self, "_state", state)
object.__setattr__(self, "_name", name)
def __setattr__(self, key: str, value: Any) -> None: # pragma: no cover
raise AttributeError("ComponentView is frozen")
@property
def state(self) -> FieldState:
return self._state
@property
def name(self) -> str:
return self._name
@property
def value(self) -> T:
return self._state.ops.value(self._state, self._name)
def _spatial_axis(self, index: int, attr: str) -> str:
axes = self._state.coordinate_spec.spatial_axes
if index >= len(axes):
raise AttributeError(
f"{attr} requires spatial axis index {index}, but "
f"coordinate_spec.spatial_axes={axes!r}"
)
return axes[index]
@property
def dt(self) -> T:
time_axis = self._state.coordinate_spec.time_axis
if time_axis is None:
raise AttributeError(
"dt requires a time axis on the coordinate spec "
f"(got {self._state.coordinate_spec!r})"
)
return self._state.ops.derivative(
self._state, self._name, axis=time_axis
)
@property
def dx(self) -> T:
return self._state.ops.derivative(
self._state, self._name, axis=self._spatial_axis(0, "dx")
)
@property
def dy(self) -> T:
return self._state.ops.derivative(
self._state, self._name, axis=self._spatial_axis(1, "dy")
)
@property
def dz(self) -> T:
return self._state.ops.derivative(
self._state, self._name, axis=self._spatial_axis(2, "dz")
)
@property
def grad(self) -> T:
"""Spatial gradient ``grad u`` of shape ``(B, n_spatial)``.
**Spatial only**: the time axis (if any) is excluded, consistent with
:attr:`lap` and :attr:`hess_spatial` -- but *not* with :attr:`hess`.
"""
return self._state.ops.gradient(self._state, self._name)
@property
def lap(self) -> T:
"""Spatial Laplacian ``sum_i d^2u/dx_i^2`` over the spatial axes only
(the time axis is excluded)."""
return self._state.ops.laplacian(self._state, self._name)
@property
def hess(self) -> T:
"""**Full** Hessian over *all* coordinate axes, shape ``(B, D, D)``.
.. warning::
Unlike :attr:`grad` and :attr:`lap` (which are spatial-only), this
includes the **time axis** when the field has one, so on a
spacetime field ``hess`` is ``(B, D, D)`` with the
``d^2/dt^2`` / mixed ``d^2/dx dt`` entries present. Use
:attr:`hess_spatial` for the spatial-only ``(B, n_spatial,
n_spatial)`` Hessian that matches :attr:`grad` / :attr:`lap`.
"""
return self._state.ops.hessian(self._state, self._name)
@property
def hess_spatial(self) -> T:
"""Spatial-only Hessian, shape ``(B, n_spatial, n_spatial)``.
Excludes the time axis; this is the spatial counterpart to the
full-spacetime :attr:`hess`, and is axis-consistent with :attr:`grad`
and :attr:`lap`.
"""
return self._state.ops.spatial_hessian(self._state, self._name)
@property
def biharm(self) -> T:
return self._state.ops.biharmonic(self._state, self._name)
@property
def grad_sq(self) -> T:
"""``|grad u|^2`` (spatial)."""
return self._state.ops.grad_squared_norm(self._state, self._name)
def dalembertian(self, *, c: float = 1.0, signature: str = "mostly_plus") -> T:
"""d'Alembert / wave operator ``box u`` (requires a time axis)."""
return self._state.ops.dalembertian(
self._state, self._name, c=c, signature=signature,
)
def diffusive_flux(self, *, diffusivity: float | str = 1.0) -> T:
"""Fickian flux ``F = -D grad u`` (shape ``(B, n_spatial)``)."""
return self._state.ops.diffusive_flux(
self._state, self._name, diffusivity=diffusivity,
)
def variable_diffusion(self, *, diffusivity: float | str = 1.0) -> T:
"""``div(D grad u)`` with constant or scalar-field ``D``."""
return self._state.ops.variable_coefficient_diffusion(
self._state, self._name, diffusivity=diffusivity,
)
def d(self, axis: int | str, order: int = 1) -> T:
"""Generic single-axis derivative escape hatch."""
return self._state.ops.derivative(
self._state, self._name, axis=axis, order=order,
)
def dn(
self,
axes: Sequence[int | str],
orders: Sequence[int],
) -> T:
"""Mixed partial: ``d^|orders| u / dx_{axes[0]}^{orders[0]} ...``."""
return self._state.ops.mixed_partial(
self._state, self._name, tuple(axes), tuple(orders),
)
def polylap(self, k: int) -> T:
"""``Delta^k u`` via the polylaplacian kernel."""
return self._state.ops.polylaplacian(self._state, self._name, k=k)
def p_lap(self, p: float) -> T:
"""``div(|grad u|^{p-2} grad u)`` via the p-Laplacian op."""
return self._state.ops.p_laplacian(self._state, self._name, p=p)
def directional(self, direction: T) -> T:
"""``d_direction u = grad u . direction``."""
return self._state.ops.directional_derivative(
self._state, self._name, direction=direction,
)
def grad_of_d(self, axis: int | str) -> T:
"""``grad(d u / dx_axis)`` over all coordinate axes."""
return self._state.ops.gradient_of_derivative(
self._state, self._name, axis=axis,
)
def integrate(self, *, rule: Any) -> T:
"""Definite integral over the domain implied by ``rule`` (quadrature)."""
return self._state.ops.integrate(self._state, self._name, rule=rule)
def line_integral(self, curve: Any, *, rule: Any) -> T:
"""Gradient-theorem line integral ``int_C grad u . dr`` along ``curve``.
Equals ``u(curve(t1)) - u(curve(t0))`` by the multivariate FTC; ``state``
must be pre-evaluated at ``curve(quadrature_nodes(rule))``.
"""
return self._state.ops.line_integral(self._state, self._name, curve, rule=rule)
def l2_norm(self, *, rule: Any) -> T:
"""``||u||_{L2}`` over the quadrature domain."""
return self._state.ops.l2_norm(self._state, self._name, rule=rule)
def sobolev_norm(
self, *, rule: Any, k: int = 1, weights: tuple[float, ...] | None = None,
) -> T:
"""``||u||_{H^k}`` over the quadrature domain (``k`` in ``{0, 1, 2}``)."""
return self._state.ops.sobolev_norm(
self._state, self._name, rule=rule, k=k, weights=weights,
)
# Allow registry-based extensions: ``state.u.symmetric_laplacian`` works
# if a third party registered "symmetric_laplacian" via ops_registry.
def __getattr__(self, attr: str) -> Any:
# Lookup in ops registry.
from omnibias.fields._core.ops_registry import lookup
op = lookup(attr)
if op is None:
raise AttributeError(
f"ComponentView has no attribute {attr!r} and no op "
f"registered under that name. Available registered ops: "
f"{tuple(self._state.ops.list_ops()) if hasattr(self._state.ops, 'list_ops') else ()}"
)
# Bind: registered ops are callables of (state, name, **kwargs);
# accessed as a property they should evaluate immediately, so we
# call op(state, name) and return the result.
return op(self._state, self._name)
def __repr__(self) -> str:
return (
f"ComponentView({self._name!r}, "
f"field={type(self._state.field).__name__})"
)
def __eq__(self, other: object) -> bool:
return (
isinstance(other, ComponentView)
and other._state is self._state
and other._name == self._name
)
def __hash__(self) -> int:
return hash((id(self._state), self._name))VectorView
omnibias.fields.VectorViewSource ↗class VectorView(Generic[T])Group view: state.velocity.
Behaves as a Sequence[ComponentView] (iteration, index, length),
plus the vector-level operator surface (curl, div, advect, ...). All
of these forward to one-line kernel calls.
Parameters
| Name | Type & default | Contract |
|---|---|---|
statepositional-or-keyword | FieldStateRequired | See the operation contract and implementation below. |
namespositional-or-keyword | Sequence[str]Required | See the operation contract and implementation below. |
Executed regression · dsl views match dispatch
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_views_match_dispatch():
nodes = _nodes(3)
st = _state("torch", "3d", nodes)
assert np.allclose(
_np(st.vel.grad_div), _np(st.ops.gradient_of_divergence(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.curl_curl), _np(st.ops.curl_of_curl(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.rate_of_rotation),
_np(st.ops.rate_of_rotation_tensor(st, ("u", "v", "w"))),
)
assert np.allclose(_np(st.c.grad_sq), _np(st.ops.grad_squared_norm(st, "c")))python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_views_match_dispatch" -qFields & methods
__init__method
__init__(self, state: FieldState, names: Sequence[str]) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
statepositional-or-keyword | FieldStateRequired | See the operation contract and implementation below. |
namespositional-or-keyword | Sequence[str]Required | See the operation contract and implementation below. |
Returns: None
Validation & failure cases
- When
len(normalised) == 0, raisesValueError('VectorView requires at least one component name'). - When
not state.components.is_component(n), raisesKeyError(f'{n!r} not in components {state.components.names!r}').
Executed regression · dsl views match dispatch
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_views_match_dispatch():
nodes = _nodes(3)
st = _state("torch", "3d", nodes)
assert np.allclose(
_np(st.vel.grad_div), _np(st.ops.gradient_of_divergence(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.curl_curl), _np(st.ops.curl_of_curl(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.rate_of_rotation),
_np(st.ops.rate_of_rotation_tensor(st, ("u", "v", "w"))),
)
assert np.allclose(_np(st.c.grad_sq), _np(st.ops.grad_squared_norm(st, "c")))python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_views_match_dispatch" -qMethod implementation
def __init__(self, state: FieldState, names: Sequence[str]) -> None:
normalised = tuple(names)
if len(normalised) == 0:
raise ValueError("VectorView requires at least one component name")
for n in normalised:
if not state.components.is_component(n):
raise KeyError(
f"{n!r} not in components {state.components.names!r}"
)
object.__setattr__(self, "_state", state)
object.__setattr__(self, "_names", normalised)__setattr__method
__setattr__(self, key: str, value: Any) -> NoneSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
keypositional-or-keyword | strRequired | See the operation contract and implementation below. |
valuepositional-or-keyword | AnyRequired | See the operation contract and implementation below. |
Returns: None
Method implementation
def __setattr__(self, key: str, value: Any) -> None: # pragma: no cover
raise AttributeError("VectorView is frozen")stateproperty
state(self) -> FieldStateSee the class contract above and the exact method implementation below.
Returns: FieldState
Method implementation
def state(self) -> FieldState:
return self._statenamesproperty
names(self) -> tuple[str, ...]See the class contract above and the exact method implementation below.
Returns: tuple[str, ...]
Method implementation
def names(self) -> tuple[str, ...]:
return self._names__len__method
__len__(self) -> intSee the class contract above and the exact method implementation below.
Returns: int
Method implementation
def __len__(self) -> int:
return len(self._names)__iter__method
__iter__(self) -> Iterator[ComponentView]See the class contract above and the exact method implementation below.
Returns: Iterator[ComponentView]
Method implementation
def __iter__(self) -> Iterator[ComponentView]:
return iter(ComponentView(self._state, n) for n in self._names)__getitem__method
__getitem__(self, key: int | str | slice) -> AnySee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
keypositional-or-keyword | int | str | sliceRequired | See the operation contract and implementation below. |
Returns: Any
Validation & failure cases
- When
key not in self._names, raisesKeyError(f'{key!r} not in vector view {self._names!r}').
Method implementation
def __getitem__(self, key: int | str | slice) -> Any:
if isinstance(key, int):
return ComponentView(self._state, self._names[key])
if isinstance(key, str):
if key not in self._names:
raise KeyError(
f"{key!r} not in vector view {self._names!r}"
)
return ComponentView(self._state, key)
if isinstance(key, slice):
return tuple(
ComponentView(self._state, n) for n in self._names[key]
)
raise TypeError(
f"VectorView index must be int, str, or slice; got {type(key).__name__}"
)valueproperty
value(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Method implementation
def value(self) -> T:
return self._state.ops.stack_components(self._state, self._names)jacproperty
jac(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Method implementation
def jac(self) -> T:
return self._state.ops.jacobian(self._state, self._names)spatial_jacproperty
spatial_jac(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Method implementation
def spatial_jac(self) -> T:
return self._state.ops.spatial_jacobian(self._state, self._names)divproperty
div(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Method implementation
def div(self) -> T:
return self._state.ops.divergence(self._state, self._names)curlproperty
curl(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Method implementation
def curl(self) -> T:
return self._state.ops.curl(self._state, self._names)vortproperty
vort(self) -> TAlias for curl (vorticity = curl of velocity).
Returns: T
Method implementation
def vort(self) -> T:
"""Alias for :attr:`curl` (vorticity = curl of velocity)."""
return self.curllapproperty
lap(self) -> TVector Laplacian: (Delta u_1, ..., Delta u_C).
Returns: T
Method implementation
def lap(self) -> T:
"""Vector Laplacian: ``(Delta u_1, ..., Delta u_C)``."""
return self._state.ops.vector_laplacian(self._state, self._names)biharmproperty
biharm(self) -> TVector biharmonic: (Delta^2 u_1, ..., Delta^2 u_C).
Returns: T
Method implementation
def biharm(self) -> T:
"""Vector biharmonic: ``(Delta^2 u_1, ..., Delta^2 u_C)``."""
return self._state.ops.vector_biharmonic(self._state, self._names)hessproperty
hess(self) -> TStacked component Hessians.
Returns: T
Method implementation
def hess(self) -> T:
"""Stacked component Hessians."""
return self._state.ops.vector_hessian(self._state, self._names)strain_rateproperty
strain_rate(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Method implementation
def strain_rate(self) -> T:
return self._state.ops.strain_rate(self._state, self._names)rate_of_rotationproperty
rate_of_rotation(self) -> TAntisymmetric spin tensor W = 0.5(J - J^T) (complements strain_rate).
Returns: T
Executed regression · dsl views match dispatch
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_views_match_dispatch():
nodes = _nodes(3)
st = _state("torch", "3d", nodes)
assert np.allclose(
_np(st.vel.grad_div), _np(st.ops.gradient_of_divergence(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.curl_curl), _np(st.ops.curl_of_curl(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.rate_of_rotation),
_np(st.ops.rate_of_rotation_tensor(st, ("u", "v", "w"))),
)
assert np.allclose(_np(st.c.grad_sq), _np(st.ops.grad_squared_norm(st, "c")))python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_views_match_dispatch" -qMethod implementation
def rate_of_rotation(self) -> T:
"""Antisymmetric spin tensor ``W = 0.5(J - J^T)`` (complements ``strain_rate``)."""
return self._state.ops.rate_of_rotation_tensor(self._state, self._names)grad_divproperty
grad_div(self) -> Tgrad(div u).
Returns: T
Executed regression · dsl views match dispatch
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_views_match_dispatch():
nodes = _nodes(3)
st = _state("torch", "3d", nodes)
assert np.allclose(
_np(st.vel.grad_div), _np(st.ops.gradient_of_divergence(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.curl_curl), _np(st.ops.curl_of_curl(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.rate_of_rotation),
_np(st.ops.rate_of_rotation_tensor(st, ("u", "v", "w"))),
)
assert np.allclose(_np(st.c.grad_sq), _np(st.ops.grad_squared_norm(st, "c")))python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_views_match_dispatch" -qMethod implementation
def grad_div(self) -> T:
"""``grad(div u)``."""
return self._state.ops.gradient_of_divergence(self._state, self._names)curl_curlproperty
curl_curl(self) -> Tcurl(curl u) (2D / 3D).
Returns: T
Executed regression · dsl views match dispatch
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_dsl_views_match_dispatch():
nodes = _nodes(3)
st = _state("torch", "3d", nodes)
assert np.allclose(
_np(st.vel.grad_div), _np(st.ops.gradient_of_divergence(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.curl_curl), _np(st.ops.curl_of_curl(st, ("u", "v", "w")))
)
assert np.allclose(
_np(st.vel.rate_of_rotation),
_np(st.ops.rate_of_rotation_tensor(st, ("u", "v", "w"))),
)
assert np.allclose(_np(st.c.grad_sq), _np(st.ops.grad_squared_norm(st, "c")))python -m pytest "packages/omnibias-fields/tests/test_vector_calculus.py::test_dsl_views_match_dispatch" -qMethod implementation
def curl_curl(self) -> T:
"""``curl(curl u)`` (2D / 3D)."""
return self._state.ops.curl_of_curl(self._state, self._names)deformation_gradientproperty
deformation_gradient(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Method implementation
def deformation_gradient(self) -> T:
return self._state.ops.deformation_gradient(self._state, self._names)dtproperty
dt(self) -> TSee the class contract above and the exact method implementation below.
Returns: T
Validation & failure cases
- When
time_axis is None, raisesAttributeError(f'dt requires a time axis on the coordinate spec (got {self._state.coordinate_spec!r})').
Method implementation
def dt(self) -> T:
time_axis = self._state.coordinate_spec.time_axis
if time_axis is None:
raise AttributeError(
"dt requires a time axis on the coordinate spec "
f"(got {self._state.coordinate_spec!r})"
)
return self._state.ops.vector_derivative(
self._state, self._names, axis=time_axis,
)advectmethod
advect(self) -> TSelf-advection: (u . nabla) u.
Returns: T
Method implementation
def advect(self) -> T:
"""Self-advection: ``(u . nabla) u``. """
return self._state.ops.advection(self._state, velocity=self._names)advect_bymethod
advect_by(self, other: VectorView) -> T(other . nabla) self. Use for cross-advection of e.g. scalars.
Parameters
| Name | Type & default | Contract |
|---|---|---|
otherpositional-or-keyword | VectorViewRequired | See the operation contract and implementation below. |
Returns: T
Validation & failure cases
- When
not isinstance(other, VectorView), raisesTypeError(f'advect_by argument must be a VectorView, got {type(other).__name__}'). - When
other.state is not self._state, raisesValueError('VectorView.advect_by: both views must share the same FieldState').
Method implementation
def advect_by(self, other: VectorView) -> T:
"""``(other . nabla) self``. Use for cross-advection of e.g. scalars."""
if not isinstance(other, VectorView):
raise TypeError(
f"advect_by argument must be a VectorView, got {type(other).__name__}"
)
if other.state is not self._state:
raise ValueError(
"VectorView.advect_by: both views must share the same FieldState"
)
return self._state.ops.advection(
self._state, velocity=other._names, target=self._names,
)material_derivativemethod
material_derivative(self) -> TD/Dt = d/dt + (u . nabla); material derivative.
Returns: T
Method implementation
def material_derivative(self) -> T:
"""``D/Dt = d/dt + (u . nabla)``; material derivative."""
return self._state.ops.material_derivative(
self._state, velocity=self._names,
)skew_advectmethod
skew_advect(self, scalar: str | None=None) -> TSkew-symmetric (energy/enstrophy-conserving) advection.
Parameters
| Name | Type & default | Contract |
|---|---|---|
scalarpositional-or-keyword | str | NoneDefault: None | See the operation contract and implementation below. |
Returns: T
Method implementation
def skew_advect(self, scalar: str | None = None) -> T:
"""Skew-symmetric (energy/enstrophy-conserving) advection."""
return self._state.ops.skew_symmetric_advection(
self._state, velocity=self._names, scalar=scalar,
)polylapmethod
polylap(self, k: int) -> TVector polylaplacian: (Delta^k u_1, ..., Delta^k u_C).
Parameters
| Name | Type & default | Contract |
|---|---|---|
kpositional-or-keyword | intRequired | See the operation contract and implementation below. |
Returns: T
Method implementation
def polylap(self, k: int) -> T:
"""Vector polylaplacian: ``(Delta^k u_1, ..., Delta^k u_C)``."""
return self._state.ops.vector_polylaplacian(self._state, self._names, k=k)__repr__method
__repr__(self) -> strSee the class contract above and the exact method implementation below.
Returns: str
Method implementation
def __repr__(self) -> str:
return (
f"VectorView({list(self._names)!r}, "
f"field={type(self._state.field).__name__})"
)__eq__method
__eq__(self, other: object) -> boolSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
otherpositional-or-keyword | objectRequired | See the operation contract and implementation below. |
Returns: bool
Method implementation
def __eq__(self, other: object) -> bool:
return (
isinstance(other, VectorView)
and other._state is self._state
and other._names == self._names
)__hash__method
__hash__(self) -> intSee the class contract above and the exact method implementation below.
Returns: int
Method implementation
def __hash__(self) -> int:
return hash((id(self._state), self._names))Read the implementation
class VectorView(Generic[T]):
"""Group view: ``state.velocity``.
Behaves as a ``Sequence[ComponentView]`` (iteration, index, length),
plus the vector-level operator surface (curl, div, advect, ...). All
of these forward to one-line kernel calls.
"""
__slots__ = ("_state", "_names")
def __init__(self, state: FieldState, names: Sequence[str]) -> None:
normalised = tuple(names)
if len(normalised) == 0:
raise ValueError("VectorView requires at least one component name")
for n in normalised:
if not state.components.is_component(n):
raise KeyError(
f"{n!r} not in components {state.components.names!r}"
)
object.__setattr__(self, "_state", state)
object.__setattr__(self, "_names", normalised)
def __setattr__(self, key: str, value: Any) -> None: # pragma: no cover
raise AttributeError("VectorView is frozen")
@property
def state(self) -> FieldState:
return self._state
@property
def names(self) -> tuple[str, ...]:
return self._names
def __len__(self) -> int:
return len(self._names)
def __iter__(self) -> Iterator[ComponentView]:
return iter(ComponentView(self._state, n) for n in self._names)
def __getitem__(self, key: int | str | slice) -> Any:
if isinstance(key, int):
return ComponentView(self._state, self._names[key])
if isinstance(key, str):
if key not in self._names:
raise KeyError(
f"{key!r} not in vector view {self._names!r}"
)
return ComponentView(self._state, key)
if isinstance(key, slice):
return tuple(
ComponentView(self._state, n) for n in self._names[key]
)
raise TypeError(
f"VectorView index must be int, str, or slice; got {type(key).__name__}"
)
@property
def value(self) -> T:
return self._state.ops.stack_components(self._state, self._names)
@property
def jac(self) -> T:
return self._state.ops.jacobian(self._state, self._names)
@property
def spatial_jac(self) -> T:
return self._state.ops.spatial_jacobian(self._state, self._names)
@property
def div(self) -> T:
return self._state.ops.divergence(self._state, self._names)
@property
def curl(self) -> T:
return self._state.ops.curl(self._state, self._names)
@property
def vort(self) -> T:
"""Alias for :attr:`curl` (vorticity = curl of velocity)."""
return self.curl
@property
def lap(self) -> T:
"""Vector Laplacian: ``(Delta u_1, ..., Delta u_C)``."""
return self._state.ops.vector_laplacian(self._state, self._names)
@property
def biharm(self) -> T:
"""Vector biharmonic: ``(Delta^2 u_1, ..., Delta^2 u_C)``."""
return self._state.ops.vector_biharmonic(self._state, self._names)
@property
def hess(self) -> T:
"""Stacked component Hessians."""
return self._state.ops.vector_hessian(self._state, self._names)
@property
def strain_rate(self) -> T:
return self._state.ops.strain_rate(self._state, self._names)
@property
def rate_of_rotation(self) -> T:
"""Antisymmetric spin tensor ``W = 0.5(J - J^T)`` (complements ``strain_rate``)."""
return self._state.ops.rate_of_rotation_tensor(self._state, self._names)
@property
def grad_div(self) -> T:
"""``grad(div u)``."""
return self._state.ops.gradient_of_divergence(self._state, self._names)
@property
def curl_curl(self) -> T:
"""``curl(curl u)`` (2D / 3D)."""
return self._state.ops.curl_of_curl(self._state, self._names)
@property
def deformation_gradient(self) -> T:
return self._state.ops.deformation_gradient(self._state, self._names)
@property
def dt(self) -> T:
time_axis = self._state.coordinate_spec.time_axis
if time_axis is None:
raise AttributeError(
"dt requires a time axis on the coordinate spec "
f"(got {self._state.coordinate_spec!r})"
)
return self._state.ops.vector_derivative(
self._state, self._names, axis=time_axis,
)
def advect(self) -> T:
"""Self-advection: ``(u . nabla) u``. """
return self._state.ops.advection(self._state, velocity=self._names)
def advect_by(self, other: VectorView) -> T:
"""``(other . nabla) self``. Use for cross-advection of e.g. scalars."""
if not isinstance(other, VectorView):
raise TypeError(
f"advect_by argument must be a VectorView, got {type(other).__name__}"
)
if other.state is not self._state:
raise ValueError(
"VectorView.advect_by: both views must share the same FieldState"
)
return self._state.ops.advection(
self._state, velocity=other._names, target=self._names,
)
def material_derivative(self) -> T:
"""``D/Dt = d/dt + (u . nabla)``; material derivative."""
return self._state.ops.material_derivative(
self._state, velocity=self._names,
)
def skew_advect(self, scalar: str | None = None) -> T:
"""Skew-symmetric (energy/enstrophy-conserving) advection."""
return self._state.ops.skew_symmetric_advection(
self._state, velocity=self._names, scalar=scalar,
)
def polylap(self, k: int) -> T:
"""Vector polylaplacian: ``(Delta^k u_1, ..., Delta^k u_C)``."""
return self._state.ops.vector_polylaplacian(self._state, self._names, k=k)
def __repr__(self) -> str:
return (
f"VectorView({list(self._names)!r}, "
f"field={type(self._state.field).__name__})"
)
def __eq__(self, other: object) -> bool:
return (
isinstance(other, VectorView)
and other._state is self._state
and other._names == self._names
)
def __hash__(self) -> int:
return hash((id(self._state), self._names))did_you_mean
omnibias.fields.did_you_meanSource ↗did_you_mean(name: str, components: ComponentSpec) -> strBuild a friendly AttributeError message for typos.
Parameters
| Name | Type & default | Contract |
|---|---|---|
namepositional-or-keyword | strRequired | See the operation contract and implementation below. |
componentspositional-or-keyword | ComponentSpecRequired | See the operation contract and implementation below. |
strRead the implementation
def did_you_mean(name: str, components: ComponentSpec) -> str:
"""Build a friendly :class:`AttributeError` message for typos."""
available_components = components.names
available_groups = tuple(g for g, _ in components.groups)
candidates = available_components + available_groups
suggestions = difflib.get_close_matches(name, candidates, n=1, cutoff=0.6)
base = (
f"FieldState has no component or group {name!r}. "
f"Available components: {available_components!r}"
)
if available_groups:
base += f" or groups: {available_groups!r}"
if suggestions:
base += f". Did you mean {suggestions[0]!r}?"
else:
base += "."
return baseTestFunctionSpace
omnibias.fields.TestFunctionSpaceSource ↗class TestFunctionSpaceDataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.
Bank of OMBU bumps used as Petrov-Galerkin test functions.
Parameters
| Name | Type & default | Contract |
|---|---|---|
bankdataclass field | BankSpecRequired | See the operation contract and implementation below. |
ordersdataclass field | tuple[int, ...]Required | See the operation contract and implementation below. |
basedataclass field | strDefault: 'tanh' | See the operation contract and implementation below. |
windowdataclass field | tuple[float, float] | NoneDefault: None | See the operation contract and implementation below. |
Executed regression · g2 boundary bound changes loss
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_g2_boundary_bound_changes_loss() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,), source=(2.0,))
field = (0.0, 1.0, -1.0) # x - x^2
on = float(weak_loss(field, space, operator=op, include_boundary_bound=True))
off = float(weak_loss(field, space, operator=op, include_boundary_bound=False))
assert on != off
bound = boundary_bound(space, deriv_bound=1.0)
assert bound.hi > 0.0python -m pytest "packages/omnibias-fields/tests/test_weak.py::test_g2_boundary_bound_changes_loss" -qRegression example · path recording quadrature for callable source
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_path_recording_quadrature_for_callable_source() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,))
quad = gauss_legendre(((0.0, 1.0),), 8)
def source(x: torch.Tensor) -> torch.Tensor:
return torch.sin(math.pi * x.reshape(-1)).reshape(x.shape)
_res, terms = weak_residual(
(0.0, 1.0, -1.0), space, operator=op, source=source, quadrature=quad
)
load_paths = {t.path for t in terms if t.name.startswith("load")}
stiff_paths = {t.path for t in terms if t.name.startswith("stiffness")}
assert "quadrature" in load_paths
assert "exact" in stiff_pathspython -m pytest packages/omnibias-fields/tests/test_weak.py -k test_path_recording_quadrature_for_callable_source -qFields & methods
bankfield
bank: BankSpecDeclared BankSpec field on TestFunctionSpace. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.
View source ↗ordersfield
orders: tuple[int, ...]Declared tuple[int, ...] field on TestFunctionSpace. A value must be provided when constructing the configuration, unless an inherited constructor supplies it.
View source ↗basefield
base: str = 'tanh'Declared str field on TestFunctionSpace. Default expression: 'tanh'.
windowfield
window: tuple[float, float] | None = NoneDeclared tuple[float, float] | None field on TestFunctionSpace. Default expression: None.
__post_init__method
__post_init__(self) -> NoneSee the class contract above and the exact method implementation below.
Returns: None
Validation & failure cases
- When
not self.orders, raisesValueError('orders must be non-empty'). - When
any((int(n) < 0 for n in self.orders)), raisesValueError('orders must be >= 0'). - When
not hi > lo, raisesValueError(f'window needs lo < hi, got {self.window}').
Executed regression · g2 boundary bound changes loss
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_g2_boundary_bound_changes_loss() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,), source=(2.0,))
field = (0.0, 1.0, -1.0) # x - x^2
on = float(weak_loss(field, space, operator=op, include_boundary_bound=True))
off = float(weak_loss(field, space, operator=op, include_boundary_bound=False))
assert on != off
bound = boundary_bound(space, deriv_bound=1.0)
assert bound.hi > 0.0python -m pytest "packages/omnibias-fields/tests/test_weak.py::test_g2_boundary_bound_changes_loss" -qMethod implementation
def __post_init__(self) -> None:
if not self.orders:
raise ValueError("orders must be non-empty")
if any(int(n) < 0 for n in self.orders):
raise ValueError("orders must be >= 0")
if self.window is not None:
lo, hi = self.window
if not (hi > lo):
raise ValueError(f"window needs lo < hi, got {self.window}")sizeproperty
size(self) -> intSee the class contract above and the exact method implementation below.
Returns: int
Executed regression · g2 boundary bound changes loss
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_g2_boundary_bound_changes_loss() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,), source=(2.0,))
field = (0.0, 1.0, -1.0) # x - x^2
on = float(weak_loss(field, space, operator=op, include_boundary_bound=True))
off = float(weak_loss(field, space, operator=op, include_boundary_bound=False))
assert on != off
bound = boundary_bound(space, deriv_bound=1.0)
assert bound.hi > 0.0python -m pytest "packages/omnibias-fields/tests/test_weak.py::test_g2_boundary_bound_changes_loss" -qMethod implementation
def size(self) -> int:
return int(self.bank.n_offsets)order_atmethod
order_at(self, index: int) -> intSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
indexpositional-or-keyword | intRequired | See the operation contract and implementation below. |
Returns: int
Validation & failure cases
- When
index < 0 or index >= self.size, raisesIndexError(f'test index {index} out of range [0, {self.size})'). - When
len(self.orders) != self.size, raisesValueError('orders must be length 1 or match the bank').
Executed regression · g2 boundary bound changes loss
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_g2_boundary_bound_changes_loss() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,), source=(2.0,))
field = (0.0, 1.0, -1.0) # x - x^2
on = float(weak_loss(field, space, operator=op, include_boundary_bound=True))
off = float(weak_loss(field, space, operator=op, include_boundary_bound=False))
assert on != off
bound = boundary_bound(space, deriv_bound=1.0)
assert bound.hi > 0.0python -m pytest "packages/omnibias-fields/tests/test_weak.py::test_g2_boundary_bound_changes_loss" -qMethod implementation
def order_at(self, index: int) -> int:
if index < 0 or index >= self.size:
raise IndexError(f"test index {index} out of range [0, {self.size})")
if len(self.orders) == 1:
return int(self.orders[0])
if len(self.orders) != self.size:
raise ValueError("orders must be length 1 or match the bank")
return int(self.orders[index])scale_atmethod
scale_at(self, index: int) -> floatSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
indexpositional-or-keyword | intRequired | See the operation contract and implementation below. |
Returns: float
Executed regression · g2 boundary bound changes loss
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_g2_boundary_bound_changes_loss() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,), source=(2.0,))
field = (0.0, 1.0, -1.0) # x - x^2
on = float(weak_loss(field, space, operator=op, include_boundary_bound=True))
off = float(weak_loss(field, space, operator=op, include_boundary_bound=False))
assert on != off
bound = boundary_bound(space, deriv_bound=1.0)
assert bound.hi > 0.0python -m pytest "packages/omnibias-fields/tests/test_weak.py::test_g2_boundary_bound_changes_loss" -qMethod implementation
def scale_at(self, index: int) -> float:
scales = self.bank.scales
if len(scales) == 1:
return float(scales[0])
return float(scales[index])offset_atmethod
offset_at(self, index: int) -> floatSee the class contract above and the exact method implementation below.
Parameters
| Name | Type & default | Contract |
|---|---|---|
indexpositional-or-keyword | intRequired | See the operation contract and implementation below. |
Returns: float
Executed regression · g2 boundary bound changes loss
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_g2_boundary_bound_changes_loss() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,), source=(2.0,))
field = (0.0, 1.0, -1.0) # x - x^2
on = float(weak_loss(field, space, operator=op, include_boundary_bound=True))
off = float(weak_loss(field, space, operator=op, include_boundary_bound=False))
assert on != off
bound = boundary_bound(space, deriv_bound=1.0)
assert bound.hi > 0.0python -m pytest "packages/omnibias-fields/tests/test_weak.py::test_g2_boundary_bound_changes_loss" -qMethod implementation
def offset_at(self, index: int) -> float:
return float(self.bank.offsets[index])Read the implementation
class TestFunctionSpace:
"""Bank of OMBU bumps used as Petrov-Galerkin test functions."""
__test__: ClassVar[bool] = False
bank: BankSpec
orders: tuple[int, ...]
base: str = "tanh"
window: tuple[float, float] | None = None
def __post_init__(self) -> None:
if not self.orders:
raise ValueError("orders must be non-empty")
if any(int(n) < 0 for n in self.orders):
raise ValueError("orders must be >= 0")
if self.window is not None:
lo, hi = self.window
if not (hi > lo):
raise ValueError(f"window needs lo < hi, got {self.window}")
@property
def size(self) -> int:
return int(self.bank.n_offsets)
def order_at(self, index: int) -> int:
if index < 0 or index >= self.size:
raise IndexError(f"test index {index} out of range [0, {self.size})")
if len(self.orders) == 1:
return int(self.orders[0])
if len(self.orders) != self.size:
raise ValueError("orders must be length 1 or match the bank")
return int(self.orders[index])
def scale_at(self, index: int) -> float:
scales = self.bank.scales
if len(scales) == 1:
return float(scales[0])
return float(scales[index])
def offset_at(self, index: int) -> float:
return float(self.bank.offsets[index])Other public import paths
omnibias.fields.TestFunctionSpaceomnibias.fields.weak.TestFunctionSpace
WeakForm
omnibias.fields.WeakFormSource ↗class WeakFormDataclass constructor fields; inherited fields, if any, are declared by the base class. A field(...) expression describes a default factory or field options, not a literal argument default.
1-D Poisson-type form int a u' v' - int f v on a box.
diffusion and source are polynomial coefficients (low degree first).
Exact assembly requires polynomial data on a box window.
Parameters
| Name | Type & default | Contract |
|---|---|---|
kinddataclass field | Literal['poisson1d']Default: 'poisson1d' | See the operation contract and implementation below. |
diffusiondataclass field | tuple[float, ...]Default: (1.0,) | See the operation contract and implementation below. |
sourcedataclass field | tuple[float, ...] | NoneDefault: None | See the operation contract and implementation below. |
Regression example · g2 boundary bound changes loss
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_g2_boundary_bound_changes_loss() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,), source=(2.0,))
field = (0.0, 1.0, -1.0) # x - x^2
on = float(weak_loss(field, space, operator=op, include_boundary_bound=True))
off = float(weak_loss(field, space, operator=op, include_boundary_bound=False))
assert on != off
bound = boundary_bound(space, deriv_bound=1.0)
assert bound.hi > 0.0python -m pytest packages/omnibias-fields/tests/test_weak.py -k test_g2_boundary_bound_changes_loss -qRegression example · path recording quadrature for callable source
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_path_recording_quadrature_for_callable_source() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,))
quad = gauss_legendre(((0.0, 1.0),), 8)
def source(x: torch.Tensor) -> torch.Tensor:
return torch.sin(math.pi * x.reshape(-1)).reshape(x.shape)
_res, terms = weak_residual(
(0.0, 1.0, -1.0), space, operator=op, source=source, quadrature=quad
)
load_paths = {t.path for t in terms if t.name.startswith("load")}
stiff_paths = {t.path for t in terms if t.name.startswith("stiffness")}
assert "quadrature" in load_paths
assert "exact" in stiff_pathspython -m pytest packages/omnibias-fields/tests/test_weak.py -k test_path_recording_quadrature_for_callable_source -qFields & methods
kindfield
kind: Literal['poisson1d'] = 'poisson1d'Declared Literal['poisson1d'] field on WeakForm. Default expression: 'poisson1d'.
diffusionfield
diffusion: tuple[float, ...] = (1.0,)Declared tuple[float, ...] field on WeakForm. Default expression: (1.0,).
sourcefield
source: tuple[float, ...] | None = NoneDeclared tuple[float, ...] | None field on WeakForm. Default expression: None.
Read the implementation
class WeakForm:
"""1-D Poisson-type form ``int a u' v' - int f v`` on a box.
``diffusion`` and ``source`` are polynomial coefficients (low degree first).
Exact assembly requires polynomial data on a box window.
"""
kind: Literal["poisson1d"] = "poisson1d"
diffusion: tuple[float, ...] = (1.0,)
source: tuple[float, ...] | None = NoneOther public import paths
omnibias.fields.WeakFormomnibias.fields.weak.WeakForm
boundary_bound
omnibias.fields.boundary_boundSource ↗boundary_bound(space: TestFunctionSpace, *, deriv_bound: float) -> IntervalCertified tail bound on the dropped boundary term (01-05).
Analytic bases are not compactly supported. The bound is
2 * |deriv_bound| * tail_bound of a unit-scale mollifier at the
window half-width, never assumed zero.
Parameters
| Name | Type & default | Contract |
|---|---|---|
spacepositional-or-keyword | TestFunctionSpaceRequired | See the operation contract and implementation below. |
deriv_boundkeyword-only | floatRequired | See the operation contract and implementation below. |
IntervalExplicit validation & failure cases
- When
space.window is None, raisesValueError('boundary_bound needs a box window').
These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.
Executed regression · g2 boundary bound changes loss
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_g2_boundary_bound_changes_loss() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,), source=(2.0,))
field = (0.0, 1.0, -1.0) # x - x^2
on = float(weak_loss(field, space, operator=op, include_boundary_bound=True))
off = float(weak_loss(field, space, operator=op, include_boundary_bound=False))
assert on != off
bound = boundary_bound(space, deriv_bound=1.0)
assert bound.hi > 0.0python -m pytest "packages/omnibias-fields/tests/test_weak.py::test_g2_boundary_bound_changes_loss" -qRead the implementation
def boundary_bound(space: TestFunctionSpace, *, deriv_bound: float) -> Interval:
"""Certified tail bound on the dropped boundary term (01-05).
Analytic bases are not compactly supported. The bound is
``2 * |deriv_bound| * tail_bound`` of a unit-scale mollifier at the
window half-width, never assumed zero.
"""
if space.window is None:
raise ValueError("boundary_bound needs a box window")
lo, hi = space.window
half = 0.5 * (hi - lo)
alpha = min(space.scale_at(i) for i in range(space.size))
spec = MollifierSpec(
base=space.base,
scale=1.0 / float(alpha),
packs=(PackSpec(order=0, mean=0.0, weight=1.0),),
)
tail = tail_bound(spec, half_width=half)
two = Interval.from_rational(2)
return two * tail.abs() * Interval.point(abs(float(deriv_bound)))Other public import paths
omnibias.fields.boundary_boundomnibias.fields.weak.boundary_bound
exact_moment
omnibias.fields.exact_momentSource ↗exact_moment(space: TestFunctionSpace, power: int, index: int) -> floatintegral_window x^{power} v_index(x) dx by antiderivative differences.
Requires order >= power so integration by parts stays inside the
closed-form tower (no leftover int x^k sigma). For power >= 1
the recursive term is scaled by 1/alpha because the primitive of
sigma^{(n)}(alpha (x-mu)) is sigma^{(n-1)} / alpha.
Parameters
| Name | Type & default | Contract |
|---|---|---|
spacepositional-or-keyword | TestFunctionSpaceRequired | See the operation contract and implementation below. |
powerpositional-or-keyword | intRequired | See the operation contract and implementation below. |
indexpositional-or-keyword | intRequired | See the operation contract and implementation below. |
floatExplicit validation & failure cases
- When
space.window is None, raisesValueError('exact_moment needs a box window'). - When
j < 0, raisesValueError(f'power must be >= 0, got {power}'). - When
n < j, raisesValueError(f'exact_moment needs test order >= power (got order={n}, power={j})').
These are explicit source guards. Backend operations may also raise for invalid shapes, dtypes, or unsupported devices.
Executed regression · g2 boundary bound changes loss
Recorded coverage confirms this test executed the API body. This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_g2_boundary_bound_changes_loss() -> None:
torch.set_default_dtype(torch.float64)
space = _space()
op = WeakForm(diffusion=(1.0,), source=(2.0,))
field = (0.0, 1.0, -1.0) # x - x^2
on = float(weak_loss(field, space, operator=op, include_boundary_bound=True))
off = float(weak_loss(field, space, operator=op, include_boundary_bound=False))
assert on != off
bound = boundary_bound(space, deriv_bound=1.0)
assert bound.hi > 0.0python -m pytest "packages/omnibias-fields/tests/test_weak.py::test_g2_boundary_bound_changes_loss" -qRegression example · g1 exact moment vs gauss
This example belongs to the pinned upstream test suite. It may use fixtures and imports from its full file. Run it from that checkout with its test dependencies installed.
def test_g1_exact_moment_vs_gauss() -> None:
space = TestFunctionSpace(
BankSpec.uniform(0.3, 0.7, 3, scales=(2.0,)),
orders=(2,),
base="tanh",
window=(0.0, 1.0),
)
exact = exact_moment(space, 0, 0)
hi = gauss_legendre(((0.0, 1.0),), 48)
lo = gauss_legendre(((0.0, 1.0),), 2)
v_hi = np.array([eval_test(space, 0, float(x), deriv=0) for x in hi.nodes[:, 0]])
v_lo = np.array([eval_test(space, 0, float(x), deriv=0) for x in lo.nodes[:, 0]])
g_hi = float(np.dot(hi.weights, v_hi))
g_lo = float(np.dot(lo.weights, v_lo))
rel = abs(exact - g_hi) / max(abs(g_hi), 1e-16)
assert rel <= 1e-13
assert abs(g_lo - exact) > abs(g_hi - exact)python -m pytest packages/omnibias-fields/tests/test_weak.py -k test_g1_exact_moment_vs_gauss -qRead the implementation
def exact_moment(space: TestFunctionSpace, power: int, index: int) -> float:
"""``integral_window x^{power} v_index(x) dx`` by antiderivative differences.
Requires ``order >= power`` so integration by parts stays inside the
closed-form tower (no leftover ``int x^k sigma``). For ``power >= 1``
the recursive term is scaled by ``1/alpha`` because the primitive of
``sigma^{(n)}(alpha (x-mu))`` is ``sigma^{(n-1)} / alpha``.
"""
if space.window is None:
raise ValueError("exact_moment needs a box window")
p, q = space.window
n = space.order_at(index)
j = int(power)
if j < 0:
raise ValueError(f"power must be >= 0, got {power}")
if n < j:
raise ValueError(
f"exact_moment needs test order >= power (got order={n}, power={j})"
)
alpha = space.scale_at(index)
mu = space.offset_at(index)
return _moment_ibp(space.base, p, q, j, n, alpha, mu)Other public import paths
omnibias.fields.exact_momentomnibias.fields.weak.exact_moment