add genetic node

golem/core/constants.py

@@ -1,6 +1,6 @@
 import numpy as np
 
-MAX_GRAPH_GEN_ATTEMPTS_PER_IND = 50
+MAX_GRAPH_GEN_ATTEMPTS_PER_IND = 20
 MAX_GRAPH_GEN_ATTEMPTS = 1000
 MAX_TUNING_METRIC_VALUE = np.inf
 MIN_TIME_FOR_TUNING_IN_SEC = 3

golem/core/optimisers/genetic/operators/node.py

@@ -0,0 +1,65 @@
+from dataclasses import dataclass, replace
+from queue import Queue
+from typing import Optional, List, Union, Any
+
+from golem.core.optimisers.genetic.operators.operator import Operator
+from golem.core.optimisers.graph import OptGraph
+
+
+@dataclass
+class GeneticOperatorTask:
+    stage: 'GeneticNode'
+
+    graphs: List[OptGraph]
+    operator_type: Optional[Any] = None
+
+    # parent data
+    parent_task: Optional['GeneticOperatorTask'] = None
+
+    fail: bool = False
+    fail_message: str = ''
+    left_tries: int = 1
+
+    def copy(self):
+        return replace(self)
+
+    def create_failed_task(self, exception: Exception):
+        failed_task = self.copy()
+        failed_task.fail = True
+        failed_task.fail_message = exception.__str__()
+        failed_task.left_tries -= 1
+        return failed_task
+
+    def create_successive_task(self, graphs: List[OptGraph], operator_type: Any):
+        task = self.copy()
+        task.graphs = graphs
+        task.operator_type = operator_type
+        task.parent_task = self
+        return task
+
+
+@dataclass(frozen=True)
+class GeneticNode:
+    name: str
+    operator: Operator
+    processed_queue: Queue
+    success_outputs: List['GeneticNode'] = None
+    fail_outputs: Optional[List['GeneticNode']] = None
+
+    def __call__(self, task: GeneticOperatorTask):
+        if task.left_tries > 0:
+            try:
+                *grouped_graphs, operator_type = self.operator(task.graphs, task.operator_type)
+                tasks = [task.create_successive_task(graphs, operator_type) for graphs in grouped_graphs]
+                next_nodes = self.success_outputs
+            except Exception as exception:
+                tasks = [task.create_failed_task(exception)]
+                next_nodes = self.fail_outputs
+
+            final_tasks = list()
+            for _task in tasks:
+                for _node in next_nodes:
+                    new_task = _task.copy()
+                    new_task.stage = _node
+                    final_tasks.append(new_task)
+            return final_tasks

golem/core/optimisers/genetic/operators/reproduction.py

@@ -408,7 +408,7 @@ def evaluation_stage(self, task: ReproducerWorkerTask) -> List[ReproducerWorkerT
             graph = task.final_graph
         individual = Individual(deepcopy(graph), metadata=self.mutation.requirements.static_individual_metadata)
         evaluated_individuals = self.evaluator([individual])
-        if evaluated_individuals and evaluated_individuals[0].fitness.valid:
+        if evaluated_individuals:# and evaluated_individuals[0].fitness.valid:
             task.fail = False
             if task.is_crossover:
                 task.crossover_fitness = evaluated_individuals[0].fitness