(self, operation: Operation)
| 322 | return downstream_ops |
| 323 | |
| 324 | def get_upstream_operations(self, operation: Operation) -> List[Operation]: |
| 325 | if not isinstance(operation, Operation): |
| 326 | raise TypeError(f'Expect an operation instance, however {type(operation)} is given.') |
| 327 | if operation.name not in self.operations: |
| 328 | raise KeyError(f'Operation {operation.name} not in current graph.') |
| 329 | upstream_ops = [] |
| 330 | for input_var in operation.inputs: |
| 331 | if input_var.source_op is not None: |
| 332 | upstream_ops.append(input_var.source_op) |
| 333 | return upstream_ops |
| 334 | |
| 335 | def topological_sort(self) -> List[Operation]: |
| 336 | visited = {operation.name: False for operation in self.operations.values()} |
no test coverage detected