refactor: add _post_extract function for walkers

Author: rememberYou

pyrdf2vec/walkers/halk.py

@@ -135,3 +135,43 @@ def _extract(self, kg: KG, entity: Vertex) -> EntityWalks:
                         )
                 canonical_walks.add(tuple(canonical_walk))
         return {entity.name: list(canonical_walks)}
+
+    def _post_extract(self, res: List[EntityWalks]) -> List[List[SWalk]]:
+        """Post processed walks.
+
+        Args:
+            res: the result of the walks extracted with multiprocessing.
+
+        Returns:
+            The 2D matrix with its number of rows equal to the number of
+            provided entities; number of column equal to the embedding size.
+
+        """
+        conv_res = list(
+            walks
+            for entity_to_walks in res
+            for walks in entity_to_walks.values()
+        )
+        walks: List[SWalk] = [
+            walk for entity_walks in conv_res for walk in entity_walks
+        ]
+
+        predicates_dict = self.build_dictionary(walks)
+        pred_thresholds = [
+            self.get_rare_predicates(predicates_dict, walks, freq_threshold)
+            for freq_threshold in self.freq_thresholds
+        ]
+        res_halk = []
+        for rare_predicates in pred_thresholds:
+            for entity_walks in conv_res:
+                canonical_walks = []
+                for walk in entity_walks:
+                    canonical_walk = [walk[0]]
+                    for i, vertex in enumerate(walk[1::2], 2):
+                        if vertex not in rare_predicates:
+                            obj = walk[i] if i % 2 == 0 else walk[i + 1]
+                            canonical_walk += [vertex, obj]
+                    if len(canonical_walk) > 1:
+                        canonical_walks.append(tuple(canonical_walk))
+                res_halk.append(canonical_walks)
+        return res_halk

pyrdf2vec/walkers/walker.py

@@ -158,7 +158,7 @@ def extract(
                     disable=True if verbose == 0 else False,
                 )
             )
-        return list(walks for elm in res for walks in elm.values())
+        return self._post_extract(res)
 
     @abstractmethod
     def _extract(self, kg: KG, entity: Vertex) -> EntityWalks:
@@ -189,6 +189,27 @@ def _init_worker(self, init_kg: KG) -> None:
         global kg
         kg = init_kg  # type: ignore
 
+    def _post_extract(self, res: List[EntityWalks]) -> List[List[SWalk]]:
+        """Post processed walks.
+
+        Args:
+            res: the result of the walks extracted with multiprocessing.
+
+        Returns:
+            The 2D matrix with its number of rows equal to the number of
+            provided entities; number of column equal to the embedding size.
+
+        Raises:
+            NotImplementedError: If this method is called, without having
+                provided an implementation.
+
+        """
+        return list(
+            walks
+            for entity_to_walks in res
+            for walks in entity_to_walks.values()
+        )
+
     def _proc(self, entity: str) -> EntityWalks:
         """Executed by each process.