Class SearchableGraph

ppq/IR/search.py:390–690

PPQ Graph Searching Engine. Args: GraphCommandProcessor ([type]): [description]

class SearchableGraph(GraphCommandProcessor):
    """PPQ Graph Searching Engine.

    Args:
        GraphCommandProcessor ([type]): [description]
    """
    def __init__(self, graph: Union[BaseGraph, Callable]) -> None:
        super().__init__(graph)
        self._cache = {}

    def process(self, command: GraphCommand) -> Any:
        if isinstance(command, TraversalCommand):
            if command.command_type == GraphCommandType.TRAVERSAL_PATTERN_MATCHING:
                return self.path_matching(
                    sp_expr=command._sp_expr,
                    rp_expr=command._rp_expr,
                    ep_expr=command._ep_expr,
                    direction=command._direction)
            else:
                return self.opset_matching(
                    sp_expr=command._sp_expr,
                    rp_expr=command._rp_expr,
                    ep_expr=command._ep_expr,
                    direction=command._direction)
        else:
            raise TypeError(
                'To execute a traversal-based pattern matching, a TraversalCommand is required here.\n'
                'Initialize your own TraversalCommand instance for invoking '
                'this function rather than using plain GraphCommand Please.')

    def _acceptable_command_types(self) -> List[GraphCommandType]:
        return [
            GraphCommandType.TRAVERSAL_PATTERN_MATCHING,
            GraphCommandType.TRAVERSAL_OPSET_MATCHING,
            GraphCommandType.CONCAT_MATCHING,
            GraphCommandType.ACTIVATION_MATCHING,
        ]

    def _path_matching(
        self,
        start_point: Operation,
        rp_expr: RelayPattern,
        ep_expr: PointPattern,
        direction: str = 'up'
    ) -> List[Path]:
        # memoization based searching.
        if start_point in self._cache: return self._cache[start_point]

        # find next operations with given direction
        if direction == 'up': following_ops = self.graph.get_upstream_operations(start_point)
        else: following_ops = self.graph.get_downstream_operations(start_point)

        ret_collection = []
        for op in following_ops:
            # if operation is a valid end point, add it to path and stop further searching.
            if ep_expr(op):
                ret_collection.append(Path(start_point).append(op))