Safe Custom Equations

Safe custom-equation records define the JSON-safe specification and expression validation layer for future user-defined physics-informed mean functions.

The current API supports:

• strict JSON-safe equation specifications,
• allowlisted expression parsing with Python ast and no arbitrary execution,
• NumPy preview evaluation with evaluate_safe_equation_numpy(...),
• finite/non-finite output reporting,
• physics-mean and optional target-residual previews with preview_safe_equation_mean(...),
• torch evaluation with evaluate_safe_equation_torch(...),
• differentiable parameter initialization with initialize_safe_equation_torch_parameters(...),
• trainable PI-GPR mean-function adapters with build_safe_equation_mean_function(...).

Use this layer before model fitting to check that a proposed physics mean has declared variables, finite defaults, supported functions, and finite outputs on sample data.

from matgpr import (
    SafeEquationParameter,
    SafeEquationSpec,
    SafeEquationVariable,
    preview_safe_equation_mean,
)

spec = SafeEquationSpec(
    name="custom_power_law",
    expression="coefficient * pow(x, exponent)",
    variables=[SafeEquationVariable("x", units="a.u.")],
    parameters=[
        SafeEquationParameter("coefficient", initial_value=2.0),
        SafeEquationParameter("exponent", initial_value=0.5),
    ],
)

preview = preview_safe_equation_mean(
    spec,
    {"x": [1.0, 4.0, 9.0]},
    target_values=[2.1, 3.8, 6.2],
)
preview.to_dict()

Use the torch evaluator when gradient flow is required:

import torch

from matgpr import (
    build_safe_equation_mean_function,
    evaluate_safe_equation_torch,
    initialize_safe_equation_torch_parameters,
)

parameters = initialize_safe_equation_torch_parameters(spec)
mean = evaluate_safe_equation_torch(
    spec,
    {"x": torch.tensor([1.0, 4.0, 9.0], dtype=torch.float64)},
    parameter_values=parameters,
)
mean.sum().backward()

Build a trainable PI-GPR mean adapter when fitting with GPyTorch:

mean_module = build_safe_equation_mean_function(
    spec,
    feature_indices={"x": 0},
)

The returned SafeEquationMeanFunction is compatible with fit_gpytorch_gpr(..., mean_module=mean_module). It uses the spec's learnable flags to split parameters into trainable and fixed physics parameters, stores adapter metadata with to_dict(), and evaluates the safe equation through the torch interpreter during training.

matgpr.safe_equations

SafeEquationVariable dataclass

One declared input variable for a safe custom equation.

Source code in matgpr/safe_equations.py

@dataclass(frozen=True)
class SafeEquationVariable:
    """One declared input variable for a safe custom equation."""

    name: str
    description: str = ""
    units: str | None = None
    required: bool = True
    bounds: tuple[float, float] | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "name", _validate_symbol_name(self.name, "variable name"))
        object.__setattr__(self, "description", str(self.description))
        object.__setattr__(self, "units", _optional_string(self.units, "variable units"))
        object.__setattr__(self, "required", bool(self.required))
        object.__setattr__(self, "bounds", _validate_bounds(self.bounds, "variable bounds"))

    @classmethod
    def from_dict(cls, record: Mapping[str, Any]) -> "SafeEquationVariable":
        """Build a variable record from a JSON-like mapping."""
        return cls(
            name=_required(record, "name", "SafeEquationVariable"),
            description=record.get("description", ""),
            units=record.get("units"),
            required=record.get("required", True),
            bounds=record.get("bounds"),
        )

    def to_dict(self) -> dict[str, object]:
        """Return a JSON-safe variable record."""
        return {
            "name": self.name,
            "description": self.description,
            "units": self.units,
            "required": self.required,
            "bounds": _bounds_to_list(self.bounds),
        }

from_dict(record) classmethod

Build a variable record from a JSON-like mapping.

Source code in matgpr/safe_equations.py

@classmethod
def from_dict(cls, record: Mapping[str, Any]) -> "SafeEquationVariable":
    """Build a variable record from a JSON-like mapping."""
    return cls(
        name=_required(record, "name", "SafeEquationVariable"),
        description=record.get("description", ""),
        units=record.get("units"),
        required=record.get("required", True),
        bounds=record.get("bounds"),
    )

to_dict()

Return a JSON-safe variable record.

