PDEs

Every PDE is registered once in pdes.registry, and both the CLI and the dashboard read it from there.

create_pde

create_pde(name, config)

Instantiate the PDE registered under name with config.

Source code in pinnrl/pdes/registry.py

def create_pde(name: str, config: PDEConfig) -> PDEBase:
    """Instantiate the PDE registered under ``name`` with ``config``."""
    return get_pde_spec(name).cls(config=config)

get_pde_spec

get_pde_spec(name)

Resolve a config key or display name (case-insensitive) to its spec.

Raises:

Type	Description
KeyError	If name matches no registered PDE.

Source code in pinnrl/pdes/registry.py

def get_pde_spec(name: str) -> PDESpec:
    """Resolve a config key or display name (case-insensitive) to its spec.

    Raises:
        KeyError: If ``name`` matches no registered PDE.
    """
    spec = _LOOKUP.get(name.strip().lower())
    if spec is None:
        available = ", ".join(s.key for s in PDE_SPECS)
        raise KeyError(f"Unknown PDE {name!r}. Available: {available}")
    return spec

PDESpec dataclass

PDESpec(key, display_name, cls)

A registered PDE.

Attributes:

Name	Type	Description
key	str	Config key under pde_configs in config.yaml.
display_name	str	Human-readable name shown by the dashboard and CLI.
cls	type[PDEBase]	The :class:PDEBase subclass implementing it.

PDEConfig dataclass

PDEConfig(
    name,
    domain,
    time_domain,
    parameters,
    boundary_conditions,
    initial_condition,
    exact_solution,
    dimension=1,
    input_dim=None,
    output_dim=None,
    architecture=None,
    device=None,
    training=None,
    trainable_parameters=list(),
    parameter_initial_guesses=dict(),
    observation_data=None,
)

Configuration dataclass for PDE parameters.

PDEBase

PDEBase(config, bandit_sampler=None)

Base class for PDEs solved by PINNs.

A subclass provides the physics:

• :meth:compute_residual — the PDE operator applied to the network.
• :meth:_create_initial_condition — u(x, t0) for its IC types.
• :meth:_exact_solution — a closed-form solution, when one exists.
• time_order = 2 for second-order-in-time PDEs; the initial loss then also enforces u_t(x, t0) = initial_condition["velocity"] (default 0).

The base class owns everything else: derivatives, collocation sampling, boundary conditions and the composite loss. boundary_conditions in the config maps a kind to its options; every spatial axis gets the condition on both faces:

• dirichlet: {"value": v} or {"left": a, "right": b} (the lower / upper face), or {"type": "exact"} to take values from the exact solution;
• neumann: {"value": g} — du/dx_axis = g on both faces;
• periodic: {} — u and du/dx_axis agree on opposite faces.

exact_solution: {} in the config means the problem has no closed-form solution; :meth:exact_solution then returns None.

Initialize PDE with given configuration.

Parameters:

Name	Type	Description	Default
config	PDEConfig	Configuration dataclass.	required
bandit_sampler		Optional :class:ResidualBanditSampler for the adaptive strategy; built lazily with defaults when omitted.	None

Source code in pinnrl/pdes/pde_base.py

def __init__(self, config: PDEConfig, bandit_sampler=None):
    """Initialize PDE with given configuration.

    Args:
        config: Configuration dataclass.
        bandit_sampler: Optional :class:`ResidualBanditSampler` for the
            ``adaptive`` strategy; built lazily with defaults when omitted.
    """
    self.config = config
    self.domain = config.domain
    self.bandit_sampler = bandit_sampler

    # One (lo, hi) pair per spatial axis; a bare (lo, hi) means one axis.
    if not self.domain:
        self.domain = [(0.0, 1.0)]
    elif isinstance(self.domain[0], (list, tuple)):
        self.domain = [(float(lo), float(hi)) for lo, hi in self.domain]
    else:
        self.domain = [(float(self.domain[0]), float(self.domain[1]))]

    # Update config with normalized domain format
    self.config.domain = self.domain

    # Handle time_domain/t_domain property
    if hasattr(config, "time_domain"):
        self.time_domain = config.time_domain
    elif hasattr(config, "t_domain"):
        self.time_domain = config.t_domain
    else:
        self.time_domain = [0.0, 1.0]

    # Convert time_domain to tuple if it's a list
    if isinstance(self.time_domain, list):
        self.time_domain = tuple(self.time_domain)

    # Setup device - Handle device explicitly and carefully
    if hasattr(config, "device") and config.device is not None:
        # If config.device is already a torch.device object, use it directly
        if isinstance(config.device, torch.device):
            self.device = config.device
        else:
            # Try to convert string to torch.device
            try:
                self.device = torch.device(str(config.device))
            except Exception:
                print(f"Warning: Invalid device '{config.device}', falling back to CPU")
                self.device = torch.device("cpu")
    else:
        # Fallback to CPU if no device specified
        self.device = torch.device("cpu")

    # Ensure config has the same device
    self.config.device = self.device

    # Store dimensionality
    self.dimension = config.dimension

    # Make sure parameters exist but preserve any existing parameters
    if not hasattr(config, "parameters"):
        config.parameters = {}
    elif config.parameters is None:
        config.parameters = {}

    # Register trainable parameters as nn.Parameter (inverse-problem mode).
    # In forward mode this dict stays empty and adds no behaviour.
    self._trainable_params: nn.ParameterDict = nn.ParameterDict()
    trainable_names = list(getattr(config, "trainable_parameters", []) or [])
    initial_guesses = dict(getattr(config, "parameter_initial_guesses", {}) or {})
    # Snapshot the original (true) values BEFORE the parameters become
    # trainable, so we can plot a reference line in the dashboard.
    self._true_parameters: dict[str, float] = {}
    for name in trainable_names:
        true_val = _as_scalar(name, config.parameters.get(name))
        if true_val is not None:
            self._true_parameters[name] = true_val
        init_val = _as_scalar(
            name, initial_guesses.get(name, true_val if true_val is not None else 1.0)
        )
        self._trainable_params[name] = nn.Parameter(torch.tensor(init_val, device=self.device))

    # Load observation data (used by the data-fitting loss in inverse mode).
    self.observation_data: dict[str, torch.Tensor] | None = self._load_observation_data(
        getattr(config, "observation_data", None)
    )
    # No time window of its own (a Well dataset, whose trajectory length can
    # differ by file): span the time range of the observations, else [0, 1].
    if self.time_domain is None:
        if self.observation_data is not None:
            t = self.observation_data["t"]
            self.time_domain = (float(t.min()), float(t.max()))
        else:
            self.time_domain = (0.0, 1.0)
        self.config.time_domain = self.time_domain

    self.initial_condition = self._create_initial_condition(
        dict(config.initial_condition or {})
    )
    self.boundary_conditions = self._parse_boundary_conditions(config.boundary_conditions or {})
    self._fixed_points: dict[tuple[str, int], Any] = {}

    # Setup neural network parameters if not specified
    if self.config.input_dim is None:
        self.config.input_dim = self.dimension + 1  # Spatial dimensions + time

    if self.config.output_dim is None:
        self.config.output_dim = 1  # Default to single output (u)

