Constraint(约束条件) 模块

ppsci.constraint

Constraint

Base class for constraint.

Parameters:

Name	Type	Description	Default
dataset	Dataset	Dataset.	required
dataloader_cfg	Dict[str, Any]	Dataloader config.	required
loss	Loss	Loss functor.	required
name	str	Name of constraint.	required

Source code in ppsci/constraint/base.py

class Constraint:
    """Base class for constraint.

    Args:
        dataset (io.Dataset): Dataset.
        dataloader_cfg (Dict[str, Any]): Dataloader config.
        loss (loss.Loss): Loss functor.
        name (str): Name of constraint.
    """

    def __init__(
        self,
        dataset: io.Dataset,
        dataloader_cfg: Dict[str, Any],
        loss: "loss.Loss",
        name: str,
    ):
        self.data_loader = data.build_dataloader(dataset, dataloader_cfg)
        self.data_iter = iter(self.data_loader)
        self.loss = loss
        self.name = name

    def __str__(self):
        return ", ".join(
            [
                self.__class__.__name__,
                f"name = {self.name}",
                f"input_keys = {self.input_keys}",
                f"output_keys = {self.output_keys}",
                f"output_expr = {self.output_expr}",
                f"label_dict = {self.label_dict}",
                f"loss = {self.loss}",
            ]
        )

BoundaryConstraint

Bases: Constraint

Class for boundary constraint.

Parameters:

Name	Type	Description	Default
output_expr	Dict[str, Callable]	Function in dict for computing output. e.g. {"u_mul_v": lambda out: out["u"] * out["v"]} means the model output u will be multiplied by model output v and the result will be named "u_mul_v".	required
label_dict	Dict[str, Union[float, Callable]]	Function in dict for computing label, which will be a reference value to participate in the loss calculation.	required
geom	Geometry	Geometry where data sampled from.	required
dataloader_cfg	Dict[str, Any]	Dataloader config.	required
loss	Loss	Loss functor.	required
random	Literal['pseudo', 'Halton', 'LHS']	Random method for sampling data in geometry. Defaults to "pseudo".	'pseudo'
criteria	Optional[Callable]	Criteria for refining specified boundaries. Defaults to None.	None
evenly	bool	Whether to use evenly distribution sampling. Defaults to False.	False
weight_dict	Optional[Dict[str, Union[float, Callable]]]	Define the weight of each constraint variable. Defaults to None.	None
name	str	Name of constraint object. Defaults to "BC".	'BC'

Examples:

>>> import ppsci
>>> rect = ppsci.geometry.Rectangle((0, 0), (1, 1))
>>> bc = ppsci.constraint.BoundaryConstraint(
...     {"u": lambda out: out["u"]},
...     {"u": 0},
...     rect,
...     {
...         "dataset": "IterableNamedArrayDataset",
...         "iters_per_epoch": 1,
...         "batch_size": 16,
...     },
...     ppsci.loss.MSELoss("mean"),
...     name="BC",
... )

Source code in ppsci/constraint/boundary_constraint.py