Source code in matgpr/safe_equations.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-safe variable record."""
    return {
        "name": self.name,
        "description": self.description,
        "units": self.units,
        "required": self.required,
        "bounds": _bounds_to_list(self.bounds),
    }

SafeEquationParameter dataclass

One declared equation parameter.

initial_value is the starting value for learnable parameters and the default value for fixed parameters when downstream tools choose to keep a parameter fixed.

Source code in matgpr/safe_equations.py

@dataclass(frozen=True)
class SafeEquationParameter:
    """One declared equation parameter.

    ``initial_value`` is the starting value for learnable parameters and the
    default value for fixed parameters when downstream tools choose to keep a
    parameter fixed.
    """

    name: str
    initial_value: float
    bounds: tuple[float, float] | None = None
    description: str = ""
    units: str | None = None
    learnable: bool = True

    def __post_init__(self) -> None:
        initial_value = _finite_float(self.initial_value, "parameter initial_value")
        bounds = _validate_bounds(self.bounds, "parameter bounds")
        if bounds is not None and not bounds[0] <= initial_value <= bounds[1]:
            raise ValueError("parameter initial_value must lie within bounds")

        object.__setattr__(self, "name", _validate_symbol_name(self.name, "parameter name"))
        object.__setattr__(self, "initial_value", initial_value)
        object.__setattr__(self, "bounds", bounds)
        object.__setattr__(self, "description", str(self.description))
        object.__setattr__(self, "units", _optional_string(self.units, "parameter units"))
        object.__setattr__(self, "learnable", bool(self.learnable))

    @classmethod
    def from_dict(cls, record: Mapping[str, Any]) -> "SafeEquationParameter":
        """Build a parameter record from a JSON-like mapping."""
        return cls(
            name=_required(record, "name", "SafeEquationParameter"),
            initial_value=_required(record, "initial_value", "SafeEquationParameter"),
            bounds=record.get("bounds"),
            description=record.get("description", ""),
            units=record.get("units"),
            learnable=record.get("learnable", True),
        )

    def to_dict(self) -> dict[str, object]:
        """Return a JSON-safe parameter record."""
        return {
            "name": self.name,
            "initial_value": self.initial_value,
            "bounds": _bounds_to_list(self.bounds),
            "description": self.description,
            "units": self.units,
            "learnable": self.learnable,
        }

from_dict(record) classmethod

Build a parameter record from a JSON-like mapping.

Source code in matgpr/safe_equations.py

@classmethod
def from_dict(cls, record: Mapping[str, Any]) -> "SafeEquationParameter":
    """Build a parameter record from a JSON-like mapping."""
    return cls(
        name=_required(record, "name", "SafeEquationParameter"),
        initial_value=_required(record, "initial_value", "SafeEquationParameter"),
        bounds=record.get("bounds"),
        description=record.get("description", ""),
        units=record.get("units"),
        learnable=record.get("learnable", True),
    )

to_dict()

Return a JSON-safe parameter record.

Source code in matgpr/safe_equations.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-safe parameter record."""
    return {
        "name": self.name,
        "initial_value": self.initial_value,
        "bounds": _bounds_to_list(self.bounds),
        "description": self.description,
        "units": self.units,
        "learnable": self.learnable,
    }

SafeEquationConstant dataclass

One declared finite constant used by a safe custom equation.

Source code in matgpr/safe_equations.py

@dataclass(frozen=True)
class SafeEquationConstant:
    """One declared finite constant used by a safe custom equation."""

    name: str
    value: float
    description: str = ""
    units: str | None = None

    def __post_init__(self) -> None:
        object.__setattr__(self, "name", _validate_symbol_name(self.name, "constant name"))
        object.__setattr__(self, "value", _finite_float(self.value, "constant value"))
        object.__setattr__(self, "description", str(self.description))
        object.__setattr__(self, "units", _optional_string(self.units, "constant units"))

    @classmethod
    def from_dict(cls, record: Mapping[str, Any]) -> "SafeEquationConstant":
        """Build a constant record from a JSON-like mapping."""
        return cls(
            name=_required(record, "name", "SafeEquationConstant"),
            value=_required(record, "value", "SafeEquationConstant"),
            description=record.get("description", ""),
            units=record.get("units"),
        )

    def to_dict(self) -> dict[str, object]:
        """Return a JSON-safe constant record."""
        return {
            "name": self.name,
            "value": self.value,
            "description": self.description,
            "units": self.units,
        }

from_dict(record) classmethod

Build a constant record from a JSON-like mapping.

Source code in matgpr/safe_equations.py

@classmethod
def from_dict(cls, record: Mapping[str, Any]) -> "SafeEquationConstant":
    """Build a constant record from a JSON-like mapping."""
    return cls(
        name=_required(record, "name", "SafeEquationConstant"),
        value=_required(record, "value", "SafeEquationConstant"),
        description=record.get("description", ""),
        units=record.get("units"),
    )

to_dict()

Return a JSON-safe constant record.