has_exact_solution property

has_exact_solution

Whether the config names a closed-form solution.

get_parameter

get_parameter(name, default=None, required=False)

Safely get a parameter value with validation.

When name has been registered as a trainable parameter (inverse problem mode), the live nn.Parameter tensor is returned so that downstream residual computations build the autograd graph through it. Otherwise the plain float from config.parameters is returned.

Parameters:

Name	Type	Description	Default
name	str	Parameter name	required
default		Default value if parameter not found	None
required	bool	Whether parameter is required (raises error if missing)	False

Returns:

Type	Description
	Parameter value (nn.Parameter if trainable, else float)

Raises:

Type	Description
ValueError	If required parameter is missing

Source code in pinnrl/pdes/pde_base.py

def get_parameter(self, name: str, default=None, required: bool = False):
    """Safely get a parameter value with validation.

    When ``name`` has been registered as a trainable parameter (inverse
    problem mode), the live ``nn.Parameter`` tensor is returned so that
    downstream residual computations build the autograd graph through it.
    Otherwise the plain float from ``config.parameters`` is returned.

    Args:
        name: Parameter name
        default: Default value if parameter not found
        required: Whether parameter is required (raises error if missing)

    Returns:
        Parameter value (``nn.Parameter`` if trainable, else float)

    Raises:
        ValueError: If required parameter is missing
    """
    # Trainable param takes precedence over the static config value.
    trainable = getattr(self, "_trainable_params", None)
    if trainable is not None and name in trainable:
        return trainable[name]

    if not hasattr(self.config, "parameters") or self.config.parameters is None:
        if required:
            raise ValueError(f"Required parameter '{name}' not found in config")
        return default

    value = self.config.parameters.get(name, default)
    if value is None and required:
        raise ValueError(f"Required parameter '{name}' not found in config")
    return value

trainable_parameters_iter

trainable_parameters_iter()

Iterator over registered trainable PDE parameters (for the optimizer).

Source code in pinnrl/pdes/pde_base.py

def trainable_parameters_iter(self):
    """Iterator over registered trainable PDE parameters (for the optimizer)."""
    if hasattr(self, "_trainable_params"):
        return self._trainable_params.parameters()
    return iter(())

get_trainable_parameter_values

get_trainable_parameter_values()

Snapshot of current trainable parameter values as plain floats.

Source code in pinnrl/pdes/pde_base.py

def get_trainable_parameter_values(self) -> dict[str, float]:
    """Snapshot of current trainable parameter values as plain floats."""
    if not hasattr(self, "_trainable_params"):
        return {}
    return {name: float(p.detach().cpu().item()) for name, p in self._trainable_params.items()}

generate_synthetic_observations

generate_synthetic_observations(
    n_points=200, noise_std=0.0, seed=0
)

Sample noisy observations from the analytical solution.

Useful for inverse-problem demos and CI tests: pick uniform points in the domain, evaluate self.exact_solution, optionally add Gaussian noise, and store the result on self.observation_data so the next compute_loss call automatically picks it up.

Parameters:

Name	Type	Description	Default
n_points	int	Number of observations.	200
noise_std	float	Standard deviation of the additive Gaussian noise.	0.0
seed	int | None	Seed for the points and the noise; None leaves it unseeded.	0