class BoundaryConstraint(base.Constraint):
    """Class for boundary constraint.

    Args:
        output_expr (Dict[str, Callable]): Function in dict for computing output.
            e.g. {"u_mul_v": lambda out: out["u"] * out["v"]} means the model output u
            will be multiplied by model output v and the result will be named "u_mul_v".
        label_dict (Dict[str, Union[float, Callable]]): Function in dict for computing
            label, which will be a reference value to participate in the loss calculation.
        geom (geometry.Geometry): Geometry where data sampled from.
        dataloader_cfg (Dict[str, Any]): Dataloader config.
        loss (loss.Loss): Loss functor.
        random (Literal["pseudo", "Halton", "LHS"], optional): Random method for sampling data in
            geometry. Defaults to "pseudo".
        criteria (Optional[Callable]): Criteria for refining specified boundaries.
            Defaults to None.
        evenly (bool, optional): Whether to use evenly distribution sampling.
            Defaults to False.
        weight_dict (Optional[Dict[str, Union[float, Callable]]]): Define the weight of each
            constraint variable. Defaults to None.
        name (str, optional): Name of constraint object. Defaults to "BC".

    Examples:
        >>> import ppsci
        >>> rect = ppsci.geometry.Rectangle((0, 0), (1, 1))
        >>> bc = ppsci.constraint.BoundaryConstraint(
        ...     {"u": lambda out: out["u"]},
        ...     {"u": 0},
        ...     rect,
        ...     {
        ...         "dataset": "IterableNamedArrayDataset",
        ...         "iters_per_epoch": 1,
        ...         "batch_size": 16,
        ...     },
        ...     ppsci.loss.MSELoss("mean"),
        ...     name="BC",
        ... ) # doctest: +SKIP
    """

    def __init__(
        self,
        output_expr: Dict[str, Callable],
        label_dict: Dict[str, Union[float, Callable]],
        geom: geometry.Geometry,
        dataloader_cfg: Dict[str, Any],
        loss: "loss.Loss",
        random: Literal["pseudo", "Halton", "LHS"] = "pseudo",
        criteria: Optional[Callable] = None,
        evenly: bool = False,
        weight_dict: Optional[Dict[str, Union[float, Callable]]] = None,
        name: str = "BC",
    ):
        self.label_dict = label_dict
        self.input_keys = geom.dim_keys
        self.output_keys = tuple(label_dict.keys())
        self.output_expr = {
            k: v for k, v in output_expr.items() if k in self.output_keys
        }

        if isinstance(criteria, str):
            criteria = eval(criteria)

        # prepare input
        input = geom.sample_boundary(
            dataloader_cfg["batch_size"] * dataloader_cfg["iters_per_epoch"],
            random,
            criteria,
            evenly,
        )
        if "area" in input:
            input["area"] *= dataloader_cfg["iters_per_epoch"]

        # prepare label
        label = {}
        for key, value in label_dict.items():
            if isinstance(value, (int, float)):
                label[key] = np.full_like(next(iter(input.values())), value)
            elif isinstance(value, sympy.Basic):
                func = sympy.lambdify(
                    sympy.symbols(geom.dim_keys),
                    value,
                    [{"amax": lambda xy, axis: np.maximum(xy[0], xy[1])}, "numpy"],
                )
                label[key] = func(
                    **{k: v for k, v in input.items() if k in geom.dim_keys}
                )
            elif callable(value):
                func = value
                label[key] = func(input)
                if isinstance(label[key], (int, float)):
                    label[key] = np.full_like(next(iter(input.values())), label[key])
            else:
                raise NotImplementedError(f"type of {type(value)} is invalid yet.")

        # prepare weight
        weight = None
        if weight_dict is not None:
            weight = {key: np.ones_like(next(iter(label.values()))) for key in label}
            for key, value in weight_dict.items():
                if isinstance(value, (int, float)):
                    weight[key] = np.full_like(next(iter(label.values())), value)
                elif isinstance(value, sympy.Basic):
                    func = sympy.lambdify(
                        [sympy.Symbol(k) for k in geom.dim_keys],
                        value,
                        [{"amax": lambda xy, _: np.maximum(xy[0], xy[1])}, "numpy"],
                    )
                    weight[key] = func(**{k: input[k] for k in geom.dim_keys})
                elif callable(value):
                    func = value
                    weight[key] = func(input)
                    if isinstance(weight[key], (int, float)):
                        weight[key] = np.full_like(
                            next(iter(input.values())), weight[key]
                        )
                else:
                    raise NotImplementedError(f"type of {type(value)} is invalid yet.")

        # wrap input, label, weight into a dataset
        if isinstance(dataloader_cfg["dataset"], str):
            dataloader_cfg["dataset"] = {"name": dataloader_cfg["dataset"]}
        dataloader_cfg["dataset"].update(
            {"input": input, "label": label, "weight": weight}
        )
        _dataset = dataset.build_dataset(dataloader_cfg["dataset"])

        # construct dataloader with dataset and dataloader_cfg
        super().__init__(_dataset, dataloader_cfg, loss, name)

InitialConstraint

Bases: Constraint

Class for initial interior constraint.

Parameters:

Name	Type	Description	Default
output_expr	Dict[str, Callable]	Function in dict for computing output. e.g. {"u_mul_v": lambda out: out["u"] * out["v"]} means the model output u will be multiplied by model output v and the result will be named "u_mul_v".	required
label_dict	Dict[str, Union[float, Callable]]	Function in dict for computing label, which will be a reference value to participate in the loss calculation.	required
geom	TimeXGeometry	Geometry where data sampled from.	required
dataloader_cfg	Dict[str, Any]	Dataloader config.	required
loss	Loss	Loss functor.	required
random	Literal['pseudo', 'Halton', 'LHS']	Random method for sampling data in geometry. Defaults to "pseudo".	'pseudo'
criteria	Optional[Callable]	Criteria for refining specified boundaries. Defaults to None.	None
evenly	bool	Whether to use evenly distribution sampling. Defaults to False.	False
weight_dict	Optional[Dict[str, Callable]]	Define the weight of each constraint variable. Defaults to None.	None
compute_sdf_derivatives	Optional[bool]	Whether compute derivatives for SDF. Defaults to False.	False
name	str	Name of constraint object. Defaults to "IC".	'IC'

Examples:

>>> import ppsci
>>> rect = ppsci.geometry.TimeXGeometry(
...     ppsci.geometry.TimeDomain(0, 1),
...     ppsci.geometry.Rectangle((0, 0), (1, 1)),
... )
>>> ic = ppsci.constraint.InitialConstraint(
...     {"u": lambda out: out["u"]},
...     {"u": 0},
...     rect,
...     {
...         "dataset": "IterableNamedArrayDataset",
...         "iters_per_epoch": 1,
...         "batch_size": 16,
...     },
...     ppsci.loss.MSELoss("mean"),
...     name="IC",
... )