Source code in matgpr/safe_equations.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-safe constant record."""
    return {
        "name": self.name,
        "value": self.value,
        "description": self.description,
        "units": self.units,
    }

SafeEquationSpec dataclass

JSON-safe custom equation specification for future PI-GPR mean functions.

Source code in matgpr/safe_equations.py

@dataclass(frozen=True)
class SafeEquationSpec:
    """JSON-safe custom equation specification for future PI-GPR mean functions."""

    name: str
    expression: str
    variables: Sequence[SafeEquationVariable | Mapping[str, Any]]
    parameters: Sequence[SafeEquationParameter | Mapping[str, Any]] = ()
    constants: Sequence[SafeEquationConstant | Mapping[str, Any]] = ()
    display_name: str = ""
    description: str = ""
    target_units: str | None = None
    assumptions: Sequence[str] = ()
    validity_limits: Mapping[str, Any] = field(default_factory=dict)
    references: Sequence[str] = ()
    metadata: Mapping[str, Any] = field(default_factory=dict)
    schema_version: str = SAFE_EQUATION_SCHEMA_VERSION

    def __post_init__(self) -> None:
        name = _validate_symbol_name(self.name, "spec name")
        expression = str(self.expression).strip()
        if not expression:
            raise ValueError("expression must be non-empty")

        variables = tuple(_coerce_variable(variable) for variable in self.variables)
        parameters = tuple(_coerce_parameter(parameter) for parameter in self.parameters)
        constants = tuple(_coerce_constant(constant) for constant in self.constants)
        if not variables:
            raise ValueError("variables must contain at least one variable")
        _validate_unique_symbols(variables, parameters, constants)

        object.__setattr__(self, "name", name)
        object.__setattr__(self, "expression", expression)
        object.__setattr__(self, "variables", variables)
        object.__setattr__(self, "parameters", parameters)
        object.__setattr__(self, "constants", constants)
        object.__setattr__(self, "display_name", str(self.display_name))
        object.__setattr__(self, "description", str(self.description))
        object.__setattr__(
            self, "target_units", _optional_string(self.target_units, "target_units")
        )
        object.__setattr__(self, "assumptions", _string_tuple(self.assumptions, "assumptions"))
        object.__setattr__(
            self,
            "validity_limits",
            _json_safe_mapping(self.validity_limits, "validity_limits"),
        )
        object.__setattr__(self, "references", _string_tuple(self.references, "references"))
        object.__setattr__(self, "metadata", _json_safe_mapping(self.metadata, "metadata"))
        object.__setattr__(self, "schema_version", str(self.schema_version))

    @classmethod
    def from_dict(cls, record: Mapping[str, Any]) -> "SafeEquationSpec":
        """Build a safe equation spec from a JSON-like mapping."""
        return cls(
            name=_required(record, "name", "SafeEquationSpec"),
            expression=_required(record, "expression", "SafeEquationSpec"),
            variables=record.get("variables", ()),
            parameters=record.get("parameters", ()),
            constants=record.get("constants", ()),
            display_name=record.get("display_name", ""),
            description=record.get("description", ""),
            target_units=record.get("target_units"),
            assumptions=record.get("assumptions", ()),
            validity_limits=record.get("validity_limits", {}),
            references=record.get("references", ()),
            metadata=record.get("metadata", {}),
            schema_version=record.get("schema_version", SAFE_EQUATION_SCHEMA_VERSION),
        )

    @property
    def variable_names(self) -> tuple[str, ...]:
        """Declared input-variable names."""
        return tuple(variable.name for variable in self.variables)

    @property
    def parameter_names(self) -> tuple[str, ...]:
        """Declared parameter names."""
        return tuple(parameter.name for parameter in self.parameters)

    @property
    def constant_names(self) -> tuple[str, ...]:
        """Declared constant names."""
        return tuple(constant.name for constant in self.constants)

    @property
    def declared_symbols(self) -> tuple[str, ...]:
        """All declared variables, parameters, and constants."""
        return (*self.variable_names, *self.parameter_names, *self.constant_names)

    def to_dict(self) -> dict[str, object]:
        """Return a strict-JSON-safe equation specification."""
        return {
            "schema_version": self.schema_version,
            "name": self.name,
            "display_name": self.display_name,
            "expression": self.expression,
            "variables": [variable.to_dict() for variable in self.variables],
            "parameters": [parameter.to_dict() for parameter in self.parameters],
            "constants": [constant.to_dict() for constant in self.constants],
            "description": self.description,
            "target_units": self.target_units,
            "assumptions": list(self.assumptions),
            "validity_limits": dict(self.validity_limits),
            "references": list(self.references),
            "metadata": dict(self.metadata),
            "declared_symbols": list(self.declared_symbols),
            "allowed_functions": list(ALLOWED_SAFE_EQUATION_FUNCTIONS),
        }

variable_names property

Declared input-variable names.

parameter_names property

Declared parameter names.

constant_names property

Declared constant names.

declared_symbols property

All declared variables, parameters, and constants.

from_dict(record) classmethod

Build a safe equation spec from a JSON-like mapping.

Source code in matgpr/safe_equations.py

@classmethod
def from_dict(cls, record: Mapping[str, Any]) -> "SafeEquationSpec":
    """Build a safe equation spec from a JSON-like mapping."""
    return cls(
        name=_required(record, "name", "SafeEquationSpec"),
        expression=_required(record, "expression", "SafeEquationSpec"),
        variables=record.get("variables", ()),
        parameters=record.get("parameters", ()),
        constants=record.get("constants", ()),
        display_name=record.get("display_name", ""),
        description=record.get("description", ""),
        target_units=record.get("target_units"),
        assumptions=record.get("assumptions", ()),
        validity_limits=record.get("validity_limits", {}),
        references=record.get("references", ()),
        metadata=record.get("metadata", {}),
        schema_version=record.get("schema_version", SAFE_EQUATION_SCHEMA_VERSION),
    )

to_dict()

Return a strict-JSON-safe equation specification.

Source code in matgpr/safe_equations.py

def to_dict(self) -> dict[str, object]:
    """Return a strict-JSON-safe equation specification."""
    return {
        "schema_version": self.schema_version,
        "name": self.name,
        "display_name": self.display_name,
        "expression": self.expression,
        "variables": [variable.to_dict() for variable in self.variables],
        "parameters": [parameter.to_dict() for parameter in self.parameters],
        "constants": [constant.to_dict() for constant in self.constants],
        "description": self.description,
        "target_units": self.target_units,
        "assumptions": list(self.assumptions),
        "validity_limits": dict(self.validity_limits),
        "references": list(self.references),
        "metadata": dict(self.metadata),
        "declared_symbols": list(self.declared_symbols),
        "allowed_functions": list(ALLOWED_SAFE_EQUATION_FUNCTIONS),
    }

SafeEquationValidationIssue dataclass

Structured validation issue for a safe equation specification.

Source code in matgpr/safe_equations.py

@dataclass(frozen=True)
class SafeEquationValidationIssue:
    """Structured validation issue for a safe equation specification."""

    code: str
    message: str
    field: str | None = None
    severity: str = "error"

    def __post_init__(self) -> None:
        if not str(self.code).strip():
            raise ValueError("issue code must be non-empty")
        if self.severity not in {"error", "warning"}:
            raise ValueError("issue severity must be 'error' or 'warning'")
        object.__setattr__(self, "code", str(self.code))
        object.__setattr__(self, "message", str(self.message))
        object.__setattr__(self, "field", _optional_string(self.field, "issue field"))

    def to_dict(self) -> dict[str, object]:
        """Return a JSON-safe validation issue."""
        return {
            "code": self.code,
            "message": self.message,
            "field": self.field,
            "severity": self.severity,
        }

to_dict()

Return a JSON-safe validation issue.

Source code in matgpr/safe_equations.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-safe validation issue."""
    return {
        "code": self.code,
        "message": self.message,
        "field": self.field,
        "severity": self.severity,
    }