Returns:

Type	Description
dict[str, Tensor]	{"x", "t", "u"} tensors, also stored on self.observation_data.

Raises:

Type	Description
ValueError	If the PDE has no exact solution configured.

Source code in pinnrl/pdes/pde_base.py

def generate_synthetic_observations(
    self,
    n_points: int = 200,
    noise_std: float = 0.0,
    seed: int | None = 0,
) -> dict[str, torch.Tensor]:
    """Sample noisy observations from the analytical solution.

    Useful for inverse-problem demos and CI tests: pick uniform points in
    the domain, evaluate ``self.exact_solution``, optionally add Gaussian
    noise, and store the result on ``self.observation_data`` so the next
    ``compute_loss`` call automatically picks it up.

    Args:
        n_points: Number of observations.
        noise_std: Standard deviation of the additive Gaussian noise.
        seed: Seed for the points and the noise; ``None`` leaves it unseeded.

    Returns:
        ``{"x", "t", "u"}`` tensors, also stored on ``self.observation_data``.

    Raises:
        ValueError: If the PDE has no exact solution configured.
    """
    gen = torch.Generator(device="cpu")
    if seed is not None:
        gen.manual_seed(int(seed))

    spatial_dims = max(int(self.dimension), 1)
    x_cols = []
    for d in range(spatial_dims):
        lo, hi = self.domain[d]
        x_cols.append(torch.rand(n_points, 1, generator=gen) * (hi - lo) + lo)
    x = (
        torch.cat(x_cols, dim=1).to(self.device)
        if spatial_dims > 1
        else x_cols[0].to(self.device)
    )

    t_lo, t_hi = self.time_domain[0], self.time_domain[1]
    t = (torch.rand(n_points, 1, generator=gen) * (t_hi - t_lo) + t_lo).to(self.device)

    # Temporarily unregister the trainable params so ``get_parameter``
    # returns the TRUE values from ``config.parameters`` (the ones
    # snapshotted in ``_true_parameters``) rather than the current guess.
    # Without this, synthetic obs generated in inverse mode would already
    # encode the wrong α and the trainer would converge to the guess
    # rather than recovering the true value.
    saved_params = self._trainable_params if hasattr(self, "_trainable_params") else None
    try:
        if saved_params is not None and self._true_parameters:
            self._trainable_params = nn.ParameterDict()
        if not self.has_exact_solution:
            raise ValueError(
                f"{self.config.name}: synthetic observations need a configured exact_solution"
            )
        with torch.no_grad():
            u = self.exact_solution(x, t)
            if noise_std and noise_std > 0:
                noise = torch.randn(u.shape, generator=gen).to(self.device) * float(noise_std)
                u = u + noise
    finally:
        if saved_params is not None:
            self._trainable_params = saved_params

    self.observation_data = {"x": x, "t": t, "u": u}
    return self.observation_data

compute_residual

compute_residual(model, x, t)

Compute the PDE residual. Must be implemented by subclasses.

Parameters:

Name	Type	Description	Default
model	Module	Neural network model.	required
x	Tensor	Spatial coordinates, shape (N, dimension).	required
t	Tensor	Time coordinates, shape (N, 1).	required

Returns:

Type	Description
Tensor	Residual tensor, shape (N, 1).

Source code in pinnrl/pdes/pde_base.py