Source code in ppsci/constraint/initial_constraint.py

class InitialConstraint(base.Constraint):
    """Class for initial interior constraint.

    Args:
        output_expr (Dict[str, Callable]): Function in dict for computing output.
            e.g. {"u_mul_v": lambda out: out["u"] * out["v"]} means the model output u
            will be multiplied by model output v and the result will be named "u_mul_v".
        label_dict (Dict[str, Union[float, Callable]]): Function in dict for computing
            label, which will be a reference value to participate in the loss calculation.
        geom (geometry.TimeXGeometry): Geometry where data sampled from.
        dataloader_cfg (Dict[str, Any]): Dataloader config.
        loss (loss.Loss): Loss functor.
        random (Literal["pseudo", "Halton", "LHS"], optional): Random method for sampling data in
            geometry. Defaults to "pseudo".
        criteria (Optional[Callable]): Criteria for refining specified boundaries.
            Defaults to None.
        evenly (bool, optional): Whether to use evenly distribution sampling.
            Defaults to False.
        weight_dict (Optional[Dict[str, Callable]]): Define the weight of each
            constraint variable. Defaults to None.
        compute_sdf_derivatives (Optional[bool]): Whether compute derivatives for SDF.
            Defaults to False.
        name (str, optional): Name of constraint object. Defaults to "IC".

    Examples:
        >>> import ppsci
        >>> rect = ppsci.geometry.TimeXGeometry(
        ...     ppsci.geometry.TimeDomain(0, 1),
        ...     ppsci.geometry.Rectangle((0, 0), (1, 1)),
        ... )
        >>> ic = ppsci.constraint.InitialConstraint(
        ...     {"u": lambda out: out["u"]},
        ...     {"u": 0},
        ...     rect,
        ...     {
        ...         "dataset": "IterableNamedArrayDataset",
        ...         "iters_per_epoch": 1,
        ...         "batch_size": 16,
        ...     },
        ...     ppsci.loss.MSELoss("mean"),
        ...     name="IC",
        ... ) # doctest: +SKIP
    """

    def __init__(
        self,
        output_expr: Dict[str, Callable],
        label_dict: Dict[str, Union[float, Callable]],
        geom: geometry.TimeXGeometry,
        dataloader_cfg: Dict[str, Any],
        loss: "loss.Loss",
        random: Literal["pseudo", "Halton", "LHS"] = "pseudo",
        criteria: Optional[Callable] = None,
        evenly: bool = False,
        weight_dict: Optional[Dict[str, Callable]] = None,
        compute_sdf_derivatives: bool = False,
        name: str = "IC",
    ):
        self.label_dict = label_dict
        self.input_keys = geom.dim_keys
        self.output_keys = tuple(label_dict.keys())
        self.output_expr = {
            k: v for k, v in output_expr.items() if k in self.output_keys
        }

        if isinstance(criteria, str):
            criteria = eval(criteria)

        # prepare input
        input = geom.sample_initial_interior(
            dataloader_cfg["batch_size"] * dataloader_cfg["iters_per_epoch"],
            random,
            criteria,
            evenly,
            compute_sdf_derivatives,
        )
        if "area" in input:
            input["area"] *= dataloader_cfg["iters_per_epoch"]

        # prepare label
        label = {}
        for key, value in label_dict.items():
            if isinstance(value, (int, float)):
                label[key] = np.full_like(next(iter(input.values())), value)
            elif isinstance(value, sympy.Basic):
                func = sympy.lambdify(
                    sympy.symbols(geom.dim_keys),
                    value,
                    [{"amax": lambda xy, _: np.maximum(xy[0], xy[1])}, "numpy"],
                )
                label[key] = func(
                    **{k: v for k, v in input.items() if k in geom.dim_keys}
                )
            elif callable(value):
                func = value
                label[key] = func(input)
                if isinstance(label[key], (int, float)):
                    label[key] = np.full_like(next(iter(input.values())), label[key])
            else:
                raise NotImplementedError(f"type of {type(value)} is invalid yet.")

        # prepare weight
        weight = None
        if weight_dict is not None:
            weight = {key: np.ones_like(next(iter(label.values()))) for key in label}
            for key, value in weight_dict.items():
                if isinstance(value, (int, float)):
                    weight[key] = np.full_like(next(iter(label.values())), value)
                elif isinstance(value, sympy.Basic):
                    func = sympy.lambdify(
                        sympy.symbols(geom.dim_keys),
                        value,
                        [{"amax": lambda xy, _: np.maximum(xy[0], xy[1])}, "numpy"],
                    )
                    weight[key] = func(
                        **{k: v for k, v in input.items() if k in geom.dim_keys}
                    )
                elif callable(value):
                    func = value
                    weight[key] = func(input)
                    if isinstance(weight[key], (int, float)):
                        weight[key] = np.full_like(
                            next(iter(input.values())), weight[key]
                        )
                else:
                    raise NotImplementedError(f"type of {type(value)} is invalid yet.")

        # wrap input, label, weight into a dataset
        if isinstance(dataloader_cfg["dataset"], str):
            dataloader_cfg["dataset"] = {"name": dataloader_cfg["dataset"]}
        dataloader_cfg["dataset"].update(
            {"input": input, "label": label, "weight": weight}
        )
        _dataset = dataset.build_dataset(dataloader_cfg["dataset"])

        # construct dataloader with dataset and dataloader_cfg
        super().__init__(_dataset, dataloader_cfg, loss, name)