SafeEquationValidationResult dataclass

Validation result for a safe equation specification.

Source code in matgpr/safe_equations.py

@dataclass(frozen=True)
class SafeEquationValidationResult:
    """Validation result for a safe equation specification."""

    issues: Sequence[SafeEquationValidationIssue | Mapping[str, Any]] = ()
    spec_name: str | None = None
    declared_symbols: Sequence[str] = ()
    expression_symbols: Sequence[str] = ()
    expression_functions: Sequence[str] = ()
    allowed_functions: Sequence[str] = ALLOWED_SAFE_EQUATION_FUNCTIONS
    allowed_operators: Sequence[str] = ALLOWED_SAFE_EQUATION_OPERATORS
    schema_version: str = SAFE_EQUATION_SCHEMA_VERSION

    def __post_init__(self) -> None:
        issues = tuple(_coerce_issue(issue) for issue in self.issues)
        object.__setattr__(self, "issues", issues)
        object.__setattr__(self, "spec_name", _optional_string(self.spec_name, "spec_name"))
        object.__setattr__(
            self,
            "declared_symbols",
            _string_tuple(self.declared_symbols, "declared_symbols"),
        )
        object.__setattr__(
            self,
            "expression_symbols",
            _string_tuple(self.expression_symbols, "expression_symbols"),
        )
        object.__setattr__(
            self,
            "expression_functions",
            _string_tuple(self.expression_functions, "expression_functions"),
        )
        object.__setattr__(
            self,
            "allowed_functions",
            _string_tuple(self.allowed_functions, "allowed_functions"),
        )
        object.__setattr__(
            self,
            "allowed_operators",
            _string_tuple(self.allowed_operators, "allowed_operators"),
        )
        object.__setattr__(self, "schema_version", str(self.schema_version))

    @property
    def is_valid(self) -> bool:
        """Whether the result contains no error-severity issues."""
        return not any(issue.severity == "error" for issue in self.issues)

    @property
    def errors(self) -> tuple[SafeEquationValidationIssue, ...]:
        """Error-severity issues."""
        return tuple(issue for issue in self.issues if issue.severity == "error")

    @property
    def warnings(self) -> tuple[SafeEquationValidationIssue, ...]:
        """Warning-severity issues."""
        return tuple(issue for issue in self.issues if issue.severity == "warning")

    def to_dict(self) -> dict[str, object]:
        """Return a JSON-safe validation result."""
        return {
            "schema_version": self.schema_version,
            "is_valid": self.is_valid,
            "spec_name": self.spec_name,
            "declared_symbols": list(self.declared_symbols),
            "expression_symbols": list(self.expression_symbols),
            "expression_functions": list(self.expression_functions),
            "allowed_functions": list(self.allowed_functions),
            "allowed_operators": list(self.allowed_operators),
            "issues": [issue.to_dict() for issue in self.issues],
            "errors": [issue.to_dict() for issue in self.errors],
            "warnings": [issue.to_dict() for issue in self.warnings],
        }

is_valid property

Whether the result contains no error-severity issues.

errors property

Error-severity issues.

warnings property

Warning-severity issues.

to_dict()

Return a JSON-safe validation result.

Source code in matgpr/safe_equations.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-safe validation result."""
    return {
        "schema_version": self.schema_version,
        "is_valid": self.is_valid,
        "spec_name": self.spec_name,
        "declared_symbols": list(self.declared_symbols),
        "expression_symbols": list(self.expression_symbols),
        "expression_functions": list(self.expression_functions),
        "allowed_functions": list(self.allowed_functions),
        "allowed_operators": list(self.allowed_operators),
        "issues": [issue.to_dict() for issue in self.issues],
        "errors": [issue.to_dict() for issue in self.errors],
        "warnings": [issue.to_dict() for issue in self.warnings],
    }

SafeEquationEvaluationResult dataclass

JSON-safe shell for future safe-equation evaluation results.

Source code in matgpr/safe_equations.py

@dataclass(frozen=True)
class SafeEquationEvaluationResult:
    """JSON-safe shell for future safe-equation evaluation results."""

    spec_name: str
    values: Sequence[float | None] = ()
    finite_count: int = 0
    nonfinite_count: int = 0
    warnings: Sequence[str] = ()
    metadata: Mapping[str, Any] = field(default_factory=dict)
    schema_version: str = SAFE_EQUATION_SCHEMA_VERSION

    def __post_init__(self) -> None:
        values = tuple(_optional_finite_float(value, "evaluation value") for value in self.values)
        object.__setattr__(self, "spec_name", _validate_symbol_name(self.spec_name, "spec_name"))
        object.__setattr__(self, "values", values)
        object.__setattr__(
            self, "finite_count", _nonnegative_int(self.finite_count, "finite_count")
        )
        object.__setattr__(
            self,
            "nonfinite_count",
            _nonnegative_int(self.nonfinite_count, "nonfinite_count"),
        )
        object.__setattr__(self, "warnings", _string_tuple(self.warnings, "warnings"))
        object.__setattr__(self, "metadata", _json_safe_mapping(self.metadata, "metadata"))
        object.__setattr__(self, "schema_version", str(self.schema_version))

    def to_dict(self) -> dict[str, object]:
        """Return a JSON-safe evaluation result."""
        return {
            "schema_version": self.schema_version,
            "spec_name": self.spec_name,
            "values": list(self.values),
            "finite_count": self.finite_count,
            "nonfinite_count": self.nonfinite_count,
            "warnings": list(self.warnings),
            "metadata": dict(self.metadata),
        }

to_dict()

Return a JSON-safe evaluation result.

Source code in matgpr/safe_equations.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-safe evaluation result."""
    return {
        "schema_version": self.schema_version,
        "spec_name": self.spec_name,
        "values": list(self.values),
        "finite_count": self.finite_count,
        "nonfinite_count": self.nonfinite_count,
        "warnings": list(self.warnings),
        "metadata": dict(self.metadata),
    }

