(self)
| 58 | self.max_flow = self.compute_max_flow() |
| 59 | |
| 60 | def compute_max_flow(self) -> int: |
| 61 | max_flow = 0 |
| 62 | while True: |
| 63 | augmenting_path = self.find_augmenting_path() |
| 64 | if not augmenting_path: |
| 65 | break |
| 66 | bottleneck = min([edge.capacity - edge.flow for edge in augmenting_path]) |
| 67 | for edge in augmenting_path: |
| 68 | edge.flow += bottleneck |
| 69 | self.edges_dict[(edge.to_node, edge.from_node)].flow -= bottleneck |
| 70 | max_flow += bottleneck |
| 71 | return max_flow |
| 72 | |
| 73 | def find_augmenting_path(self) -> Optional[List[Edge]]: |
| 74 | # BFS |
no test coverage detected