Skip to content

fix connection problem.zoo #577

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: dev
Choose a base branch
from
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
72 changes: 51 additions & 21 deletions pina/problem/zoo/inverse_poisson_2d_square.py
Original file line number Diff line number Diff line change
@@ -1,14 +1,51 @@
"""Formulation of the inverse Poisson problem in a square domain."""

import warnings
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

warnings.formatwarning = custom_warning_format
warnings.filterwarnings("always", category=ResourceWarning)


def _load_tensor_from_url(url, labels):
"""
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.
:return: A LabelTensor object if successful, otherwise None.
:rtype: LabelTensor | None
"""
try:
response = requests.get(url)
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.",
ResourceWarning,
)
return None


def laplace_equation(input_, output_, params_):
Expand All @@ -29,35 +66,26 @@ 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"],
# loading data
input_url = (
"https://github.com/mathLab/PINA/raw/refs/heads/master"
"/tutorials/tutorial7/data/pts_0.5_0.5"
)

# 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"]
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
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()
Expand All @@ -83,5 +111,7 @@ 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),
}

if input_data is not None and input_data is not None:
conditions["data"] = Condition(input=input_data, target=output_data)
7 changes: 4 additions & 3 deletions tests/test_solver/test_competitive_pinn.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,10 @@
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]
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))
Expand Down
7 changes: 4 additions & 3 deletions tests/test_solver/test_gradient_pinn.py
Original file line number Diff line number Diff line change
Expand Up @@ -35,9 +35,10 @@ class DummyTimeProblem(TimeDependentProblem):
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]
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))
Expand Down
7 changes: 4 additions & 3 deletions tests/test_solver/test_pinn.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,10 @@
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]
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))
Expand Down
7 changes: 4 additions & 3 deletions tests/test_solver/test_rba_pinn.py
Original file line number Diff line number Diff line change
Expand Up @@ -23,9 +23,10 @@
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]
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))
Expand Down
7 changes: 4 additions & 3 deletions tests/test_solver/test_self_adaptive_pinn.py
Original file line number Diff line number Diff line change
Expand Up @@ -24,9 +24,10 @@
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]
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))
Expand Down
Loading