SafeEquationMeanPreview dataclass

JSON-safe shell for future physics-mean preview payloads.

Source code in matgpr/safe_equations.py

@dataclass(frozen=True)
class SafeEquationMeanPreview:
    """JSON-safe shell for future physics-mean preview payloads."""

    spec_name: str
    preview_ready: bool
    valid_input_row_count: int = 0
    invalid_input_row_count: int = 0
    finite_mean_count: int = 0
    nonfinite_mean_count: int = 0
    physics_mean_summary: Mapping[str, Any] = field(default_factory=dict)
    target_residual_summary: Mapping[str, Any] | None = None
    warnings: Sequence[str] = ()
    metadata: Mapping[str, Any] = field(default_factory=dict)
    schema_version: str = SAFE_EQUATION_SCHEMA_VERSION

    def __post_init__(self) -> None:
        object.__setattr__(self, "spec_name", _validate_symbol_name(self.spec_name, "spec_name"))
        object.__setattr__(self, "preview_ready", bool(self.preview_ready))
        object.__setattr__(
            self,
            "valid_input_row_count",
            _nonnegative_int(self.valid_input_row_count, "valid_input_row_count"),
        )
        object.__setattr__(
            self,
            "invalid_input_row_count",
            _nonnegative_int(self.invalid_input_row_count, "invalid_input_row_count"),
        )
        object.__setattr__(
            self,
            "finite_mean_count",
            _nonnegative_int(self.finite_mean_count, "finite_mean_count"),
        )
        object.__setattr__(
            self,
            "nonfinite_mean_count",
            _nonnegative_int(self.nonfinite_mean_count, "nonfinite_mean_count"),
        )
        object.__setattr__(
            self,
            "physics_mean_summary",
            _json_safe_mapping(self.physics_mean_summary, "physics_mean_summary"),
        )
        residual_summary = (
            None
            if self.target_residual_summary is None
            else _json_safe_mapping(self.target_residual_summary, "target_residual_summary")
        )
        object.__setattr__(self, "target_residual_summary", residual_summary)
        object.__setattr__(self, "warnings", _string_tuple(self.warnings, "warnings"))
        object.__setattr__(self, "metadata", _json_safe_mapping(self.metadata, "metadata"))
        object.__setattr__(self, "schema_version", str(self.schema_version))

    def to_dict(self) -> dict[str, object]:
        """Return a JSON-safe physics-mean preview shell."""
        return {
            "schema_version": self.schema_version,
            "spec_name": self.spec_name,
            "preview_ready": self.preview_ready,
            "valid_input_row_count": self.valid_input_row_count,
            "invalid_input_row_count": self.invalid_input_row_count,
            "finite_mean_count": self.finite_mean_count,
            "nonfinite_mean_count": self.nonfinite_mean_count,
            "physics_mean_summary": dict(self.physics_mean_summary),
            "target_residual_summary": (
                None if self.target_residual_summary is None else dict(self.target_residual_summary)
            ),
            "warnings": list(self.warnings),
            "metadata": dict(self.metadata),
        }

to_dict()

Return a JSON-safe physics-mean preview shell.

Source code in matgpr/safe_equations.py

def to_dict(self) -> dict[str, object]:
    """Return a JSON-safe physics-mean preview shell."""
    return {
        "schema_version": self.schema_version,
        "spec_name": self.spec_name,
        "preview_ready": self.preview_ready,
        "valid_input_row_count": self.valid_input_row_count,
        "invalid_input_row_count": self.invalid_input_row_count,
        "finite_mean_count": self.finite_mean_count,
        "nonfinite_mean_count": self.nonfinite_mean_count,
        "physics_mean_summary": dict(self.physics_mean_summary),
        "target_residual_summary": (
            None if self.target_residual_summary is None else dict(self.target_residual_summary)
        ),
        "warnings": list(self.warnings),
        "metadata": dict(self.metadata),
    }

SafeEquationMeanFunction

Bases: PhysicsInformedMean

Physics-informed mean function backed by a safe custom equation spec.

Source code in matgpr/safe_equations.py

class SafeEquationMeanFunction(PhysicsInformedMean):
    """Physics-informed mean function backed by a safe custom equation spec."""

    def __init__(
        self,
        spec: SafeEquationSpec | Mapping[str, Any],
        *,
        feature_indices: Mapping[str, int] | None = None,
        column_indices: Mapping[str, int] | None = None,
        parameter_values: Mapping[str, Any] | None = None,
        positive_parameters: Sequence[str] | None = None,
        infer_positive_parameters: bool = True,
        feature_means: Mapping[str, float] | None = None,
        feature_stds: Mapping[str, float] | None = None,
        target_mean: float = 0.0,
        target_std: float = 1.0,
    ):
        resolved = _require_valid_spec(spec)
        resolved_feature_indices = _resolve_safe_equation_feature_indices(
            resolved,
            feature_indices=feature_indices,
            column_indices=column_indices,
        )
        learnable_parameters, fixed_parameters = _safe_equation_parameter_maps(
            resolved,
            parameter_values=parameter_values,
        )
        resolved_positive_parameters = _resolve_safe_equation_positive_parameters(
            resolved,
            learnable_parameters=learnable_parameters,
            positive_parameters=positive_parameters,
            infer_positive_parameters=infer_positive_parameters,
        )

        super().__init__(
            equation=_SafeEquationTorchEquation(resolved),
            feature_indices=resolved_feature_indices,
            learnable_parameters=learnable_parameters,
            positive_parameters=resolved_positive_parameters,
            fixed_parameters=fixed_parameters,
            feature_means=feature_means,
            feature_stds=feature_stds,
            target_mean=target_mean,
            target_std=target_std,
        )
        self.safe_equation_spec = resolved
        self.safe_equation_feature_names = tuple(resolved_feature_indices)
        self.safe_equation_parameter_bounds = {
            parameter.name: _bounds_to_list(parameter.bounds) for parameter in resolved.parameters
        }

    def to_dict(self) -> dict[str, object]:
        """Return JSON-safe metadata describing the mean-function adapter."""
        return {
            "schema_version": self.safe_equation_spec.schema_version,
            "spec": self.safe_equation_spec.to_dict(),
            "feature_indices": dict(self.feature_indices),
            "feature_names": list(self.safe_equation_feature_names),
            "learnable_parameters": list(self._learnable_parameter_names),
            "fixed_parameters": list(self.fixed_parameters),
            "positive_parameters": sorted(self.positive_parameters),
            "parameter_bounds": dict(self.safe_equation_parameter_bounds),
            "current_parameter_values": self.current_parameter_values(),
        }