IntegralConstraint

Bases: Constraint

Class for integral constraint.

Parameters:

Name	Type	Description	Default
output_expr	Dict[str, Callable]	Function in dict for computing output. e.g. {"u_mul_v": lambda out: out["u"] * out["v"]} means the model output u will be multiplied by model output v and the result will be named "u_mul_v".	required
label_dict	Dict[str, Union[float, Callable]]	Function in dict for computing label, which will be a reference value to participate in the loss calculation.	required
geom	Geometry	Geometry where data sampled from.	required
dataloader_cfg	Dict[str, Any]	Dataloader config.	required
loss	Loss	Loss functor.	required
random	Literal['pseudo', 'Halton', 'LHS']	Random method for sampling data in geometry. Defaults to "pseudo".	'pseudo'
criteria	Optional[Callable]	Criteria for refining specified boundaries. Defaults to None.	None
weight_dict	Optional[Dict[str, Callable]]	Define the weight of each constraint variable. Defaults to None.	None
name	str	Name of constraint object. Defaults to "IgC".	'IgC'

Examples:

>>> import ppsci
>>> rect = ppsci.geometry.Rectangle((0, 0), (1, 1))
>>> igc = ppsci.constraint.IntegralConstraint(
...     {"u": lambda out: out["u"]},
...     {"u": 0},
...     rect,
...     {
...         "dataset": "IterableNamedArrayDataset",
...         "iters_per_epoch": 1,
...         "batch_size": 16,
...         "integral_batch_size": 8,
...     },
...     ppsci.loss.MSELoss("mean"),
...     name="IgC",
... )

Source code in ppsci/constraint/integral_constraint.py

