diff --git a/Tests/Data/Parabolic/T/1D_neumann/plotLine.py b/Tests/Data/Parabolic/T/1D_neumann/plotLine.py
index 5751e663d7c339aa381c035d86260b587dd2eb31..e489a69dba79dd873e766c88c2cb7666333949c9 100755
--- a/Tests/Data/Parabolic/T/1D_neumann/plotLine.py
+++ b/Tests/Data/Parabolic/T/1D_neumann/plotLine.py
@@ -5,20 +5,20 @@ import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import temperature_analytical
-from vtk import *
+import vtk
from vtk.numpy_interface import dataset_adapter as dsa
from vtk.util.numpy_support import vtk_to_numpy
def probeFileAlongLine(filename):
- reader = vtkXMLUnstructuredGridReader()
+ reader = vtk.vtkXMLUnstructuredGridReader()
reader.SetFileName(filename)
- line = vtkLineSource()
+ line = vtk.vtkLineSource()
line.SetResolution(10000)
line.SetPoint1(0, 0, 0)
line.SetPoint2(60, 0, 0)
- line_probe = vtkProbeFilter()
+ line_probe = vtk.vtkProbeFilter()
line_probe.SetInputConnection(line.GetOutputPort())
line_probe.SetSourceConnection(reader.GetOutputPort())
line_probe.Update()
@@ -33,7 +33,7 @@ def convertPointDataToPandasDF(line_data):
data[k] = vtk_to_numpy(v)
coords = line_data.GetPoints()[:, 0] # x-coordinates
- df = pd.concat(
+ df = pd.concat( # noqa: PD901
[pd.DataFrame(v) for k, v in data.items()], axis=1, keys=list(data.keys())
)
df.index = coords
@@ -267,7 +267,7 @@ def singleTimeStep(ts):
data = []
for c in cases:
line = probeFileAlongLine(c + "_ts_" + str(ts) + "_t_" + str(t) + ".000000.vtu")
- df = convertPointDataToPandasDF(line)
+ df = convertPointDataToPandasDF(line) # noqa: PD901
df["analytical"] = temperature_analytical.temperature(df.index, t)
df["error"] = df["temperature"].sub(df["analytical"], axis=0)
print(
@@ -280,7 +280,7 @@ def singleTimeStep(ts):
)
data.append(df[["temperature", "error", "analytical"]])
- df = pd.concat(data, keys=cases, names=["case"], axis=1)
+ df = pd.concat(data, keys=cases, names=["case"], axis=1) # noqa: PD901
plotCases(
df[df[("picard_masslumping", "error")].abs() > 1e-6],
"temperature_error_ts_" + str(ts) + "_t_" + str(t),
diff --git a/Tests/Data/Parabolic/T/1D_neumann/temperature_analytical.py b/Tests/Data/Parabolic/T/1D_neumann/temperature_analytical.py
index aba814e795001e90b15c57cf78ccf3d6f2a3efab..34669ad4d3512bd7cdd8227b40c06e9deafbab1d 100755
--- a/Tests/Data/Parabolic/T/1D_neumann/temperature_analytical.py
+++ b/Tests/Data/Parabolic/T/1D_neumann/temperature_analytical.py
@@ -2,10 +2,10 @@
# Solution of heatequation in a semi-infinite domain.
import numpy as np
+import vtk
from scipy.special import erfc
-from vtk import *
-r = vtkXMLUnstructuredGridReader()
+r = vtk.vtkXMLUnstructuredGridReader()
r.SetFileName("mesh.vtu")
r.Update()
m = r.GetOutput()
@@ -30,7 +30,7 @@ def temperature(x, t):
def addSolution(t):
- T = vtkDoubleArray()
+ T = vtk.vtkDoubleArray()
T.SetName("temperature_" + str(t) + "s")
T.SetNumberOfComponents(1)
T.SetNumberOfTuples(ps.GetNumberOfPoints())
@@ -45,7 +45,7 @@ def addSolution(t):
for t in 78125 * np.array([1, 3, 65, 405, 500]):
addSolution(t)
-w = vtkXMLUnstructuredGridWriter()
+w = vtk.vtkXMLUnstructuredGridWriter()
w.SetFileName("x.vtu")
w.SetInputData(m)
w.Update()
diff --git a/Tests/Data/Parabolic/T/3D_3BHEs_array_python_interface/bcs_tespy_and_serverCommunication.py b/Tests/Data/Parabolic/T/3D_3BHEs_array_python_interface/bcs_tespy_and_serverCommunication.py
index d0ac5c8a57b47050148e1dcf48c4a41b21ab1811..eda9741dc4e1d8704a60670c7ee45e18ae8260cd 100644
--- a/Tests/Data/Parabolic/T/3D_3BHEs_array_python_interface/bcs_tespy_and_serverCommunication.py
+++ b/Tests/Data/Parabolic/T/3D_3BHEs_array_python_interface/bcs_tespy_and_serverCommunication.py
@@ -5,19 +5,21 @@
# http://www.opengeosys.org/project/license
###
-import sys
-
-print(sys.version)
import os
+import sys
+from pathlib import Path
import numpy as np
+from pandas import read_csv
+from tespy.networks import load_network
try:
import ogs.callbacks as OpenGeoSys
except ModuleNotFoundError:
import OpenGeoSys
-from pandas import read_csv
-from tespy.networks import load_network
+
+
+print(sys.version)
# User setting ++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++++
# parameters
@@ -148,56 +150,48 @@ class BC(OpenGeoSys.BHENetwork):
df.loc[:, "flowrate"] = 0
cur_flowrate = df["flowrate"].tolist()
return (True, True, Tout_val, cur_flowrate)
- else:
- # read Tout_val to dataframe
- for i in range(n_BHE):
- df.loc[df.index[i], "Tout_val"] = Tout_val[i]
- # TESPy solver
- cur_Tin_val, cur_flowrate = get_tespy_results(t)
- # check norm if network achieves the converge
- if_success = False
- pre_Tin_val = Tin_val
- norm = np.linalg.norm(
- abs(np.asarray(pre_Tin_val) - np.asarray(cur_Tin_val))
- )
- if norm < 10e-6:
- if_success = True
- # return to OGS
- return (True, if_success, cur_Tin_val, cur_flowrate)
-
- def serverCommunicationPreTimestep(self, t, dt, Tin_val, Tout_val, flowrate):
+
+ # read Tout_val to dataframe
+ for i in range(n_BHE):
+ df.loc[df.index[i], "Tout_val"] = Tout_val[i]
+ # TESPy solver
+ cur_Tin_val, cur_flowrate = get_tespy_results(t)
+ # check norm if network achieves the converge
+ if_success = False
+ pre_Tin_val = Tin_val
+ norm = np.linalg.norm(abs(np.asarray(pre_Tin_val) - np.asarray(cur_Tin_val)))
+ if norm < 10e-6:
+ if_success = True
+ # return to OGS
+ return (True, if_success, cur_Tin_val, cur_flowrate)
+
+ def serverCommunicationPreTimestep(self, t, _dt, Tin_val, Tout_val, flowrate):
# TODO: Code for SimualtionX simulation
# with t; only take the last results for each time point
# TODO: say SimulationX the next time point from OGS
- tin_file = open("T_in.txt", "a")
- tin_file.write(str(t) + str(Tin_val) + "\n")
- tin_file.close()
+ with Path("T_in.txt").open(mode="a") as fd:
+ fd.write(str(t) + str(Tin_val) + "\n")
- tout_file = open("T_out.txt", "a")
- tout_file.write(str(t) + str(Tout_val) + "\n")
- tout_file.close()
+ with Path("T_out.txt").open(mode="a") as fd:
+ fd.write(str(t) + str(Tout_val) + "\n")
- flowrate_file = open("flowrate.txt", "a")
- flowrate_file.write(str(t) + str(flowrate) + "\n")
- flowrate_file.close()
+ with Path("flowrate.txt").open(mode="a") as fd:
+ fd.write(str(t) + str(flowrate) + "\n")
return (Tin_val, flowrate)
- def serverCommunicationPostTimestep(self, t, dt, Tin_val, Tout_val, flowrate):
+ def serverCommunicationPostTimestep(self, t, _dt, Tin_val, Tout_val, flowrate):
Tin_val = [305, 305, 305]
- tin = open("T_in.txt", "a")
- tin.write("post: " + str(t) + str(Tin_val) + "\n")
- tin.close()
+ with Path("T_in.txt").open(mode="a") as fd:
+ fd.write("post: " + str(t) + str(Tin_val) + "\n")
- tout = open("T_out.txt", "a")
- tout.write("post: " + str(t) + str(Tout_val) + "\n")
- tout.close()
+ with Path("T_out.txt").open(mode="a") as fd:
+ fd.write("post: " + str(t) + str(Tout_val) + "\n")
- flowrate_file = open("flowrate.txt", "a")
- flowrate_file.write("post: " + str(t) + str(flowrate) + "\n")
- flowrate_file.close()
+ with Path("flowrate.txt").open(mode="a") as fd:
+ fd.write("post: " + str(t) + str(flowrate) + "\n")
return
@@ -206,14 +200,14 @@ class BC(OpenGeoSys.BHENetwork):
# initialize the tespy model of the bhe network
# load path of network model:
# loading the TESPy model
-if ogs_prj_directory != "":
- os.chdir(ogs_prj_directory)
+if ogs_prj_directory != "": # noqa: F821
+ os.chdir(ogs_prj_directory) # noqa: F821
nw = load_network("./pre/tespy_nw")
# set if print the network iteration info
nw.set_attr(iterinfo=False)
# create bhe dataframe of the network system from bhe_network.csv
-df = create_dataframe()
+df = create_dataframe() # noqa: PD901
n_BHE = np.size(df.iloc[:, 0])
# create local variables of the components label and connections label in