Improve parallel import robustness and include missing unit tests

Author: kshitij-maths

ezyrb/parallel/__init__.py

@@ -17,8 +17,12 @@
 
 from .reduction import Reduction
 from .pod import POD
-from .ae import AE
-from .ae_eddl import AE_EDDL
+try:
+    from .ae import AE
+    from .ae_eddl import AE_EDDL
+except ImportError:
+    AE = None
+    AE_EDDL = None
 from .approximation import Approximation
 from .rbf import RBF
 from .linear import Linear

ezyrb/parallel/pod.py

@@ -11,8 +11,17 @@
     from numpy.linalg import eigh
 import numpy as np
 
-from pycompss.api.task import task
-from pycompss.api.parameter import INOUT, IN
+try:
+    from pycompss.api.task import task
+    from pycompss.api.parameter import INOUT, IN
+except ImportError:
+    # Fallback: Define a 'do-nothing' decorator and dummy constants
+    def task(*args, **kwargs):
+        return lambda f: f
+    
+    INOUT = None
+    IN = None
+
 from .reduction import Reduction
 
 

ezyrb/parallel/reducedordermodel.py

@@ -4,7 +4,7 @@
 import copy
 import pickle
 import numpy as np
-from scipy.spatial.qhull import Delaunay
+from scipy.spatial import Delaunay
 from sklearn.model_selection import KFold
 from pycompss.api.api import compss_wait_on
 
@@ -60,30 +60,32 @@ def fit(self, *args, **kwargs):
         :param \*args: additional parameters to pass to the `fit` method.
         :param \**kwargs: additional parameters to pass to the `fit` method.
         """
-        self.reduction.fit(self.database.snapshots.T)
-        reduced_output = self.reduction.transform(
-            self.database.snapshots.T, self.scaler_red
-        )
+        self.reduction.fit(self.database.snapshots_matrix.T)
+        reduced_output = self.reduction.transform(self.database.snapshots_matrix.T).T
 
         self.approximation.fit(
-            self.database.parameters, reduced_output, *args, **kwargs
+            self.database.parameters_matrix, reduced_output, *args, **kwargs
         )
 
         return self
 
     def predict(self, mu):
-        """
-        Calculate predicted solution for given mu
+        r"""
+        Predict the solution for given parameters mu.
+        
+        This method distributes the evaluation tasks across the 
+        available computational nodes using the PyCOMPSs framework.
+
+        :param numpy.ndarray mu: The parameters $\mu \in \mathbb{R}^d$ to evaluate.
+        :return: The predicted snapshot $u(\mu)$.
         """
         mu = np.atleast_2d(mu)
-        if hasattr(self, "database") and self.database.scaler_parameters:
-            mu = self.database.scaler_parameters.transform(mu)
-
-        predicted_red_sol = self.approximation.predict(mu, self.scaler_red)
+        
+        predicted_red_sol = self.approximation.predict(mu)
 
         predicted_sol = self.reduction.inverse_transform(
-            predicted_red_sol, self.database
-        )
+            predicted_red_sol.T
+        ).T
 
         return predicted_sol
 
@@ -166,8 +168,8 @@ def kfold_cv_error(self, n_splits, *args, norm=np.linalg.norm, **kwargs):
             ).fit(*args, **kwargs)
 
             test = self.database[test_index]
-            predicted_test.append(rom.predict(test.parameters))
-            original_test.append(test.snapshots)
+            predicted_test.append(rom.predict(test.parameters_matrix))
+            original_test.append(test.snapshots_matrix)
 
         predicted_test = compss_wait_on(predicted_test)
         for j in range(len(predicted_test)):
@@ -216,8 +218,8 @@ def loo_error(self, *args, norm=np.linalg.norm, **kwargs):
                 copy.deepcopy(self.approximation),
             ).fit(*args, **kwargs)
 
-            predicted_test.append(rom.predict(test_db.parameters))
-            original_test.append(test_db.snapshots)
+            predicted_test.append(rom.predict(test_db.parameters_matrix))
+            original_test.append(test_db.snapshots_matrix)
 
         predicted_test = compss_wait_on(predicted_test)
         for j in range(len(predicted_test)):
@@ -246,7 +248,7 @@ def optimal_mu(self, error=None, k=1):
         if error is None:
             error = self.loo_error()
 
-        mu = self.database.parameters
+        mu = self.database.parameters_matrix
         tria = Delaunay(mu)
 
         error_on_simplex = np.array(

tests/test_parallel.py

@@ -0,0 +1,49 @@
+import unittest
+import numpy as np
+import warnings
+
+from ezyrb import Database, POD, RBF
+from ezyrb.parallel import ReducedOrderModel
+
+class TestParallelROM(unittest.TestCase):
+    def setUp(self):
+        self.params = np.array([[1.0], [2.0], [3.0]])
+        self.snapshots = np.array([[10.0], [20.0], [30.0]])
+        self.db = Database(self.params, self.snapshots)
+
+    def test_initialization(self):
+        rom = ReducedOrderModel(self.db, POD(), RBF())
+        self.assertEqual(rom.database.parameters_matrix.shape, (3, 1))
+        self.assertIsNotNone(rom.reduction)
+        self.assertIsNotNone(rom.approximation)
+
+    def test_fit(self):
+        rom = ReducedOrderModel(self.db, POD(), RBF())
+        rom.fit()
+        self.assertTrue(hasattr(rom.reduction, 'modes'))
+
+    def test_predict_scalar(self):
+        rom = ReducedOrderModel(self.db, POD(), RBF())
+        rom.fit()
+        pred = rom.predict([1.5])
+        self.assertTrue(pred is not None)
+
+    def test_predict_db(self):
+        rom = ReducedOrderModel(self.db, POD(), RBF())
+        rom.fit()
+        pred = rom.predict(self.params)
+        
+        if hasattr(pred, 'snapshots_matrix'):
+            np.testing.assert_allclose(pred.snapshots_matrix, self.snapshots, rtol=1e-5)
+        else:
+            np.testing.assert_allclose(pred, self.snapshots, rtol=1e-5)
+
+    def test_wrong_dimensions(self):
+        with self.assertRaises(Exception):
+            bad_params = np.array([[1.0], [2.0]]) # Only 2 params
+            bad_db = Database(bad_params, self.snapshots) # 3 snapshots
+            rom = ReducedOrderModel(bad_db, POD(), RBF())
+            rom.fit()
+
+if __name__ == '__main__':
+    unittest.main()