(
self,
start_point: Operation,
rp_expr: RelayPattern,
ep_expr: PointPattern,
direction: str = 'up'
)
| 426 | ] |
| 427 | |
| 428 | def _path_matching( |
| 429 | self, |
| 430 | start_point: Operation, |
| 431 | rp_expr: RelayPattern, |
| 432 | ep_expr: PointPattern, |
| 433 | direction: str = 'up' |
| 434 | ) -> List[Path]: |
| 435 | # memoization based searching. |
| 436 | if start_point in self._cache: return self._cache[start_point] |
| 437 | |
| 438 | # find next operations with given direction |
| 439 | if direction == 'up': following_ops = self.graph.get_upstream_operations(start_point) |
| 440 | else: following_ops = self.graph.get_downstream_operations(start_point) |
| 441 | |
| 442 | ret_collection = [] |
| 443 | for op in following_ops: |
| 444 | # if operation is a valid end point, add it to path and stop further searching. |
| 445 | if ep_expr(op): |
| 446 | ret_collection.append(Path(start_point).append(op)) |
| 447 | |
| 448 | else: |
| 449 | # if operation is not a valid relay point, end searching here. |
| 450 | if not rp_expr(start_point, op): continue |
| 451 | |
| 452 | # searching following operations. |
| 453 | for path in self._path_matching(start_point=op, rp_expr=rp_expr, |
| 454 | ep_expr=ep_expr, direction=direction): |
| 455 | ret_collection.append(path.copy().append_left(start_point)) |
| 456 | |
| 457 | self._cache[start_point] = ret_collection |
| 458 | return ret_collection |
| 459 | |
| 460 | def _opset_matching( |
| 461 | self, |
no test coverage detected