跳转至

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 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
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', '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 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
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", "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",
        ... )
    """

    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", "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
        }
        # "area" will be kept in "output_dict" for computation.
        if isinstance(geom, geometry.Mesh):
            self.output_keys += ("area",)

        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 = {key: np.ones_like(next(iter(label.values()))) for key in label}
        if weight_dict is not None:
            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', '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 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
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", "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",
        ... )
    """

    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", "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
        }
        # "area" will be kept in "output_dict" for computation.
        if isinstance(geom.geometry, geometry.Mesh):
            self.output_keys += ("area",)

        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 = {key: np.ones_like(next(iter(label.values()))) for key in label}
        if weight_dict is not None:
            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', '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 
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
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", "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",
        ... )
    """

    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", "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
        }
        # "area" will be kept in "output_dict" for computation.
        if isinstance(geom, geometry.Mesh):
            self.output_keys += ("area",)

        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 = {key: np.ones_like(next(iter(label.values()))) for key in label}
        if weight_dict is not None:
            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', '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 
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
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", "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",
        ... )
    """

    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", "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
        }
        # "area" will be kept in "output_dict" for computation.
        if isinstance(geom, geometry.Mesh):
            self.output_keys += ("area",)

        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 = {key: np.ones_like(next(iter(label.values()))) for key in label}
        if weight_dict is not None:
            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.")

        if "sdf" in input:
            input.pop("sdf")

        # 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', '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 
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
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", "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", "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
        }
        # "area" will be kept in "output_dict" for computation.
        if isinstance(geom, geometry.Mesh):
            self.output_keys += ("area",)

        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 = {key: np.ones_like(next(iter(label.values()))) for key in label}
        if weight_dict is not None:
            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 
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
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}",
            ]
        )
最后更新: November 17, 2023
创建日期: November 6, 2023