class IntegralConstraint(base.Constraint):
    """Class for integral constraint.

    Args:
        output_expr (Dict[str, Callable]): Function in dict for computing output.
            e.g. {"u_mul_v": lambda out: out["u"] * out["v"]} means the model output u
            will be multiplied by model output v and the result will be named "u_mul_v".
        label_dict (Dict[str, Union[float, Callable]]): Function in dict for computing
            label, which will be a reference value to participate in the loss calculation.
        geom (geometry.Geometry): Geometry where data sampled from.
        dataloader_cfg (Dict[str, Any]): Dataloader config.
        loss (loss.Loss): Loss functor.
        random (Literal["pseudo", "Halton", "LHS"], optional): Random method for sampling data in
            geometry. Defaults to "pseudo".
        criteria (Optional[Callable]): Criteria for refining specified boundaries.
            Defaults to None.
        weight_dict (Optional[Dict[str, Callable]]): Define the weight of each
            constraint variable. Defaults to None.
        name (str, optional): Name of constraint object. Defaults to "IgC".

    Examples:
        >>> import ppsci
        >>> rect = ppsci.geometry.Rectangle((0, 0), (1, 1))
        >>> igc = ppsci.constraint.IntegralConstraint(
        ...     {"u": lambda out: out["u"]},
        ...     {"u": 0},
        ...     rect,
        ...     {
        ...         "dataset": "IterableNamedArrayDataset",
        ...         "iters_per_epoch": 1,
        ...         "batch_size": 16,
        ...         "integral_batch_size": 8,
        ...     },
        ...     ppsci.loss.MSELoss("mean"),
        ...     name="IgC",
        ... ) # doctest: +SKIP
    """

    def __init__(
        self,
        output_expr: Dict[str, Callable],
        label_dict: Dict[str, Union[float, Callable]],
        geom: geometry.Geometry,
        dataloader_cfg: Dict[str, Any],
        loss: "loss.Loss",
        random: Literal["pseudo", "Halton", "LHS"] = "pseudo",
        criteria: Optional[Callable] = None,
        weight_dict: Optional[Dict[str, Callable]] = None,
        name: str = "IgC",
    ):
        self.label_dict = label_dict
        self.input_keys = geom.dim_keys
        self.output_keys = tuple(label_dict.keys())
        self.output_expr = {
            k: v for k, v in output_expr.items() if k in self.output_keys
        }

        if isinstance(criteria, str):
            criteria = eval(criteria)

        # prepare input
        input_list: List[Dict[str, np.ndarray]] = []
        for _ in range(
            dataloader_cfg["batch_size"] * dataloader_cfg["iters_per_epoch"]
        ):
            input = geom.sample_boundary(
                dataloader_cfg["integral_batch_size"], random, criteria
            )
            input_list.append(input)
        input = misc.stack_dict_list(input_list)
        # shape of each input is [batch_size, integral_batch_size, ndim]

        # prepare label
        # shape of each label is [batch_size, ndim]
        label = {}
        for key, value in label_dict.items():
            if isinstance(value, (int, float)):
                label[key] = np.full(
                    (next(iter(input.values())).shape[0], 1),
                    value,
                    paddle.get_default_dtype(),
                )
            elif isinstance(value, sympy.Basic):
                func = sympy.lambdify(
                    sympy.symbols(geom.dim_keys),
                    value,
                    [{"amax": lambda xy, _: np.maximum(xy[0], xy[1])}, "numpy"],
                )
                label[key] = func(
                    **{k: v for k, v in input.items() if k in geom.dim_keys}
                )
            elif callable(value):
                func = value
                label[key] = func(input)
                if isinstance(label[key], (int, float)):
                    label[key] = np.full(
                        (next(iter(input.values())).shape[0], 1),
                        label[key],
                        paddle.get_default_dtype(),
                    )
            else:
                raise NotImplementedError(f"type of {type(value)} is invalid yet.")

        # prepare weight
        # shape of each weight is [batch_size, ndim]
        weight = None
        if weight_dict is not None:
            weight = {key: np.ones_like(next(iter(label.values()))) for key in label}
            for key, value in weight_dict.items():
                if isinstance(value, (int, float)):
                    weight[key] = np.full_like(next(iter(label.values())), value)
                elif isinstance(value, sympy.Basic):
                    func = sympy.lambdify(
                        sympy.symbols(geom.dim_keys),
                        value,
                        [{"amax": lambda xy, _: np.maximum(xy[0], xy[1])}, "numpy"],
                    )
                    weight[key] = func(
                        **{k: v for k, v in input.items() if k in geom.dim_keys}
                    )
                elif callable(value):
                    func = value
                    weight[key] = func(input)
                    if isinstance(weight[key], (int, float)):
                        weight[key] = np.full_like(
                            next(iter(input.values())), weight[key]
                        )
                else:
                    raise NotImplementedError(f"type of {type(value)} is invalid yet.")

        # wrap input, label, weight into a dataset
        if isinstance(dataloader_cfg["dataset"], str):
            dataloader_cfg["dataset"] = {"name": dataloader_cfg["dataset"]}
        dataloader_cfg["dataset"].update(
            {"input": input, "label": label, "weight": weight}
        )
        _dataset = dataset.build_dataset(dataloader_cfg["dataset"])

        # construct dataloader with dataset and dataloader_cfg
        super().__init__(_dataset, dataloader_cfg, loss, name)

InteriorConstraint

Bases: Constraint

Class for interior constraint.

Parameters:

Name	Type	Description	Default
output_expr	Dict[str, Callable]	Function in dict for computing output. e.g. {"u_mul_v": lambda out: out["u"] * out["v"]} means the model output u will be multiplied by model output v and the result will be named "u_mul_v".	required
label_dict	Dict[str, Union[float, Callable]]	Function in dict for computing label, which will be a reference value to participate in the loss calculation.	required
geom	Geometry	Geometry where data sampled from.	required
dataloader_cfg	Dict[str, Any]	Dataloader config.	required
loss	Loss	Loss functor.	required
random	Literal['pseudo', 'Halton', 'LHS']	Random method for sampling data in geometry. Defaults to "pseudo".	'pseudo'
criteria	Optional[Callable]	Criteria for refining specified boundaries. Defaults to None.	None
evenly	bool	Whether to use evenly distribution sampling. Defaults to False.	False
weight_dict	Optional[Dict[str, Union[Callable, float]]]	Define the weight of each constraint variable. Defaults to None.	None
compute_sdf_derivatives	Optional[bool]	Whether compute derivatives for SDF. Defaults to False.	False
name	str	Name of constraint object. Defaults to "EQ".	'EQ'

Examples:

>>> import ppsci
>>> rect = ppsci.geometry.Rectangle((0, 0), (1, 1))
>>> pde_constraint = ppsci.constraint.InteriorConstraint(
...     {"u": lambda out: out["u"]},
...     {"u": 0},
...     rect,
...     {
...         "dataset": "IterableNamedArrayDataset",
...         "iters_per_epoch": 1,
...         "batch_size": 16,
...     },
...     ppsci.loss.MSELoss("mean"),
...     name="EQ",
... )

Source code in ppsci/constraint/interior_constraint.py