to_dict()

Return JSON-safe metadata describing the mean-function adapter.

Source code in matgpr/safe_equations.py

def to_dict(self) -> dict[str, object]:
    """Return JSON-safe metadata describing the mean-function adapter."""
    return {
        "schema_version": self.safe_equation_spec.schema_version,
        "spec": self.safe_equation_spec.to_dict(),
        "feature_indices": dict(self.feature_indices),
        "feature_names": list(self.safe_equation_feature_names),
        "learnable_parameters": list(self._learnable_parameter_names),
        "fixed_parameters": list(self.fixed_parameters),
        "positive_parameters": sorted(self.positive_parameters),
        "parameter_bounds": dict(self.safe_equation_parameter_bounds),
        "current_parameter_values": self.current_parameter_values(),
    }

validate_safe_equation_spec(spec)

Validate the schema-level safety of a custom equation spec.

This validation layer checks names, duplicate symbols, finite numeric values, parameter bounds, JSON-safe metadata, and expression AST safety.

Source code in matgpr/safe_equations.py

def validate_safe_equation_spec(
    spec: SafeEquationSpec | Mapping[str, Any],
) -> SafeEquationValidationResult:
    """Validate the schema-level safety of a custom equation spec.

    This validation layer checks names, duplicate symbols, finite numeric
    values, parameter bounds, JSON-safe metadata, and expression AST safety.
    """
    spec_name = _spec_name_from_unknown(spec)
    try:
        resolved = spec if isinstance(spec, SafeEquationSpec) else SafeEquationSpec.from_dict(spec)
    except (TypeError, ValueError, KeyError) as exc:
        issue = SafeEquationValidationIssue(
            code="invalid_spec",
            message=str(exc),
            field=None,
            severity="error",
        )
        return SafeEquationValidationResult(issues=(issue,), spec_name=spec_name)

    issues, expression_symbols, expression_functions = _validate_expression(resolved)
    return SafeEquationValidationResult(
        issues=tuple(issues),
        spec_name=resolved.name,
        declared_symbols=resolved.declared_symbols,
        expression_symbols=tuple(sorted(expression_symbols)),
        expression_functions=tuple(sorted(expression_functions)),
        allowed_functions=ALLOWED_SAFE_EQUATION_FUNCTIONS,
        allowed_operators=ALLOWED_SAFE_EQUATION_OPERATORS,
    )

build_safe_equation_mean_function(spec, *, feature_indices=None, column_indices=None, parameter_values=None, positive_parameters=None, infer_positive_parameters=True, feature_means=None, feature_stds=None, target_mean=0.0, target_std=1.0)

Build a trainable PI-GPR mean function from a safe equation spec.

Source code in matgpr/safe_equations.py

def build_safe_equation_mean_function(
    spec: SafeEquationSpec | Mapping[str, Any],
    *,
    feature_indices: Mapping[str, int] | None = None,
    column_indices: Mapping[str, int] | None = None,
    parameter_values: Mapping[str, Any] | None = None,
    positive_parameters: Sequence[str] | None = None,
    infer_positive_parameters: bool = True,
    feature_means: Mapping[str, float] | None = None,
    feature_stds: Mapping[str, float] | None = None,
    target_mean: float = 0.0,
    target_std: float = 1.0,
) -> SafeEquationMeanFunction:
    """Build a trainable PI-GPR mean function from a safe equation spec."""
    return SafeEquationMeanFunction(
        spec,
        feature_indices=feature_indices,
        column_indices=column_indices,
        parameter_values=parameter_values,
        positive_parameters=positive_parameters,
        infer_positive_parameters=infer_positive_parameters,
        feature_means=feature_means,
        feature_stds=feature_stds,
        target_mean=target_mean,
        target_std=target_std,
    )

evaluate_safe_equation_numpy(spec, variable_values, *, parameter_values=None, constant_values=None)

Evaluate a validated safe equation with NumPy.

The expression is never passed to eval. The function validates the equation spec, parses the already allowlisted AST, and interprets only the small supported math grammar. Non-finite outputs are returned as None so to_dict() remains strict-JSON-safe.

Source code in matgpr/safe_equations.py

