diff --git a/pina/problem/zoo/inverse_poisson_2d_square.py b/pina/problem/zoo/inverse_poisson_2d_square.py index f112ebfc0..ab89c4c16 100644 --- a/pina/problem/zoo/inverse_poisson_2d_square.py +++ b/pina/problem/zoo/inverse_poisson_2d_square.py @@ -1,5 +1,6 @@ """Formulation of the inverse Poisson problem in a square domain.""" +import warnings import requests import torch from io import BytesIO @@ -9,6 +10,44 @@ from ...domain import CartesianDomain from ...equation import Equation, FixedValue from ...problem import SpatialProblem, InverseProblem +from ...utils import custom_warning_format, check_consistency + +warnings.formatwarning = custom_warning_format +warnings.filterwarnings("always", category=ResourceWarning) + + +def _load_tensor_from_url(url, labels, timeout=10): + """ + Downloads a tensor file from a URL and wraps it in a LabelTensor. + + This function fetches a `.pth` file containing tensor data, extracts it, + and returns it as a LabelTensor using the specified labels. If the file + cannot be retrieved (e.g., no internet connection), a warning is issued + and None is returned. + + :param str url: URL to the remote `.pth` tensor file. + :param list[str] | tuple[str] labels: Labels for the resulting LabelTensor. + :param int timeout: Timeout for the request in seconds. + :return: A LabelTensor object if successful, otherwise None. + :rtype: LabelTensor | None + """ + # Try to download the tensor file from the given URL + try: + response = requests.get(url, timeout=timeout) + response.raise_for_status() + tensor = torch.load( + BytesIO(response.content), weights_only=False + ).tensor.detach() + return LabelTensor(tensor, labels) + + # If the request fails, issue a warning and return None + except requests.exceptions.RequestException as e: + warnings.warn( + f"Could not download data for 'InversePoisson2DSquareProblem' " + f"from '{url}'. Reason: {e}. Skipping data loading.", + ResourceWarning, + ) + return None def laplace_equation(input_, output_, params_): @@ -29,35 +68,13 @@ def laplace_equation(input_, output_, params_): return delta_u - force_term -# URL of the file -url = "https://github.com/mathLab/PINA/raw/refs/heads/master/tutorials/tutorial7/data/pts_0.5_0.5" -# Download the file -response = requests.get(url) -response.raise_for_status() -file_like_object = BytesIO(response.content) -# Set the data -input_data = LabelTensor( - torch.load(file_like_object, weights_only=False).tensor.detach(), - ["x", "y", "mu1", "mu2"], -) - -# URL of the file -url = "https://github.com/mathLab/PINA/raw/refs/heads/master/tutorials/tutorial7/data/pinn_solution_0.5_0.5" -# Download the file -response = requests.get(url) -response.raise_for_status() -file_like_object = BytesIO(response.content) -# Set the data -output_data = LabelTensor( - torch.load(file_like_object, weights_only=False).tensor.detach(), ["u"] -) - - class InversePoisson2DSquareProblem(SpatialProblem, InverseProblem): r""" Implementation of the inverse 2-dimensional Poisson problem in the square domain :math:`[0, 1] \times [0, 1]`, with unknown parameter domain :math:`[-1, 1] \times [-1, 1]`. + The `"data"` condition is added only if the required files are + downloaded successfully. :Example: >>> problem = InversePoisson2DSquareProblem() @@ -83,5 +100,52 @@ class InversePoisson2DSquareProblem(SpatialProblem, InverseProblem): "g3": Condition(domain="g3", equation=FixedValue(0.0)), "g4": Condition(domain="g4", equation=FixedValue(0.0)), "D": Condition(domain="D", equation=Equation(laplace_equation)), - "data": Condition(input=input_data, target=output_data), } + + def __init__(self, load=True, data_size=1.0): + """ + Initialization of the :class:`InversePoisson2DSquareProblem`. + + :param bool load: If True, it attempts to load data from remote URLs. + Set to False to skip data loading (e.g., if no internet connection). + :param float data_size: The fraction of the total data to use for the + "data" condition. If set to 1.0, all available data is used. + If set to 0.0, no data is used. Default is 1.0. + :raises ValueError: If `data_size` is not in the range [0.0, 1.0]. + :raises ValueError: If `data_size` is not a float. + """ + super().__init__() + + # Check consistency + check_consistency(load, bool) + check_consistency(data_size, float) + if not 0.0 <= data_size <= 1.0: + raise ValueError( + f"data_size must be in the range [0.0, 1.0], got {data_size}." + ) + + # Load data if requested + if load: + + # Define URLs for input and output data + input_url = ( + "https://github.com/mathLab/PINA/raw/refs/heads/master" + "/tutorials/tutorial7/data/pts_0.5_0.5" + ) + output_url = ( + "https://github.com/mathLab/PINA/raw/refs/heads/master" + "/tutorials/tutorial7/data/pinn_solution_0.5_0.5" + ) + + # Define input and output data + input_data = _load_tensor_from_url( + input_url, ["x", "y", "mu1", "mu2"] + ) + output_data = _load_tensor_from_url(output_url, ["u"]) + + # Add the "data" condition + if input_data is not None and output_data is not None: + n_data = int(input_data.shape[0] * data_size) + self.conditions["data"] = Condition( + input=input_data[:n_data], target=output_data[:n_data] + ) diff --git a/tests/test_problem_zoo/test_inverse_poisson_2d_square.py b/tests/test_problem_zoo/test_inverse_poisson_2d_square.py index 20a60e636..4304c8a5e 100644 --- a/tests/test_problem_zoo/test_inverse_poisson_2d_square.py +++ b/tests/test_problem_zoo/test_inverse_poisson_2d_square.py @@ -1,12 +1,25 @@ from pina.problem.zoo import InversePoisson2DSquareProblem from pina.problem import InverseProblem, SpatialProblem +import pytest -def test_constructor(): - problem = InversePoisson2DSquareProblem() +@pytest.mark.parametrize("load", [True, False]) +@pytest.mark.parametrize("data_size", [0.01, 0.05]) +def test_constructor(load, data_size): + + # Define the problem with or without loading data + problem = InversePoisson2DSquareProblem(load=load, data_size=data_size) + + # Discretise the domain problem.discretise_domain(n=10, mode="random", domains="all") + + # Check if the problem is correctly set up assert problem.are_all_domains_discretised assert isinstance(problem, InverseProblem) assert isinstance(problem, SpatialProblem) assert hasattr(problem, "conditions") assert isinstance(problem.conditions, dict) + + # Should fail if data_size is not in the range [0.0, 1.0] + with pytest.raises(ValueError): + problem = InversePoisson2DSquareProblem(load=load, data_size=3.0) diff --git a/tests/test_solver/test_competitive_pinn.py b/tests/test_solver/test_competitive_pinn.py index 741390e31..8f585f029 100644 --- a/tests/test_solver/test_competitive_pinn.py +++ b/tests/test_solver/test_competitive_pinn.py @@ -20,14 +20,9 @@ # define problems problem = Poisson() problem.discretise_domain(10) -inverse_problem = InversePoisson() +inverse_problem = InversePoisson(load=True, data_size=0.01) inverse_problem.discretise_domain(10) -# reduce the number of data points to speed up testing -data_condition = inverse_problem.conditions["data"] -data_condition.input = data_condition.input[:10] -data_condition.target = data_condition.target[:10] - # add input-output condition to test supervised learning input_pts = torch.rand(10, len(problem.input_variables)) input_pts = LabelTensor(input_pts, problem.input_variables) diff --git a/tests/test_solver/test_gradient_pinn.py b/tests/test_solver/test_gradient_pinn.py index 6e6c76c65..c28fc347e 100644 --- a/tests/test_solver/test_gradient_pinn.py +++ b/tests/test_solver/test_gradient_pinn.py @@ -31,14 +31,9 @@ class DummyTimeProblem(TimeDependentProblem): # define problems problem = Poisson() problem.discretise_domain(10) -inverse_problem = InversePoisson() +inverse_problem = InversePoisson(load=True, data_size=0.01) inverse_problem.discretise_domain(10) -# reduce the number of data points to speed up testing -data_condition = inverse_problem.conditions["data"] -data_condition.input = data_condition.input[:10] -data_condition.target = data_condition.target[:10] - # add input-output condition to test supervised learning input_pts = torch.rand(10, len(problem.input_variables)) input_pts = LabelTensor(input_pts, problem.input_variables) diff --git a/tests/test_solver/test_pinn.py b/tests/test_solver/test_pinn.py index ee501d876..d726047ef 100644 --- a/tests/test_solver/test_pinn.py +++ b/tests/test_solver/test_pinn.py @@ -20,14 +20,9 @@ # define problems problem = Poisson() problem.discretise_domain(10) -inverse_problem = InversePoisson() +inverse_problem = InversePoisson(load=True, data_size=0.01) inverse_problem.discretise_domain(10) -# reduce the number of data points to speed up testing -data_condition = inverse_problem.conditions["data"] -data_condition.input = data_condition.input[:10] -data_condition.target = data_condition.target[:10] - # add input-output condition to test supervised learning input_pts = torch.rand(10, len(problem.input_variables)) input_pts = LabelTensor(input_pts, problem.input_variables) diff --git a/tests/test_solver/test_rba_pinn.py b/tests/test_solver/test_rba_pinn.py index 92ef5396a..b464f3a7c 100644 --- a/tests/test_solver/test_rba_pinn.py +++ b/tests/test_solver/test_rba_pinn.py @@ -19,14 +19,9 @@ # define problems problem = Poisson() problem.discretise_domain(10) -inverse_problem = InversePoisson() +inverse_problem = InversePoisson(load=True, data_size=0.01) inverse_problem.discretise_domain(10) -# reduce the number of data points to speed up testing -data_condition = inverse_problem.conditions["data"] -data_condition.input = data_condition.input[:10] -data_condition.target = data_condition.target[:10] - # add input-output condition to test supervised learning input_pts = torch.rand(10, len(problem.input_variables)) input_pts = LabelTensor(input_pts, problem.input_variables) diff --git a/tests/test_solver/test_self_adaptive_pinn.py b/tests/test_solver/test_self_adaptive_pinn.py index f7ef7b59f..b2d1361ca 100644 --- a/tests/test_solver/test_self_adaptive_pinn.py +++ b/tests/test_solver/test_self_adaptive_pinn.py @@ -20,14 +20,9 @@ # define problems problem = Poisson() problem.discretise_domain(10) -inverse_problem = InversePoisson() +inverse_problem = InversePoisson(load=True, data_size=0.01) inverse_problem.discretise_domain(10) -# reduce the number of data points to speed up testing -data_condition = inverse_problem.conditions["data"] -data_condition.input = data_condition.input[:10] -data_condition.target = data_condition.target[:10] - # add input-output condition to test supervised learning input_pts = torch.rand(10, len(problem.input_variables)) input_pts = LabelTensor(input_pts, problem.input_variables)