fix connection issue

Author: GiovanniCanali

pina/problem/zoo/inverse_poisson_2d_square.py

@@ -4,20 +4,19 @@
 import requests
 import torch
 from io import BytesIO
-from requests.exceptions import RequestException
 from ... import Condition
 from ... import LabelTensor
 from ...operator import laplacian
 from ...domain import CartesianDomain
 from ...equation import Equation, FixedValue
 from ...problem import SpatialProblem, InverseProblem
-from ...utils import custom_warning_format
+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):
+def _load_tensor_from_url(url, labels, timeout=10):
     """
     Downloads a tensor file from a URL and wraps it in a LabelTensor.
 
@@ -28,21 +27,24 @@ def _load_tensor_from_url(url, labels):
 
     :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)
+        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)
-    except RequestException as e:
-        print(
-            "Could not download data for 'InversePoisson2DSquareProblem' "
-            f"from '{url}'. "
-            f"Reason: {e}. Skipping data loading.",
+
+    # 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
@@ -66,19 +68,6 @@ def laplace_equation(input_, output_, params_):
     return delta_u - force_term
 
 
-# loading 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"
-)
-input_data = _load_tensor_from_url(input_url, ["x", "y", "mu1", "mu2"])
-output_data = _load_tensor_from_url(output_url, ["u"])
-
-
 class InversePoisson2DSquareProblem(SpatialProblem, InverseProblem):
     r"""
     Implementation of the inverse 2-dimensional Poisson problem in the square
@@ -113,5 +102,50 @@ class InversePoisson2DSquareProblem(SpatialProblem, InverseProblem):
         "D": Condition(domain="D", equation=Equation(laplace_equation)),
     }
 
-    if input_data is not None and input_data is not None:
-        conditions["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]
+                )

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)

tests/test_solver/test_competitive_pinn.py

@@ -20,15 +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
-if hasattr(inverse_problem.conditions, "data"):
-    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)

tests/test_solver/test_gradient_pinn.py

@@ -31,15 +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
-if hasattr(inverse_problem.conditions, "data"):
-    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)

tests/test_solver/test_pinn.py

@@ -20,15 +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
-if hasattr(inverse_problem.conditions, "data"):
-    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)

tests/test_solver/test_rba_pinn.py

@@ -19,15 +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
-if hasattr(inverse_problem.conditions, "data"):
-    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)

tests/test_solver/test_self_adaptive_pinn.py

@@ -20,15 +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
-if hasattr(inverse_problem.conditions, "data"):
-    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)