def evaluate_safe_equation_numpy(
    spec: SafeEquationSpec | Mapping[str, Any],
    variable_values: Mapping[str, Any] | Any,
    *,
    parameter_values: Mapping[str, Any] | None = None,
    constant_values: Mapping[str, Any] | None = None,
) -> SafeEquationEvaluationResult:
    """Evaluate a validated safe equation with NumPy.

    The expression is never passed to ``eval``. The function validates the
    equation spec, parses the already allowlisted AST, and interprets only the
    small supported math grammar. Non-finite outputs are returned as ``None`` so
    ``to_dict()`` remains strict-JSON-safe.
    """
    resolved = _resolve_spec_or_failure(spec)
    if isinstance(resolved, SafeEquationEvaluationResult):
        return resolved
    validation = validate_safe_equation_spec(resolved)
    if not validation.is_valid:
        return _evaluation_failure(
            spec_name=resolved.name,
            warnings=_validation_warnings(validation),
            metadata={
                "evaluation_ready": False,
                "validation": validation.to_dict(),
            },
        )

    context_result = _build_numpy_context(
        resolved,
        variable_values,
        parameter_values=parameter_values,
        constant_values=constant_values,
    )
    if context_result["warnings"]:
        return _evaluation_failure(
            spec_name=resolved.name,
            warnings=context_result["warnings"],
            metadata={
                "evaluation_ready": False,
                "validation": validation.to_dict(),
                "input_row_count": context_result["input_row_count"],
            },
        )

    try:
        tree = ast.parse(resolved.expression, mode="eval")
        with np.errstate(all="ignore"):
            raw_values = _evaluate_numpy_node(tree, context_result["context"])
    except (FloatingPointError, OverflowError, ValueError, TypeError) as exc:
        return _evaluation_failure(
            spec_name=resolved.name,
            warnings=(f"evaluation failed: {exc}",),
            metadata={
                "evaluation_ready": False,
                "validation": validation.to_dict(),
                "input_row_count": context_result["input_row_count"],
            },
        )

    values_array = np.asarray(raw_values, dtype=float)
    if values_array.ndim == 0:
        row_count = int(context_result["input_row_count"])
        values_array = np.full(row_count, float(values_array), dtype=float)
    values_array = np.ravel(values_array)
    finite_mask = np.isfinite(values_array)
    finite_count = int(np.count_nonzero(finite_mask))
    nonfinite_count = int(values_array.size - finite_count)
    warnings: list[str] = []
    if nonfinite_count:
        warnings.append(f"equation produced {nonfinite_count} non-finite output value(s)")

    return SafeEquationEvaluationResult(
        spec_name=resolved.name,
        values=_json_safe_array_values(values_array),
        finite_count=finite_count,
        nonfinite_count=nonfinite_count,
        warnings=warnings,
        metadata={
            "evaluation_ready": nonfinite_count == 0,
            "input_row_count": int(values_array.size),
            "expression_symbols": list(validation.expression_symbols),
            "expression_functions": list(validation.expression_functions),
            "parameter_values": context_result["parameter_values"],
            "constant_values": context_result["constant_values"],
        },
    )

evaluate_safe_equation_torch(spec, variable_values, *, parameter_values=None, constant_values=None, dtype=None, device=None)

Evaluate a validated safe equation with torch operations.

This evaluator uses the same allowlisted AST as the NumPy preview path, but returns a torch tensor directly so gradients can flow through tensor variables or parameters. It is intended for future PI-GPR mean-function conversion and lower-level research workflows, not JSON API payloads.

Source code in matgpr/safe_equations.py

def evaluate_safe_equation_torch(
    spec: SafeEquationSpec | Mapping[str, Any],
    variable_values: Mapping[str, Any] | Any,
    *,
    parameter_values: Mapping[str, Any] | None = None,
    constant_values: Mapping[str, Any] | None = None,
    dtype: torch.dtype | None = None,
    device: str | torch.device | None = None,
) -> torch.Tensor:
    """Evaluate a validated safe equation with torch operations.

    This evaluator uses the same allowlisted AST as the NumPy preview path, but
    returns a torch tensor directly so gradients can flow through tensor
    variables or parameters. It is intended for future PI-GPR mean-function
    conversion and lower-level research workflows, not JSON API payloads.
    """
    resolved = _require_valid_spec(spec)
    dtype, device = _infer_torch_options(
        variable_values,
        parameter_values=parameter_values,
        constant_values=constant_values,
        dtype=dtype,
        device=device,
    )
    context, row_count = _build_torch_context(
        resolved,
        variable_values,
        parameter_values=parameter_values,
        constant_values=constant_values,
        dtype=dtype,
        device=device,
    )
    tree = ast.parse(resolved.expression, mode="eval")
    result = _evaluate_torch_node(tree, context)
    if not isinstance(result, torch.Tensor):
        result = torch.as_tensor(result, dtype=dtype, device=device)
    else:
        result = result.to(dtype=dtype, device=device)
    if result.ndim == 0:
        result = result.reshape(()).expand(row_count)
    return result.reshape(-1)

initialize_safe_equation_torch_parameters(spec, *, dtype=torch.float64, device=None, requires_grad=True, include_fixed=True)

Initialize declared equation parameters as torch tensors.

Learnable parameters receive requires_grad=True by default. Parameters marked learnable=False are included as fixed tensors when include_fixed=True and never require gradients. The returned tensors can be passed directly to evaluate_safe_equation_torch(...) or wrapped by a downstream torch.nn.Module/ParameterDict.

Source code in matgpr/safe_equations.py