def compute_residual(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """Compute the PDE residual. Must be implemented by subclasses.

    Args:
        model: Neural network model.
        x: Spatial coordinates, shape ``(N, dimension)``.
        t: Time coordinates, shape ``(N, 1)``.

    Returns:
        Residual tensor, shape ``(N, 1)``.
    """
    raise NotImplementedError("Subclasses must implement compute_residual")

laplacian

laplacian(f, x)

Sum of the pure second derivatives of f along each column of x.

x must be the leaf tensor f was computed from (requires_grad). Differentiating with respect to the whole tensor and then taking a column is what makes this correct in more than one dimension; a column slice x[:, d:d+1] is not part of the graph.

Source code in pinnrl/pdes/pde_base.py

def laplacian(self, f: torch.Tensor, x: torch.Tensor) -> torch.Tensor:
    """Sum of the pure second derivatives of ``f`` along each column of ``x``.

    ``x`` must be the leaf tensor ``f`` was computed from (``requires_grad``).
    Differentiating with respect to the whole tensor and then taking a column
    is what makes this correct in more than one dimension; a column slice
    ``x[:, d:d+1]`` is not part of the graph.
    """
    grad = self._grad(f, x)
    lap = torch.zeros_like(f)
    for d in range(x.shape[1]):
        lap = lap + self._grad(grad[:, d : d + 1], x)[:, d : d + 1]
    return lap

compute_derivatives

compute_derivatives(
    model,
    x,
    t,
    temporal_derivatives=None,
    spatial_derivatives=None,
)

Evaluate u = model([x, t]) and the requested derivatives.

Parameters:

Name	Type	Description	Default
model	Module	Neural network model.	required
x	Tensor	Spatial coordinates, shape (N, dimension).	required
t	Tensor	Time coordinates, shape (N, 1).	required
temporal_derivatives	list[int]	Orders of ∂/∂t to return (1 and/or 2).	None
spatial_derivatives	set[int]	Orders of the pure ∂/∂x_d derivatives to return (1 to 4).	None

Returns a dict with "u" and, for the requested orders only:

• time: "dt", "dt2";
• space, 1-D: "dx", "dx2", "dx3", "dx4";
• space, n-D: pure derivatives per axis, "dx1", "dx1x1", ... "dx2x2";
• "laplacian" whenever order 2 is requested.

Raises:

Type	Description
ValueError	For temporal orders above 2 or spatial orders above 4.

Source code in pinnrl/pdes/pde_base.py

def compute_derivatives(
    self,
    model: torch.nn.Module,
    x: torch.Tensor,
    t: torch.Tensor,
    temporal_derivatives: list[int] = None,
    spatial_derivatives: set[int] = None,
) -> dict[str, torch.Tensor]:
    """Evaluate ``u = model([x, t])`` and the requested derivatives.

    Args:
        model: Neural network model.
        x: Spatial coordinates, shape ``(N, dimension)``.
        t: Time coordinates, shape ``(N, 1)``.
        temporal_derivatives: Orders of ``∂/∂t`` to return (1 and/or 2).
        spatial_derivatives: Orders of the pure ``∂/∂x_d`` derivatives to
            return (1 to 4).

    Returns a dict with ``"u"`` and, for the requested orders only:

    * time: ``"dt"``, ``"dt2"``;
    * space, 1-D: ``"dx"``, ``"dx2"``, ``"dx3"``, ``"dx4"``;
    * space, n-D: pure derivatives per axis, ``"dx1"``, ``"dx1x1"``, ... ``"dx2x2"``;
    * ``"laplacian"`` whenever order 2 is requested.

    Raises:
        ValueError: For temporal orders above 2 or spatial orders above 4.
    """
    temporal = sorted(set(temporal_derivatives or ()) - {0})
    spatial = sorted(set(spatial_derivatives or ()) - {0})
    if temporal and temporal[-1] > 2:
        raise ValueError(
            f"Temporal derivative order {temporal[-1]} is not supported. Maximum order is 2."
        )
    if spatial and spatial[-1] > 4:
        raise ValueError(
            f"Spatial derivative order {spatial[-1]} is not supported. Maximum order is 4."
        )

    x, t, u = self._prepare_inputs(model, x, t)
    derivatives = {"u": u}

    current = u
    for order in range(1, (temporal[-1] if temporal else 0) + 1):
        current = self._grad(current, t)
        if order in temporal:
            derivatives["dt" if order == 1 else f"dt{order}"] = current

    for dim in range(self.dimension):
        current = u
        for order in range(1, (spatial[-1] if spatial else 0) + 1):
            current = self._grad(current, x)[:, dim : dim + 1]
            if order not in spatial:
                continue
            if self.dimension == 1:
                key = "dx" if order == 1 else f"dx{order}"
            else:
                key = "d" + f"x{dim + 1}" * order
            derivatives[key] = current

    if 2 in spatial:
        if self.dimension == 1:
            derivatives["laplacian"] = derivatives["dx2"]
        else:
            derivatives["laplacian"] = sum(
                derivatives["d" + f"x{d + 1}" * 2] for d in range(self.dimension)
            )
    return derivatives

exact_solution

exact_solution(x, t)

Closed-form u(x, t), or None when exact_solution is not configured.

Source code in pinnrl/pdes/pde_base.py

def exact_solution(self, x: torch.Tensor, t: torch.Tensor) -> torch.Tensor | None:
    """Closed-form ``u(x, t)``, or ``None`` when ``exact_solution`` is not configured."""
    if not self.has_exact_solution:
        return None
    # Exact solutions start from the IC at t = 0; shift so they hold for any t0.
    return self._exact_solution(x, t - float(self.time_domain[0]))

generate_collocation_points

generate_collocation_points(
    num_points, strategy="uniform", **kwargs
)

Generate collocation points for training.

Parameters:

Name	Type	Description	Default
num_points	int	Number of points to generate.	required
strategy	str	Sampling strategy, one of: uniform: jittered space-time grid (:meth:_sample_uniform); stratified: Latin-hypercube-style stratified random; residual_based: RAD (:meth:_sample_residual_based), driven by the model kwarg and the optional k, c, pool_factor and replacement kwargs; adaptive: the experimental residual-bandit sampler (1-D only), driven by the model kwarg and an optional progress in [0, 1]. residual_based and adaptive fall back to uniform when no model is passed.	'uniform'
**kwargs		Strategy options listed above.	{}

Returns:

Type	Description
tuple[Tensor, Tensor]	(x, t) with shapes (num_points, dimension) and (num_points, 1).

Raises:

Type	Description
ValueError	For an unknown strategy, or adaptive with a model on a PDE with more than one spatial dimension.

Source code in pinnrl/pdes/pde_base.py

def generate_collocation_points(
    self, num_points: int, strategy: str = "uniform", **kwargs
) -> tuple[torch.Tensor, torch.Tensor]:
    """Generate collocation points for training.

    Args:
        num_points: Number of points to generate.
        strategy: Sampling strategy, one of:

            * ``uniform``: jittered space-time grid (:meth:`_sample_uniform`);
            * ``stratified``: Latin-hypercube-style stratified random;
            * ``residual_based``: RAD (:meth:`_sample_residual_based`),
              driven by the ``model`` kwarg and the optional ``k``, ``c``,
              ``pool_factor`` and ``replacement`` kwargs;
            * ``adaptive``: the experimental residual-bandit sampler (1-D
              only), driven by the ``model`` kwarg and an optional
              ``progress`` in [0, 1].

            ``residual_based`` and ``adaptive`` fall back to ``uniform``
            when no ``model`` is passed.
        **kwargs: Strategy options listed above.

    Returns:
        ``(x, t)`` with shapes ``(num_points, dimension)`` and ``(num_points, 1)``.

    Raises:
        ValueError: For an unknown strategy, or ``adaptive`` with a model on
            a PDE with more than one spatial dimension.
    """
    if strategy == "uniform":
        x, t = self._sample_uniform(num_points)

    elif strategy == "stratified":
        x, t = self._sample_stratified(num_points)

    elif strategy == "residual_based":
        model = kwargs.get("model", None)
        rad_kwargs = {
            key: kwargs[key]
            for key in ("k", "c", "pool_factor", "replacement")
            if key in kwargs
        }
        x, t = self._sample_residual_based(num_points, model, **rad_kwargs)

    elif strategy == "adaptive":
        model = kwargs.get("model", None)
        if model is None:
            x, t = self._sample_uniform(num_points)
        else:
            if self.dimension != 1:
                raise ValueError(
                    "The 'adaptive' (RL bandit) sampler supports 1-D PDEs only; "
                    "use 'residual_based' for higher dimensions."
                )
            if self.bandit_sampler is None:
                from pinnrl.config import RLConfig

                self.bandit_sampler = RLConfig().build_sampler(self)
            x, t = self.bandit_sampler.sample(
                self, model, num_points, progress=kwargs.get("progress", 0.0)
            )
    else:
        raise ValueError(f"Unknown sampling strategy: {strategy}")

    # Ensure the points are on the correct device before returning
    x = x.to(self.device)
    t = t.to(self.device)

    return x, t

compute_loss

compute_loss(model, x, t)

Loss components and their weighted total.

Returns residual (at the collocation points x, t), boundary, initial, data (observations; zero without any) and total. training.mode selects the objective (see :meth:weighted_total): data_only keeps only the data term; every other mode adds the weighted physics terms to the data term. Weights come from training.loss_weights (residual/pde, boundary, initial, data; 1, 10, 10 and 1 when a key is absent). With adaptive weighting enabled the physics weights here are 1 and the trainer re-weights the total.

Source code in pinnrl/pdes/pde_base.py

def compute_loss(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> dict[str, torch.Tensor]:
    """Loss components and their weighted total.

    Returns ``residual`` (at the collocation points ``x, t``), ``boundary``,
    ``initial``, ``data`` (observations; zero without any) and ``total``.
    ``training.mode`` selects the objective (see :meth:`weighted_total`):
    ``data_only`` keeps only the data term; every other mode adds the
    weighted physics terms to the data term. Weights come from
    ``training.loss_weights`` (``residual``/``pde``, ``boundary``,
    ``initial``, ``data``; 1, 10, 10 and 1 when a key is absent). With
    adaptive weighting enabled the physics weights here are 1 and the
    trainer re-weights the total.
    """
    losses = {
        "residual": self._apply_loss_fn(self.compute_residual(model, x, t)),
        "boundary": self._boundary_loss(model),
        "initial": self._initial_loss(model),
        "data": self._compute_data_loss(model),
    }

    weights = self._training_option("loss_weights") or {}
    adaptive = self._training_option("adaptive_weights")
    if adaptive is not None and getattr(adaptive, "enabled", False):
        physics_weights = (1.0, 1.0, 1.0)  # the trainer re-weights with its own
    else:
        physics_weights = (
            float(weights.get("pde", weights.get("residual", 1.0))),
            float(weights.get("boundary", 10.0)),
            float(weights.get("initial", 10.0)),
        )
    losses["total"] = self.weighted_total(losses, *physics_weights)
    return losses

weighted_total

weighted_total(losses, w_residual, w_boundary, w_initial)

Combine loss components under training.mode.

data_only drops the physics terms; outside forward the data term always counts (its weight is at least 1).

Source code in pinnrl/pdes/pde_base.py

def weighted_total(self, losses, w_residual, w_boundary, w_initial) -> torch.Tensor:
    """Combine loss components under ``training.mode``.

    ``data_only`` drops the physics terms; outside ``forward`` the data term
    always counts (its weight is at least 1).
    """
    mode = self._training_option("mode", "forward")
    physics = 0.0 if mode == "data_only" else 1.0
    data_weight = float((self._training_option("loss_weights") or {}).get("data", 1.0))
    if mode != "forward" and data_weight <= 0.0:
        data_weight = 1.0
    return (
        physics
        * (
            w_residual * losses["residual"]
            + w_boundary * losses["boundary"]
            + w_initial * losses["initial"]
        )
        + data_weight * losses["data"]
    )

validate

validate(model, num_points=1000)

Error of model against the exact solution on a jittered grid.

Returns l2_error (relative L2, ||u - u*|| / ||u*||), mse, max_error and mean_error.

Raises:

Type	Description
ValueError	If the PDE has no exact solution configured.

Source code in pinnrl/pdes/pde_base.py

def validate(self, model: torch.nn.Module, num_points: int = 1000) -> dict[str, float]:
    """Error of ``model`` against the exact solution on a jittered grid.

    Returns ``l2_error`` (relative L2, ``||u - u*|| / ||u*||``), ``mse``,
    ``max_error`` and ``mean_error``.

    Raises:
        ValueError: If the PDE has no exact solution configured.
    """
    if not self.has_exact_solution:
        raise ValueError(f"{self.config.name} has no exact solution configured")
    x, t = self.generate_collocation_points(num_points)
    with torch.no_grad():
        u_pred = model(torch.cat([x, t], dim=1))[:, 0:1]
    u_exact = self.exact_solution(x, t)
    error = u_pred - u_exact
    return {
        "l2_error": (torch.linalg.norm(error) / torch.linalg.norm(u_exact)).item(),
        "mse": torch.mean(error**2).item(),
        "max_error": torch.max(torch.abs(error)).item(),
        "mean_error": torch.mean(torch.abs(error)).item(),
    }

HeatEquation

HeatEquation(config, **kwargs)

Bases: PDEBase

Heat equation u_t = alpha * ∇²u.

Parameters: alpha (thermal diffusivity, required).

Initial conditions, each with the exact solution of the same type (amplitude A, frequency k; L_d is the length of spatial axis d):

• sin_exp_decay (alias sine)::

  u = A · Π_d sin(2π k x_d / L_d) · exp(-alpha · Σ_d (2π k / L_d)² · t)

Each factor is an eigenfunction of ∂²/∂x_d² with eigenvalue -(2π k / L_d)², so the product decays at the summed rate. * sine_2d (2-D; frequency_x kx, frequency_y ky)::

  u = A · sin(kx π x) · sin(ky π y) · exp(-alpha π² (kx² + ky²) t)

Source code in pinnrl/pdes/heat_equation.py

def __init__(self, config: PDEConfig, **kwargs):
    super().__init__(config)

alpha property

alpha

Thermal diffusivity coefficient.

compute_residual

compute_residual(model, x, t)

u_t - alpha ∇²u.

Source code in pinnrl/pdes/heat_equation.py

def compute_residual(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """``u_t - alpha ∇²u``."""
    d = self.compute_derivatives(model, x, t, temporal_derivatives=[1], spatial_derivatives=[2])
    return d["dt"] - self.alpha * d["laplacian"]

WaveEquation

WaveEquation(config, **kwargs)

Bases: PDEBase

Wave equation u_tt = c² ∇²u.

Parameters: c (wave speed, default 1).

Second order in time: the initial loss also enforces u_t(x, 0) = initial_condition["velocity"] (default 0).

Initial conditions, each with the exact solution of the same type — standing waves that vanish on the faces of [0, 1]^d for integer frequencies (amplitude A, frequency k):

• sine: u = A Π_d sin(kπ x_d) · cos(ω t) with ω = c k π √d, since ∇² of the product is -d (kπ)² times itself.
• sine_2d (frequency_x kx, frequency_y ky): u = A sin(kx π x) sin(ky π y) cos(ω t) with ω = c π √(kx² + ky²).

Both start at rest (u_t(x, 0) = 0).

Source code in pinnrl/pdes/wave_equation.py

def __init__(self, config: PDEConfig, **kwargs):
    super().__init__(config)

c property

c

Wave speed.

compute_residual

compute_residual(model, x, t)

u_tt - c² ∇²u.

Source code in pinnrl/pdes/wave_equation.py

def compute_residual(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """``u_tt - c² ∇²u``."""
    d = self.compute_derivatives(model, x, t, temporal_derivatives=[2], spatial_derivatives=[2])
    return d["dt2"] - self.c**2 * d["laplacian"]

BurgersEquation

BurgersEquation(config, **kwargs)

Bases: PDEBase

Viscous Burgers equation u_t + u (Σ_d u_{x_d}) = nu ∇²u.

Parameters: nu (viscosity, default 0.01; viscosity is accepted as an alias).

Initial conditions: sine — u0 = A Π_d sin(kπ x_d) (amplitude A, default -1; frequency k, default 1).

Exact solution (1-D): cole_hopf — the Cole-Hopf integral for the configured sine initial condition and nu, evaluated by :func:pinnrl.benchmarks.reference.burgers_cole_hopf.

Source code in pinnrl/pdes/burgers_equation.py

def __init__(self, config: PDEConfig, **kwargs):
    super().__init__(config)

nu property

nu

Viscosity coefficient.

compute_residual

compute_residual(model, x, t)

u_t + u Σ_d u_{x_d} - nu ∇²u.

Source code in pinnrl/pdes/burgers_equation.py

def compute_residual(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """``u_t + u Σ_d u_{x_d} - nu ∇²u``."""
    d = self.compute_derivatives(
        model, x, t, temporal_derivatives=[1], spatial_derivatives=[1, 2]
    )
    u = d["u"]
    if self.dimension == 1:
        convection = u * d["dx"]
    else:
        convection = sum(u * d[f"dx{dim + 1}"] for dim in range(self.dimension))
    return d["dt"] + convection - self.nu * d["laplacian"]

KdVEquation

KdVEquation(config, **kwargs)

Bases: PDEBase

Korteweg-de Vries equation u_t + alpha u u_x + beta u_xxx = 0 (1-D).

Parameters: alpha (nonlinearity, default 6), beta (dispersion, default 1).

Initial condition and exact solution: soliton with speed c (default 1) and optional position x0::

u = (3c / alpha) · sech²( ½ √(c / beta) · (x - x0 - c t) )

Substituting u = f(x - c t) and integrating twice shows this travels at exactly c and decays to zero away from its crest at x0 + c t.

Source code in pinnrl/pdes/kdv_equation.py

def __init__(self, config: PDEConfig, **kwargs):
    if config.dimension != 1:
        raise ValueError("KdVEquation is 1-D only")
    super().__init__(config)

alpha property

alpha

Nonlinear coefficient.

beta property

beta

Dispersion coefficient.

speed property

speed

Speed of the soliton in the initial condition.

compute_residual

compute_residual(model, x, t)

u_t + alpha u u_x + beta u_xxx.

Source code in pinnrl/pdes/kdv_equation.py

def compute_residual(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """``u_t + alpha u u_x + beta u_xxx``."""
    d = self.compute_derivatives(
        model, x, t, temporal_derivatives=[1], spatial_derivatives=[1, 3]
    )
    return d["dt"] + self.alpha * d["u"] * d["dx"] + self.beta * d["dx3"]

ConvectionEquation

ConvectionEquation(config, **kwargs)

Bases: PDEBase

Linear advection u_t + v · ∇u = 0.

Parameters: velocity (a number, or one entry per spatial axis).

Initial conditions: sine (alias sin) — the plane wave u0 = A sin(kπ Σ_d x_d) (amplitude A, frequency k).

Exact solution: sine (alias sin) — the initial profile transported with the flow, u(x, t) = u0(x - v t).

Source code in pinnrl/pdes/convection_equation.py

def __init__(self, config: PDEConfig, **kwargs):
    super().__init__(config)

velocity property

velocity

Advection velocity, one entry per spatial axis.

compute_residual

compute_residual(model, x, t)

u_t + Σ_d v_d u_{x_d}.

Source code in pinnrl/pdes/convection_equation.py

def compute_residual(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """``u_t + Σ_d v_d u_{x_d}``."""
    x, t, u = self._prepare_inputs(model, x, t)
    grad_u = self._grad(u, x)
    convection = sum(v * grad_u[:, d : d + 1] for d, v in enumerate(self.velocity))
    return self._grad(u, t) + convection

AllenCahnEquation

AllenCahnEquation(config, **kwargs)

Bases: PDEBase

Allen-Cahn equation u_t = epsilon² ∇²u + r (u - u³).

Parameters: epsilon (interface width, default 0.1), reaction_rate r (default 1).

Initial conditions:

• tanh — the planar kink normal to (1, ..., 1)/√d, u = tanh(√r · Σ_d x_d / (√(2d) · epsilon)). It solves epsilon² φ'' + r (φ - φ³) = 0, so it is stationary.
• x2_cos — the benchmark u0 = x² cos(π x) (no closed form; score it against :func:pinnrl.benchmarks.reference.allen_cahn_reference).

Exact solution: tanh — the stationary kink above.

Source code in pinnrl/pdes/allen_cahn.py

def __init__(self, config: PDEConfig, **kwargs):
    super().__init__(config)

epsilon property

epsilon

Interface width parameter.

reaction_rate property

reaction_rate

Strength of the double-well reaction term.

compute_residual

compute_residual(model, x, t)

u_t - epsilon² ∇²u - r (u - u³).

Source code in pinnrl/pdes/allen_cahn.py

def compute_residual(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """``u_t - epsilon² ∇²u - r (u - u³)``."""
    d = self.compute_derivatives(model, x, t, temporal_derivatives=[1], spatial_derivatives=[2])
    u = d["u"]
    return d["dt"] - self.epsilon**2 * d["laplacian"] - self.reaction_rate * (u - u**3)

CahnHilliardEquation

CahnHilliardEquation(config, **kwargs)

Bases: PDEBase

Cahn-Hilliard equation u_t = M ∇²μ with μ = u³ - u - kappa ∇²u.

Parameters: mobility M (alias M, default 1) and kappa (gradient energy; if absent, epsilon² with epsilon default 0.1).

Initial conditions:

• tanh — the planar kink tanh(Σ_d x_d / √(2 d kappa)). It makes μ ≡ 0, so it is a stationary solution.
• random (from the base class) — a smooth, reproducible perturbation for spinodal-decomposition runs. No closed form.

Exact solution: tanh — the stationary kink above.

Source code in pinnrl/pdes/cahn_hilliard.py

def __init__(self, config: PDEConfig, **kwargs):
    super().__init__(config)

mobility property

mobility

Mobility M.

kappa property

kappa

Gradient-energy coefficient (interface width squared).

epsilon property

epsilon

Interface width sqrt(kappa).

compute_residual

compute_residual(model, x, t)

u_t - M ∇²(u³ - u - kappa ∇²u).

Source code in pinnrl/pdes/cahn_hilliard.py

def compute_residual(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """``u_t - M ∇²(u³ - u - kappa ∇²u)``."""
    x, t, u = self._prepare_inputs(model, x, t)
    # Clamp u inside μ so u³ cannot overflow for an untrained network.
    u_clamped = torch.clamp(u, -10.0, 10.0)
    mu = u_clamped**3 - u_clamped - self.kappa * self.laplacian(u, x)
    return self._grad(u, t) - self.mobility * self.laplacian(mu, x)

BlackScholesEquation

BlackScholesEquation(config, **kwargs)

Bases: PDEBase

Black-Scholes equation in time to maturity τ (the t input).

The PDE is::

V_τ = ½ σ² Σ_d S_d² V_{S_d S_d} + r Σ_d S_d V_{S_d} - r V

(in more than one dimension the assets are uncorrelated: no cross terms).

Writing it in τ = T - t turns the payoff at maturity into the initial condition at τ = 0, which is what the PINN loss imposes.

Parameters: sigma (volatility, default 0.2), r (risk-free rate, default 0.05).

Initial condition: option (alias call_option) — the payoff max(S - K, 0) for option_type: call (the default) or max(K - S, 0) for put, with strike K (alias strike_price, default 100) on S = Σ_d S_d.

Exact solution (1-D): black_scholes — the closed-form European price S Φ(d1) - K e^{-rτ} Φ(d2) (call) or K e^{-rτ} Φ(-d2) - S Φ(-d1) (put) for the configured option.

Source code in pinnrl/pdes/black_scholes.py

def __init__(self, config: PDEConfig, **kwargs):
    super().__init__(config)

sigma property

sigma

Volatility.

r property

r

Risk-free interest rate.

compute_residual

compute_residual(model, x, t)

V_τ - ½σ² Σ S² V_SS - r Σ S V_S + r V.

Source code in pinnrl/pdes/black_scholes.py

def compute_residual(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """``V_τ - ½σ² Σ S² V_SS - r Σ S V_S + r V``."""
    d = self.compute_derivatives(
        model, x, t, temporal_derivatives=[1], spatial_derivatives=[1, 2]
    )
    if self.dimension == 1:
        v_s, v_ss = d["dx"], d["dx2"]
    else:
        v_s = torch.cat([d[f"dx{i + 1}"] for i in range(self.dimension)], dim=1)
        v_ss = torch.cat([d["d" + f"x{i + 1}" * 2] for i in range(self.dimension)], dim=1)
    return (
        d["dt"]
        - 0.5 * self.sigma**2 * torch.sum(x**2 * v_ss, dim=1, keepdim=True)
        - self.r * torch.sum(x * v_s, dim=1, keepdim=True)
        + self.r * d["u"]
    )

PendulumEquation

PendulumEquation(config, **kwargs)

Bases: PDEBase

Damped nonlinear pendulum θ'' + b θ' + (g/L) sin θ = 0.

The angle θ(t) is the network output; the spatial input is a dummy coordinate the solution does not depend on.

Parameters: g (default 9.81), L (default 1), damping b (default 0).

Second order in time: the initial loss enforces θ(0) = initial_angle and θ'(0) = velocity (default 0).

Initial condition: small_angle (alias fixed) with initial_angle (alias value, default 0.5).

Exact solution: ode — the full nonlinear damped trajectory from the configured initial angle and velocity, integrated with scipy to 1e-11 relative tolerance (there is no elementary closed form; the small-angle formula θ0 cos(ωt) is only an approximation).

Source code in pinnrl/pdes/pendulum_equation.py

def __init__(self, config: PDEConfig, **kwargs):
    super().__init__(config)

g property

g

Gravitational acceleration.

L property

L

Pendulum length.

damping property

damping

Linear damping coefficient b.

compute_residual

compute_residual(model, x, t)

θ'' + b θ' + (g/L) sin θ.

Source code in pinnrl/pdes/pendulum_equation.py

def compute_residual(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """``θ'' + b θ' + (g/L) sin θ``."""
    d = self.compute_derivatives(model, x, t, temporal_derivatives=[1, 2])
    return d["dt2"] + self.damping * d["dt"] + (self.g / self.L) * torch.sin(d["u"])

compute_energy

compute_energy(model, x, t)

Mechanical energy per unit mass, ½ L² θ'² + g L (1 - cos θ).

Source code in pinnrl/pdes/pendulum_equation.py

def compute_energy(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> torch.Tensor:
    """Mechanical energy per unit mass, ``½ L² θ'² + g L (1 - cos θ)``."""
    d = self.compute_derivatives(model, x, t, temporal_derivatives=[1])
    return 0.5 * self.L**2 * d["dt"] ** 2 + self.g * self.L * (1 - torch.cos(d["u"]))

compute_phase_space

compute_phase_space(model, x, t)

Phase-space coordinates (θ, θ').

Source code in pinnrl/pdes/pendulum_equation.py

def compute_phase_space(
    self, model: torch.nn.Module, x: torch.Tensor, t: torch.Tensor
) -> tuple[torch.Tensor, torch.Tensor]:
    """Phase-space coordinates ``(θ, θ')``."""
    d = self.compute_derivatives(model, x, t, temporal_derivatives=[1])
    return d["u"], d["dt"]