class InteriorConstraint(base.Constraint):
    """Class for interior constraint.

    Args:
        output_expr (Dict[str, Callable]): Function in dict for computing output.
            e.g. {"u_mul_v": lambda out: out["u"] * out["v"]} means the model output u
            will be multiplied by model output v and the result will be named "u_mul_v".
        label_dict (Dict[str, Union[float, Callable]]): Function in dict for computing
            label, which will be a reference value to participate in the loss calculation.
        geom (geometry.Geometry): Geometry where data sampled from.
        dataloader_cfg (Dict[str, Any]): Dataloader config.
        loss (loss.Loss): Loss functor.
        random (Literal["pseudo", "Halton", "LHS"], optional): Random method for sampling data in
            geometry. Defaults to "pseudo".
        criteria (Optional[Callable]): Criteria for refining specified boundaries.
            Defaults to None.
        evenly (bool, optional): Whether to use evenly distribution sampling.
            Defaults to False.
        weight_dict (Optional[Dict[str, Union[Callable, float]]]): Define the
            weight of each constraint variable. Defaults to None.
        compute_sdf_derivatives (Optional[bool]): Whether compute derivatives for SDF.
            Defaults to False.
        name (str, optional): Name of constraint object. Defaults to "EQ".

    Examples:
        >>> import ppsci
        >>> rect = ppsci.geometry.Rectangle((0, 0), (1, 1))
        >>> pde_constraint = ppsci.constraint.InteriorConstraint(
        ...     {"u": lambda out: out["u"]},
        ...     {"u": 0},
        ...     rect,
        ...     {
        ...         "dataset": "IterableNamedArrayDataset",
        ...         "iters_per_epoch": 1,
        ...         "batch_size": 16,
        ...     },
        ...     ppsci.loss.MSELoss("mean"),
        ...     name="EQ",
        ... ) # doctest: +SKIP
    """

    def __init__(
        self,
        output_expr: Dict[str, Callable],
        label_dict: Dict[str, Union[float, Callable]],
        geom: geometry.Geometry,
        dataloader_cfg: Dict[str, Any],
        loss: "loss.Loss",
        random: Literal["pseudo", "Halton", "LHS"] = "pseudo",
        criteria: Optional[Callable] = None,
        evenly: bool = False,
        weight_dict: Optional[Dict[str, Union[Callable, float]]] = None,
        compute_sdf_derivatives: bool = False,
        name: str = "EQ",
    ):
        self.label_dict = label_dict
        self.input_keys = geom.dim_keys
        self.output_keys = tuple(label_dict.keys())
        self.output_expr = {
            k: v for k, v in output_expr.items() if k in self.output_keys
        }

        if isinstance(criteria, str):
            criteria = eval(criteria)

        # prepare input
        input = geom.sample_interior(
            dataloader_cfg["batch_size"] * dataloader_cfg["iters_per_epoch"],
            random,
            criteria,
            evenly,
            compute_sdf_derivatives,
        )
        if "area" in input:
            input["area"] *= dataloader_cfg["iters_per_epoch"]

        # prepare label
        label = {}
        for key, value in label_dict.items():
            if isinstance(value, (int, float)):
                label[key] = np.full_like(next(iter(input.values())), value)
            elif isinstance(value, sympy.Basic):
                func = sympy.lambdify(
                    sympy.symbols(geom.dim_keys),
                    value,
                    [{"amax": lambda xy, _: np.maximum(xy[0], xy[1])}, "numpy"],
                )
                label[key] = func(
                    **{k: v for k, v in input.items() if k in geom.dim_keys}
                )
            elif callable(value):
                func = value
                label[key] = func(input)
                if isinstance(label[key], (int, float)):
                    label[key] = np.full_like(next(iter(input.values())), label[key])
            else:
                raise NotImplementedError(f"type of {type(value)} is invalid yet.")

        # prepare weight
        weight = None
        if weight_dict is not None:
            weight = {key: np.ones_like(next(iter(label.values()))) for key in label}
            for key, value in weight_dict.items():
                if isinstance(value, str):
                    if value == "sdf":
                        weight[key] = input["sdf"]
                    else:
                        raise NotImplementedError(f"string {value} is invalid yet.")
                elif isinstance(value, (int, float)):
                    weight[key] = np.full_like(next(iter(label.values())), float(value))
                elif isinstance(value, sympy.Basic):
                    func = sympy.lambdify(
                        sympy.symbols(geom.dim_keys),
                        value,
                        [{"amax": lambda xy, _: np.maximum(xy[0], xy[1])}, "numpy"],
                    )
                    weight[key] = func(
                        **{k: v for k, v in input.items() if k in geom.dim_keys}
                    )
                elif callable(value):
                    func = value
                    weight[key] = func(input)
                    if isinstance(weight[key], (int, float)):
                        weight[key] = np.full_like(
                            next(iter(input.values())), weight[key]
                        )
                else:
                    raise NotImplementedError(f"type of {type(value)} is invalid yet.")

        # wrap input, label, weight into a dataset
        if isinstance(dataloader_cfg["dataset"], str):
            dataloader_cfg["dataset"] = {"name": dataloader_cfg["dataset"]}
        dataloader_cfg["dataset"].update(
            {"input": input, "label": label, "weight": weight}
        )
        _dataset = dataset.build_dataset(dataloader_cfg["dataset"])

        # construct dataloader with dataset and dataloader_cfg
        super().__init__(_dataset, dataloader_cfg, loss, name)

