FIX: Distributed Dataset

ai/src/itwinai/backend/executors.py

@@ -42,28 +42,27 @@ def __init__(self, pipeline, class_dict, param_space):
         self.param_space = param_space
 
         # Init ray
-        ray.init()
+        ray.init(ignore_reinit_error=True, num_gpus=2)
+        print('Ray is initialized')
 
     def worker_fn(self, config, pipeline, class_dict):
-        print('Worker fn')
         # Should have same structure pipe and params
-        # def replace(pipe, params):
-        #     for param in params:
-        #         if type(pipe[param]) != dict:
-        #             pipe[param] = params[param]
-        #         else:
-        #             replace(pipe[param], params[param])
-        #     return pipe
-        #
-        # template = pipeline
-        # with open(template, 'r') as f:
-        #     doc = yaml.safe_load(f)
-        # doc = replace(doc, config)
-        #
-        # executor = LocalExecutor(doc, class_dict)
-        # executor.setup(None)
-        # executor.execute(None)
-        return {"test": 1}
+        def replace(pipe, params):
+            for param in params:
+                if type(pipe[param]) != dict:
+                    pipe[param] = params[param]
+                else:
+                    replace(pipe[param], params[param])
+            return pipe
+
+        template = pipeline
+        with open(template, 'r') as f:
+            doc = yaml.safe_load(f)
+        doc = replace(doc, config)
+
+        executor = LocalExecutor(doc, class_dict)
+        executor.setup(None)
+        executor.execute(None)
 
     def execute(self, args):
         print('Execute')