anypinn.cli.scaffold.burgers_1d.ode_csv

Burgers Equation 1D — nonlinear PDE inverse problem definition (CSV data).

GRID_SIZE = 50 module-attribute

validation: ValidationRegistry = {NU_KEY: lambda x: torch.full_like(x, TRUE_NU)} module-attribute

BurgersResidualScorer

ResidualScorer protocol implementation for adaptive collocation.

Source code in src/anypinn/cli/scaffold/burgers_1d/ode_csv.py

class BurgersResidualScorer:
    """ResidualScorer protocol implementation for adaptive collocation."""

    def __init__(self, fields: FieldsRegistry, params: ParamsRegistry) -> None:
        self.fields = fields
        self.params = params

    def residual_score(self, x: Tensor) -> Tensor:
        device = next(iter(self.fields.values())).parameters().__next__().device
        x = x.detach().to(device).requires_grad_(True)
        with torch.enable_grad():
            res = burgers_residual(x, self.fields, self.params)
        # res may be (n, d) due to scalar param broadcasting; reduce to (n,)
        return res.detach().cpu().abs().mean(dim=-1)

fields = fields instance-attribute

params = params instance-attribute

__init__(fields: FieldsRegistry, params: ParamsRegistry) -> None

Source code in src/anypinn/cli/scaffold/burgers_1d/ode_csv.py

def __init__(self, fields: FieldsRegistry, params: ParamsRegistry) -> None:
    self.fields = fields
    self.params = params

residual_score(x: Tensor) -> Tensor

Source code in src/anypinn/cli/scaffold/burgers_1d/ode_csv.py

def residual_score(self, x: Tensor) -> Tensor:
    device = next(iter(self.fields.values())).parameters().__next__().device
    x = x.detach().to(device).requires_grad_(True)
    with torch.enable_grad():
        res = burgers_residual(x, self.fields, self.params)
    # res may be (n, d) due to scalar param broadcasting; reduce to (n,)
    return res.detach().cpu().abs().mean(dim=-1)

burgers_residual(x: Tensor, fields: FieldsRegistry, params: ParamsRegistry) -> Tensor

PDE residual: du/dt + udu/dx - nud2u/dx2 = 0.

Source code in src/anypinn/cli/scaffold/burgers_1d/ode_csv.py

def burgers_residual(x: Tensor, fields: FieldsRegistry, params: ParamsRegistry) -> Tensor:
    """PDE residual: du/dt + u*du/dx - nu*d2u/dx2 = 0."""
    u = fields[U_KEY](x)
    nu = params[NU_KEY](x)
    du_dt = partial(u, x, dim=1, order=1)
    du_dx = partial(u, x, dim=0, order=1)
    d2u_dx2 = partial(u, x, dim=0, order=2)
    return du_dt + u * du_dx - nu * d2u_dx2

create_data_module(hp: PINNHyperparameters, fields: FieldsRegistry, params: ParamsRegistry) -> Burgers1DDataModule

Source code in src/anypinn/cli/scaffold/burgers_1d/ode_csv.py

def create_data_module(
    hp: PINNHyperparameters,
    fields: FieldsRegistry,
    params: ParamsRegistry,
) -> Burgers1DDataModule:
    scorer = BurgersResidualScorer(fields, params)
    return Burgers1DDataModule(
        hp=hp,
        true_nu=TRUE_NU,
        grid_size=GRID_SIZE,
        residual_scorer=scorer,
        validation=validation,
    )

create_problem(hp: PINNHyperparameters) -> Problem

Source code in src/anypinn/cli/scaffold/burgers_1d/ode_csv.py

def create_problem(hp: PINNHyperparameters) -> Problem:
    rff = RandomFourierFeatures(in_dim=2, num_features=128, scale=1.0, seed=42)
    field_u = Field(
        config=MLPConfig(
            in_dim=rff.out_dim,
            out_dim=1,
            hidden_layers=hp.fields_config.hidden_layers,
            activation=hp.fields_config.activation,
            output_activation=hp.fields_config.output_activation,
            encode=rff,
        )
    )
    param_nu = Parameter(
        config=ScalarConfig(init_value=hp.params_config.init_value),
    )

    fields = FieldsRegistry({U_KEY: field_u})
    params = ParamsRegistry({NU_KEY: param_nu})

    bcs = [
        DirichletBCConstraint(
            BoundaryCondition(sampler=_left_boundary, value=_zero, n_pts=100),
            field_u,
            log_key="loss/bc_left",
            weight=10.0,
        ),
        DirichletBCConstraint(
            BoundaryCondition(sampler=_right_boundary, value=_zero, n_pts=100),
            field_u,
            log_key="loss/bc_right",
            weight=10.0,
        ),
        DirichletBCConstraint(
            BoundaryCondition(sampler=_initial_condition, value=_ic_value, n_pts=100),
            field_u,
            log_key="loss/ic",
            weight=10.0,
        ),
    ]

    pde = PDEResidualConstraint(
        fields=fields,
        params=params,
        residual_fn=burgers_residual,
        log_key="loss/pde_residual",
        weight=1.0,
    )

    data = DataConstraint(
        fields=fields,
        params=params,
        predict_data=predict_data,
        weight=5.0,
    )

    return Problem(
        constraints=[pde, *bcs, data],
        criterion=build_criterion(hp.criterion),
        fields=fields,
        params=params,
    )

predict_data(x_data: Tensor, fields: FieldsRegistry, _params: ParamsRegistry) -> Tensor

Source code in src/anypinn/cli/scaffold/burgers_1d/ode_csv.py

def predict_data(x_data: Tensor, fields: FieldsRegistry, _params: ParamsRegistry) -> Tensor:
    return fields[U_KEY](x_data).unsqueeze(1)