PeriodicConstraint

Bases: Constraint

Class for periodic constraint.

Parameters:

Name	Type	Description	Default
output_expr	Dict[str, Callable]	Function in dict for computing output. e.g. {"u_mul_v": lambda out: out["u"] * out["v"]} means the model output u will be multiplied by model output v and the result will be named "u_mul_v".	required
label_dict	Dict[str, Union[float, Callable]]	Function in dict for computing label, which will be a reference value to participate in the loss calculation.	required
geom	Geometry	Geometry where data sampled from.	required
dataloader_cfg	Dict[str, Any]	Dataloader config.	required
periodic_key	str	Name of dimension which periodic constraint applied to.	required
loss	Loss	Loss functor.	required
random	Literal['pseudo', 'Halton', 'LHS']	Random method for sampling data in geometry. Defaults to "pseudo".	'pseudo'
criteria	Optional[Callable]	Criteria for refining specified boundaries. Defaults to None.	None
evenly	bool	Whether to use evenly distribution sampling. Defaults to False.	False
weight_dict	Optional[Dict[str, Callable]]	Define the weight of each constraint variable. Defaults to None.	None
name	str	Name of constraint object. Defaults to "PeriodicBC".	'PeriodicBC'

Source code in ppsci/constraint/periodic_constraint.py

class PeriodicConstraint(base.Constraint):
    """Class for periodic constraint.

    Args:
        output_expr (Dict[str, Callable]): Function in dict for computing output.
            e.g. {"u_mul_v": lambda out: out["u"] * out["v"]} means the model output u
            will be multiplied by model output v and the result will be named "u_mul_v".
        label_dict (Dict[str, Union[float, Callable]]): Function in dict for computing
            label, which will be a reference value to participate in the loss calculation.
        geom (geometry.Geometry): Geometry where data sampled from.
        dataloader_cfg (Dict[str, Any]): Dataloader config.
        periodic_key (str): Name of dimension which periodic constraint applied to.
        loss (loss.Loss): Loss functor.
        random (Literal["pseudo", "Halton", "LHS"], optional): Random method for sampling data in
            geometry. Defaults to "pseudo".
        criteria (Optional[Callable]): Criteria for refining specified boundaries.
            Defaults to None.
        evenly (bool, optional):  Whether to use evenly distribution sampling.
            Defaults to False.
        weight_dict (Optional[Dict[str, Callable]]): Define the weight of each
            constraint variable. Defaults to None.
        name (str, optional): Name of constraint object. Defaults to "PeriodicBC".
    """

    def __init__(
        self,
        output_expr: Dict[str, Callable],
        label_dict: Dict[str, Union[float, Callable]],
        geom: geometry.Geometry,
        periodic_key: str,
        dataloader_cfg: Dict[str, Any],
        loss: "loss.Loss",
        random: Literal["pseudo", "Halton", "LHS"] = "pseudo",
        criteria: Optional[Callable] = None,
        evenly: bool = False,
        weight_dict: Optional[Dict[str, Callable]] = None,
        name: str = "PeriodicBC",
    ):
        self.input_keys = geom.dim_keys
        self.output_keys = tuple(output_expr.keys())
        self.output_expr = {
            k: v for k, v in output_expr.items() if k in self.output_keys
        }

        if isinstance(criteria, str):
            criteria = eval(criteria)

        if dataloader_cfg["batch_size"] % 2 > 0:
            raise ValueError(
                f"batch_size({dataloader_cfg['sampler']['batch_size']}) "
                "should be positive and even when using PeriodicConstraint"
            )
        if dataloader_cfg.get("shuffle", False):
            raise ValueError(
                f"shuffle({dataloader_cfg['sampler']['batch_size']}) "
                "should be False when using PeriodicConstraint"
            )

        # prepare input
        _bs_half = dataloader_cfg["batch_size"] // 2
        input = geom.sample_boundary(
            _bs_half * dataloader_cfg["iters_per_epoch"],
            random,
            criteria,
            evenly,
        )
        if "area" in input:
            input["area"] *= dataloader_cfg["iters_per_epoch"]

        input_periodic = geom.periodic_point(
            input,
            geom.geometry.dim_keys.index(periodic_key)
            if isinstance(geom, geometry.TimeXGeometry)
            else geom.dim_keys.index(periodic_key),
        )
        # concatenate original data next to periodic data, i.e.
        # [orignal1, periodic1, orignal2, periodic2, ..., orignalN, periodicN]
        mixed_input = {}
        for key in input:
            mixed_input[key] = []
            for iter_id in range(dataloader_cfg["iters_per_epoch"]):
                mixed_input[key].append(
                    input[key][iter_id * _bs_half : (iter_id + 1) * _bs_half]
                )
                mixed_input[key].append(
                    input_periodic[key][iter_id * _bs_half : (iter_id + 1) * _bs_half]
                )
            mixed_input[key] = np.vstack(mixed_input[key])

        # prepare label, keep label the same shape as input_periodic
        label = {}
        for key, value in label_dict.items():
            # set all label's to zero for dummy data.
            label[key] = np.full(
                (next(iter(mixed_input.values())).shape[0], 1),
                0,
                paddle.get_default_dtype(),
            )

        # # prepare weight, keep weight the same shape as input_periodic
        weight = None
        if weight_dict is not None:
            weight = {key: np.ones_like(next(iter(label.values()))) for key in label}
            for key, value in weight_dict.items():
                if isinstance(value, (int, float)):
                    weight[key] = np.full_like(next(iter(label.values())), value)
                elif isinstance(value, sympy.Basic):
                    func = sympy.lambdify(
                        [sympy.Symbol(k) for k in geom.dim_keys],
                        value,
                        [{"amax": lambda xy, _: np.maximum(xy[0], xy[1])}, "numpy"],
                    )
                    weight[key] = func(**{k: mixed_input[k] for k in geom.dim_keys})
                elif callable(value):
                    func = value
                    weight[key] = func(mixed_input)
                    if isinstance(weight[key], (int, float)):
                        weight[key] = np.full_like(
                            next(iter(mixed_input.values())), weight[key]
                        )
                else:
                    raise NotImplementedError(f"type of {type(value)} is invalid yet.")

        # wrap input, label, weight into a dataset
        if isinstance(dataloader_cfg["dataset"], str):
            dataloader_cfg["dataset"] = {"name": dataloader_cfg["dataset"]}
        dataloader_cfg["dataset"].update(
            {"input": mixed_input, "label": label, "weight": weight}
        )
        _dataset = dataset.build_dataset(dataloader_cfg["dataset"])

        # construct dataloader with dataset and dataloader_cfg
        super().__init__(_dataset, dataloader_cfg, loss, name)