def initialize_safe_equation_torch_parameters(
    spec: SafeEquationSpec | Mapping[str, Any],
    *,
    dtype: torch.dtype = torch.float64,
    device: str | torch.device | None = None,
    requires_grad: bool = True,
    include_fixed: bool = True,
) -> dict[str, torch.Tensor]:
    """Initialize declared equation parameters as torch tensors.

    Learnable parameters receive ``requires_grad=True`` by default. Parameters
    marked ``learnable=False`` are included as fixed tensors when
    ``include_fixed=True`` and never require gradients. The returned tensors can
    be passed directly to ``evaluate_safe_equation_torch(...)`` or wrapped by a
    downstream ``torch.nn.Module``/``ParameterDict``.
    """
    resolved = _require_valid_spec(spec)
    tensors: dict[str, torch.Tensor] = {}
    for parameter in resolved.parameters:
        if not parameter.learnable and not include_fixed:
            continue
        tensor = torch.tensor(
            parameter.initial_value,
            dtype=dtype,
            device=device,
            requires_grad=bool(requires_grad and parameter.learnable),
        )
        tensors[parameter.name] = tensor
    return tensors

preview_safe_equation_mean(spec, variable_values, *, target_values=None, parameter_values=None, constant_values=None)

Preview physics-mean outputs and optional target residuals.

This helper is designed for UI/API validation before model fitting. It reports input-row validity, finite physics-mean counts, compact summary statistics, and optional target - physics_mean residual summaries.

Source code in matgpr/safe_equations.py

def preview_safe_equation_mean(
    spec: SafeEquationSpec | Mapping[str, Any],
    variable_values: Mapping[str, Any] | Any,
    *,
    target_values: Any | None = None,
    parameter_values: Mapping[str, Any] | None = None,
    constant_values: Mapping[str, Any] | None = None,
) -> SafeEquationMeanPreview:
    """Preview physics-mean outputs and optional target residuals.

    This helper is designed for UI/API validation before model fitting. It
    reports input-row validity, finite physics-mean counts, compact summary
    statistics, and optional ``target - physics_mean`` residual summaries.
    """
    resolved = _resolve_spec_or_failure(spec)
    if isinstance(resolved, SafeEquationEvaluationResult):
        return SafeEquationMeanPreview(
            spec_name=resolved.spec_name,
            preview_ready=False,
            warnings=resolved.warnings,
            metadata=resolved.metadata,
        )

    input_quality = _input_row_quality(resolved, variable_values)
    evaluation = evaluate_safe_equation_numpy(
        resolved,
        variable_values,
        parameter_values=parameter_values,
        constant_values=constant_values,
    )
    mean_values = _values_to_numpy(evaluation.values)
    finite_mean_mask = np.isfinite(mean_values)
    finite_mean_count = int(np.count_nonzero(finite_mean_mask))
    nonfinite_mean_count = int(mean_values.size - finite_mean_count)
    warnings = [*input_quality["warnings"], *evaluation.warnings]

    target_residual_summary = None
    if target_values is not None:
        target_summary = _target_residual_summary(
            target_values=target_values,
            mean_values=mean_values,
        )
        warnings.extend(target_summary["warnings"])
        target_residual_summary = target_summary["summary"]

    preview_ready = (
        bool(evaluation.metadata.get("evaluation_ready", False))
        and input_quality["invalid_input_row_count"] == 0
        and nonfinite_mean_count == 0
    )
    if nonfinite_mean_count and not evaluation.warnings:
        warnings.append(f"physics mean contains {nonfinite_mean_count} non-finite value(s)")

    return SafeEquationMeanPreview(
        spec_name=resolved.name,
        preview_ready=preview_ready,
        valid_input_row_count=input_quality["valid_input_row_count"],
        invalid_input_row_count=input_quality["invalid_input_row_count"],
        finite_mean_count=finite_mean_count,
        nonfinite_mean_count=nonfinite_mean_count,
        physics_mean_summary=_summary_stats(mean_values),
        target_residual_summary=target_residual_summary,
        warnings=tuple(dict.fromkeys(warnings)),
        metadata={
            "evaluation": evaluation.to_dict(),
            "input_row_count": int(mean_values.size),
        },
    )

safe_equation_schema_snapshot()

Return a JSON-safe snapshot of the current safe-equation schema surface.

Source code in matgpr/safe_equations.py

def safe_equation_schema_snapshot() -> dict[str, object]:
    """Return a JSON-safe snapshot of the current safe-equation schema surface."""
    return {
        "schema_version": SAFE_EQUATION_SCHEMA_VERSION,
        "allowed_functions": list(ALLOWED_SAFE_EQUATION_FUNCTIONS),
        "allowed_operators": list(ALLOWED_SAFE_EQUATION_OPERATORS),
        "identifier_rule": _IDENTIFIER_PATTERN.pattern,
        "records": {
            "SafeEquationSpec": {
                "required_fields": ["name", "expression", "variables"],
                "optional_fields": [
                    "parameters",
                    "constants",
                    "display_name",
                    "description",
                    "target_units",
                    "assumptions",
                    "validity_limits",
                    "references",
                    "metadata",
                ],
            },
            "SafeEquationVariable": {
                "required_fields": ["name"],
                "optional_fields": ["description", "units", "required", "bounds"],
            },
            "SafeEquationParameter": {
                "required_fields": ["name", "initial_value"],
                "optional_fields": ["bounds", "description", "units", "learnable"],
            },
            "SafeEquationConstant": {
                "required_fields": ["name", "value"],
                "optional_fields": ["description", "units"],
            },
        },
        "safety_rules": [
            "No arbitrary Python execution.",
            "Only declared variables, parameters, constants, and allowlisted functions are valid.",
            "Expression parsing is performed with ast.parse(..., mode='eval') and a manual allowlist.",
            "All numeric defaults and bounds must be finite.",
            "All payload metadata must be strict-JSON-safe.",
            "NumPy evaluation uses a manual AST interpreter, never eval().",
            "Non-finite evaluation outputs are represented as null values in JSON payloads.",
            "Torch evaluation uses a manual AST interpreter, never eval().",
            "Torch evaluation preserves gradients through tensor variables and parameters.",
            "PI-GPR mean conversion is handled by build_safe_equation_mean_function(...).",
        ],
    }