- Modélisation d'un réseau de flux
Un problème de flux en réseau modélise le transport d'une quantité à travers un graphe orienté dont chaque arc possède une capacité maximale. Les applicatoins couvrent la logistique, les télécommunications, la distribution électrique ou encore la planification de tâches.
Formellement, on dispose d'un graphe orienté G=(V,E). Chaque arc (i,j) est muni d'une capacité c(i,j) et d'un flux f(i,j) vérifiant :
0 ≤ f(i,j) ≤ c(i,j);- Conservation du flux en tout sommet intermédiaire : somme des entrants = somme des sortants ;
- La source génère le flux et le puits l'absorbe.
On distingue la capacité (capacité maximale d'un arc) du flux (utilisation réelle). La source/puits est définie par la structure du graphe, tandis que les nœuds d'offre/demande portent des valeurs fixes de production ou de consommation.
- Flot maximum
Le flot maximum consiste à envoyer le plus grand volume possible de la source vers le puits sans violer les capacités. Les méthodes principales sont les chemins augmentants (Edmonds-Karp, Dinitz) et les algorithmes de préflot (HLPP).
NetworkX offre plusieurs fonctions : maximum_flow, maximum_flow_value, minimum_cut, ainsi que des implémentations spécifiques (edmonds_karp, dinitz, preflow_push, shortest_augmenting_path, boykov_kolmogorov).
Exemple : réseau de canalisations pétrolières.
import networkx as nx
import matplotlib.pyplot as plt
g = nx.DiGraph()
aretes = [
('src','A',6),('src','C',8),('A','B',3),('A','D',3),('B','dest',10),
('C','D',4),('C','F',4),('D','E',3),('D','G',6),('E','B',7),
('E','J',4),('F','H',4),('G','E',7),('H','G',1),('H','I',3),
('I','J',3),('J','dest',5)
]
for u, v, cap in aretes:
g.add_edge(u, v, capacity=cap)
from networkx.algorithms.flow import shortest_augmenting_path
valeur, flot = nx.maximum_flow(g, 'src', 'dest', flow_func=shortest_augmenting_path)
capacites = nx.get_edge_attributes(g, 'capacity')
etiquettes, actifs = {}, []
for u, v in g.edges():
f = flot.get(u, {}).get(v, 0)
etiquettes[(u, v)] = f'c={capacites[(u,v)]}, f={f}'
if f > 0:
actifs.append((u, v))
print("Flot maximum :", valeur)
print("Répartition :", flot)
pos = {
'src': (1,8), 'A':(6,7.5), 'B':(9,8), 'C':(1.5,6), 'D':(4,6),
'E':(8,5.5), 'F':(2,4), 'G':(5,4), 'H':(1,2), 'I':(5.5,2.5),
'J':(9.5,2), 'dest':(11,6)
}
nx.draw(g, pos, with_labels=True, node_color='lightblue', node_size=400)
nx.draw_networkx_edge_labels(g, pos, etiquettes, font_color='navy')
nx.draw_networkx_edges(g, pos, edgelist=actifs, edge_color='crimson', width=2)
plt.show()
Le dictionnaire flot donne, pour chaque arc, le volume transporté. Les arcs colorés forment le soutien du flot optimal.
- Flot de coût minimum
Cette variante associe un coût unitaire à chaque arc et cherche à acheminer une demande fixée au moindre coût. NetworkX utilise le simplexe réseau via min_cost_flow, min_cost_flow_cost ou network_simplex. Les nœuds portent un attribut demand : négatif pour l'offre, positif pour la demande.
Le programme ci-dessous explore les coûts pour des demandes croissantes jusqu'à rupture de capacité.
import networkx as nx
graphe = nx.DiGraph()
graphe.add_edges_from([
('s','n1',{'capacity':7,'weight':4}),
('s','n2',{'capacity':8,'weight':4}),
('n1','n3',{'capacity':9,'weight':1}),
('n2','n1',{'capacity':5,'weight':5}),
('n2','n4',{'capacity':9,'weight':4}),
('n3','n4',{'capacity':6,'weight':2}),
('n3','t',{'capacity':10,'weight':6}),
('n4','n1',{'capacity':2,'weight':1}),
('n4','t',{'capacity':5,'weight':2})
])
dernier_flot, dernier_cout, dernier_dict = None, None, None
for q in range(1, 25):
graphe.nodes['s']['demand'] = -q
graphe.nodes['t']['demand'] = q
try:
cout = nx.min_cost_flow_cost(graphe)
repartition = nx.min_cost_flow(graphe)
print(f"Demande {q:2d} -> coût {cout}")
dernier_flot, dernier_cout, dernier_dict = q, cout, repartition
except nx.NetworkXUnfeasible:
print(f"La demande {q} dépasse la capacité du réseau.")
break
print("Dernière demande réalisable :", dernier_flot)
print("Coût associé :", dernier_cout)
print("Répartition finale :", dernier_dict)
À faible demande, le résultat coïncide avec le plus court chemin ; à la limite de capacité, il coïncide avec le flot maximum de coût minimum.
- Flot maximum de coût minimum
Lorsqu'on veut simultanément maximiser le flux et minimiser le coût, on utilise max_flow_min_cost. La fonction renvoie un dictionnaire de flux ; cost_of_flow calcule le coût total.
import networkx as nx
g = nx.DiGraph()
g.add_edges_from([
('s','v1',{'capacity':13,'weight':7}),
('s','v2',{'capacity':9,'weight':9}),
('v1','v3',{'capacity':6,'weight':6}),
('v1','v4',{'capacity':5,'weight':5}),
('v2','v1',{'capacity':4,'weight':4}),
('v2','v3',{'capacity':5,'weight':2}),
('v2','v5',{'capacity':5,'weight':5}),
('v3','v4',{'capacity':5,'weight':2}),
('v3','v5',{'capacity':4,'weight':1}),
('v3','t',{'capacity':4,'weight':4}),
('v4','t',{'capacity':9,'weight':7}),
('v5','t',{'capacity':9,'weight':5})
])
repartition = nx.max_flow_min_cost(g, 's', 't')
cout_total = nx.cost_of_flow(g, repartition)
volume = sum(repartition['s'][j] for j in repartition['s'])
print("Volume maximum :", volume)
print("Coût minimum correspondant :", cout_total)
print("Arcs utilisés :", [(i,j) for i in repartition for j in repartition[i] if repartition[i][j] > 0])
- Extensions courantes
5.1 Flux de valeur imposée
Pour obtenir un flot de valeur exacte v, on fixe les attributs demand de la source et du puits, puis on applique min_cost_flow. Cette approche fournit non seulement un flot réalisable mais aussi le flot de moindre coût pour cette valeur.
5.2 Multi-source et multi-puits
Deux approches existent. La première ajoute une super-source et un super-puits reliés aux nœuds d'offre et de demande par des arcs de capacité infinie. La seconde utilise directement l'attribut demand sur chaque nœud, ce que NetworkX exploite dans min_cost_flow.
5.3 Capacités sur les nœuds
On ramène une contrainte de capacité sur un nœud N à une contrainte d'arc en divisant N en N_in et N_out. Les arcs entrants arrivent à N_in, les arcs sortants partent de N_out, et un nouvel arc (N_in, N_out) porte la capacité nodale.
- Application : logistique multi-source multi-puits
Deux usines F1, F2 approvisionnent quatre détaillants R1..R4 via deux entrepôts W1, W2. Les demandes et capacités sont données. Le modèle NetworkX suivant minimise le coût de transport.
import networkx as nx
g_log = nx.DiGraph()
g_log.add_edges_from([
('F1','W1',{'capacity':9e5,'weight':2}),
('F1','W2',{'capacity':9e5,'weight':3}),
('F2','W1',{'capacity':9e5,'weight':3}),
('F2','W2',{'capacity':9e5,'weight':1}),
('W1','R1',{'capacity':9e5,'weight':2}),
('W1','R2',{'capacity':9e5,'weight':6}),
('W1','R3',{'capacity':9e5,'weight':3}),
('W1','R4',{'capacity':9e5,'weight':6}),
('W2','R1',{'capacity':9e5,'weight':4}),
('W2','R2',{'capacity':9e5,'weight':4}),
('W2','R3',{'capacity':9e5,'weight':6}),
('W2','R4',{'capacity':9e5,'weight':5})
])
demandes = {'F1':-600,'F2':-400,'R1':200,'R2':150,'R3':350,'R4':300}
for noeud, val in demandes.items():
g_log.nodes[noeud]['demand'] = val
cout_min = nx.min_cost_flow_cost(g_log)
flot_min = nx.min_cost_flow(g_log)
print("Coût minimum logistique :", cout_min)
print("Flux par arc :", flot_min)
Le même problème peut être résolu par programmation linéaire avec PuLP. Définissons les variables de flux pour chaque liaison et les contraintes de conservation.
import pulp
prob = pulp.LpProblem("LogistiqueMultiSource", pulp.LpMinimize)
routes = [('F1','W1'),('F1','W2'),('F2','W1'),('F2','W2'),
('W1','R1'),('W1','R2'),('W1','R3'),('W1','R4'),
('W2','R1'),('W2','R2'),('W2','R3'),('W2','R4')]
couts = {('F1','W1'):2,('F1','W2'):3,('F2','W1'):3,('F2','W2'):1,
('W1','R1'):2,('W1','R2'):6,('W1','R3'):3,('W1','R4'):6,
('W2','R1'):4,('W2','R2'):4,('W2','R3'):6,('W2','R4'):5}
x = pulp.LpVariable.dicts("x", routes, lowBound=0, cat='Continuous')
prob += pulp.lpSum(couts[r] * x[r] for r in routes)
prob += x[('F1','W1')] + x[('F1','W2')] <= 600
prob += x[('F2','W1')] + x[('F2','W2')] <= 400
prob += x[('W1','R1')] + x[('W2','R1')] == 200
prob += x[('W1','R2')] + x[('W2','R2')] == 150
prob += x[('W1','R3')] + x[('W2','R3')] == 350
prob += x[('W1','R4')] + x[('W2','R4')] == 300
prob += x[('F1','W1')] + x[('F2','W1')] == x[('W1','R1')]+x[('W1','R2')]+x[('W1','R3')]+x[('W1','R4')]
prob += x[('F1','W2')] + x[('F2','W2')] == x[('W2','R1')]+x[('W2','R2')]+x[('W2','R3')]+x[('W2','R4')]
prob.solve()
print("Coût optimal :", pulp.value(prob.objective))
for r in routes:
print(r, x[r].varValue)
- Multi-flot avec PuLP
Le multi-flot modélise plusieurs marchandises partageant les mêmes arcs. NetworkX ne fournit pas de fonction dédiée ; PuLP permet de déclarer des variables par marchandise et par arc, d'imposer la conservation du flux pour chaque produit et de respecter les capacités partagées.
import networkx as nx
import pulp
g = nx.DiGraph()
g.add_edges_from([
('O','A',{'capacity':7,'weight':4}),
('O','B',{'capacity':8,'weight':4}),
('A','C',{'capacity':9,'weight':1}),
('B','A',{'capacity':5,'weight':5}),
('B','D',{'capacity':9,'weight':4}),
('C','D',{'capacity':6,'weight':2}),
('C','E',{'capacity':10,'weight':6}),
('C','F',{'capacity':10,'weight':6}),
('D','A',{'capacity':2,'weight':1}),
('D','E',{'capacity':5,'weight':2}),
('D','F',{'capacity':5,'weight':2})
])
poids = nx.get_edge_attributes(g, 'weight')
capacites = nx.get_edge_attributes(g, 'capacity')
cap_max = max(capacites.values())
qA = pulp.LpVariable.dicts("produitA", g.edges(), lowBound=0, upBound=cap_max, cat='Continuous')
qB = pulp.LpVariable.dicts("produitB", g.edges(), lowBound=0, upBound=cap_max, cat='Continuous')
prob = pulp.LpProblem("MultiFlot", pulp.LpMinimize)
prob += pulp.lpSum(poids[e] * (qA[e] + qB[e]) for e in g.edges())
for e in g.edges():
prob += qA[e] + qB[e] <= capacites[e]
offreA, offreB = 6, 5
for n in g.nodes():
entrantA = sum(qA[e] for e in g.in_edges(nbunch=n))
sortantA = sum(qA[e] for e in g.out_edges(nbunch=n))
entrantB = sum(qB[e] for e in g.in_edges(nbunch=n))
sortantB = sum(qB[e] for e in g.out_edges(nbunch=n))
if g.in_degree(n) == 0:
prob += sortantA == offreA
prob += sortantB == offreB
elif g.out_degree(n) == 0:
if n == 'F':
prob += entrantA == offreA
if n == 'E':
prob += entrantB == offreB
else:
prob += entrantA == sortantA
prob += entrantB == sortantB
prob.solve()
print("Coût total :", pulp.value(prob.objective))
for v in prob.variables():
if v.varValue > 0:
print(v.name, "=", v.varValue)
Ce modèle s'adapte à des coûts différenciés par produit et à des contraintes supplémentaires telles que des flots entiers ou des temps de transit.