SupervisedConstraint

Bases: Constraint

Class for supervised constraint.

Parameters:

Name	Type	Description	Default
dataloader_cfg	Dict[str, Any]	Dataloader config.	required
loss	Loss	Loss functor.	required
output_expr	Optional[Dict[str, Callable]]	List of label expression. Defaults to None.	None
name	str	Name of constraint object. Defaults to "Sup".	'Sup'

Examples:

>>> import ppsci
>>> bc_sup = ppsci.constraint.SupervisedConstraint(
...     {
...         "dataset": {
...             "name": "IterableCSVDataset",
...             "file_path": "/path/to/file.csv",
...             "input_keys": ("x", "y"),
...             "label_keys": ("u", "v"),
...         },
...     },
...     ppsci.loss.MSELoss("mean"),
...     name="bc_sup",
... )

Source code in ppsci/constraint/supervised_constraint.py

class SupervisedConstraint(base.Constraint):
    """Class for supervised constraint.

    Args:
        dataloader_cfg (Dict[str, Any]): Dataloader config.
        loss (loss.Loss): Loss functor.
        output_expr (Optional[Dict[str, Callable]]): List of label expression.
            Defaults to None.
        name (str, optional): Name of constraint object. Defaults to "Sup".

    Examples:
        >>> import ppsci
        >>> bc_sup = ppsci.constraint.SupervisedConstraint(
        ...     {
        ...         "dataset": {
        ...             "name": "IterableCSVDataset",
        ...             "file_path": "/path/to/file.csv",
        ...             "input_keys": ("x", "y"),
        ...             "label_keys": ("u", "v"),
        ...         },
        ...     },
        ...     ppsci.loss.MSELoss("mean"),
        ...     name="bc_sup",
        ... )  # doctest: +SKIP
    """

    def __init__(
        self,
        dataloader_cfg: Dict[str, Any],
        loss: "loss.Loss",
        output_expr: Optional[Dict[str, Callable]] = None,
        name: str = "Sup",
    ):
        # build dataset
        _dataset = dataset.build_dataset(dataloader_cfg["dataset"])

        self.input_keys = _dataset.input_keys
        self.output_keys = (
            tuple(output_expr.keys())
            if output_expr is not None
            else _dataset.label_keys
        )

        self.output_expr = output_expr
        if self.output_expr is None:
            self.output_expr = {
                key: (lambda out, k=key: out[k]) for key in self.output_keys
            }

        # construct dataloader with dataset and dataloader_cfg
        super().__init__(_dataset, dataloader_cfg, loss, name)

    def __str__(self):
        return ", ".join(
            [
                self.__class__.__name__,
                f"name = {self.name}",
                f"input_keys = {self.input_keys}",
                f"output_keys = {self.output_keys}",
                f"output_expr = {self.output_expr}",
                f"loss = {self.loss}",
            ]
        )