Informations pratiques

«  » »
Module pour résoudre le problème de placement de nœuds dans un graphe multi-graphe.

Ce module utilise le solveur CP-SAT d’OR-Tools pour placer des nœuds de différents types
(externaux, splices, relais) sur une grille, en respectant des contraintes de positionnement
et de routage des arêtes.
«  » »

import networkx as nx
from networkx.classes import number_of_edges
from ortools.sat.python import cp_model
from collections import defaultdict

# Nettoyage et fusion des routes
def clean_and_merge_route(route):
if not route or len(route) <= 2: return route
dedup = [route[0]]
for p in route[1:]:
if p != dedup[-1]: dedup.append(p)
if len(dedup) <= 2: return dedup
clean = [dedup[0]]
for i in range(1, len(dedup) – 1):
prev = clean[-1]
curr = dedup[i]
nxt = dedup[i + 1]
if (prev[1] == curr[1] == nxt[1]) or (prev[0] == curr[0] == nxt[0]): continue
clean.append(curr)
clean.append(dedup[-1])
return clean

def max_equipment_edges(G):
«  » »
Retourne le nombre maximum d’arêtes reliées à un même équipement dans le graphe.

Args:
G: Un objet MultiGraph NetworkX

Returns:
tuple: (nom_equipement, nombre_arêtes)
«  » »
equipment_counts = defaultdict(int)

for u, v, k, data in G.edges(keys=True, data=True):
equip1 = G.nodes[u].get(‘equipment’, ‘NEQ’)
equip2 = G.nodes[v].get(‘equipment’, ‘NEQ’)
equipment_counts[equip1] += 1
equipment_counts[equip2] += 1

equipment_counts[‘NEQ’] = 0
if equipment_counts:
max_equipment = max(equipment_counts.items(), key=lambda x: x[1])
return max_equipment[0], max_equipment[1] + 30
else:
return None, 0

def solve_nx_placement(G: nx.MultiGraph, backbone_orientation: str = ‘vertical’, allow_bent_leaves: bool = False) -> nx.MultiGraph:
«  » »
Résout le problème de placement des nœuds dans un graphe multi-graphe.

Args:
G: Le graphe multi-graphe à placer
backbone_orientation: Orientation des backbones (‘vertical’ ou ‘horizontal’)
allow_bent_leaves: On permet ou non les virages pour les arêtes
feuilles (qui relient au moins un noeud avec un unique voisin)

Returns:
Le graphe avec les nœuds placés et les routes calculées, ou un graphe vide si la résolution échoue
«  » »

model = cp_model.CpModel()

_, grid_size = max_equipment_edges(G)

# Clause de garde architecturale : stopper si la grille est nulle (graphe vide)
if grid_size == 0:
return G

print(grid_size)

node_vars = {}
equipment_groups = {}
relays_of_node = {n: [] for n in G.nodes()}

halo_x_ints, halo_y_ints = [], []
routing_h_x_ints, routing_h_y_ints = [], []
routing_v_x_ints, routing_v_y_ints = [], []
collision_blocks = []

# ==========================================
# 1. DÉCLARATION DES NŒUDS & SPLICES
# ==========================================

for n in G.nodes():
data = G.nodes[n]

if data.get(« type ») == « external »:
x = model.NewIntVar(0, grid_size, f »x_{n} »)
y = model.NewIntVar(0, grid_size, f »y_{n} »)

b_top = model.NewBoolVar(f »ext_top_{n} »)
b_bot = model.NewBoolVar(f »ext_bot_{n} »)
b_left = model.NewBoolVar(f »ext_left_{n} »)
b_right = model.NewBoolVar(f »ext_right_{n} »)

model.Add(y == 0).OnlyEnforceIf(b_top)
model.Add(y == grid_size).OnlyEnforceIf(b_bot)
model.Add(x == 0).OnlyEnforceIf(b_left)
model.Add(x == grid_size).OnlyEnforceIf(b_right)

model.AddBoolOr([b_top, b_bot, b_left, b_right])

node_vars[n] = {
« is_external »: True,
« is_splice »: False,
« is_relay »: False,
« x »: x,
« y »: y
}
continue

if data.get(« type ») == « splice »:
deg = G.degree(n)
is_backbone = deg >= 5

p_x = model.NewIntVar(0, grid_size, f »x_{n} »)
p_y = model.NewIntVar(0, grid_size, f »y_{n} »)

L = model.NewIntVar(1, grid_size, f »L_{n} ») if is_backbone else model.NewIntVar(0, 0, f »L_{n} »)

node_vars[n] = {
« is_external »: False,
« is_splice »: True,
« is_relay »: False,
« x »: p_x,
« y »: p_y,
« is_backbone »: is_backbone,
« L »: L
}

if is_backbone:
if backbone_orientation == « horizontal »:
x_end = model.NewIntVar(0, grid_size, «  »)
model.Add(x_end == p_x + L)
routing_h_x_ints.append(model.NewIntervalVar(p_x, L, x_end, «  »))
routing_h_y_ints.append(model.NewIntervalVar(p_y, 1, p_y + 1, «  »))
else:
y_end = model.NewIntVar(0, grid_size, «  »)
model.Add(y_end == p_y + L)
routing_v_x_ints.append(model.NewIntervalVar(p_x, 1, p_x + 1, «  »))
routing_v_y_ints.append(model.NewIntervalVar(p_y, L, y_end, «  »))
continue

if data.get(‘type’) == ‘relay’:
neighbors = list(G.neighbors(n))
principal = None
h_relay = 0

for neighbor in set(neighbors):
edge_count = G.number_of_edges(n, neighbor)
if G.nodes[neighbor].get(‘type’) not in [‘relay’, ‘splice’, ‘external’]:
principal = neighbor
h_relay = edge_count
break

w = 1
x = model.NewIntVar(0, grid_size – w, f’x_{n}’)
y = model.NewIntVar(0, grid_size, f’y_{n}’)
h = h_relay if h_relay > 0 else 1
y_end = model.NewIntVar(0, grid_size, f’y_end_{n}’)
model.Add(y_end == y + h)

is_left = model.NewBoolVar(f’relay_left_{n}’)

node_vars[n] = {
‘is_external’: False,
‘is_splice’: False,
‘is_relay’: True,
‘x’: x,
‘y’: y,
‘h’: h,
‘w’: w,
‘y_end’: y_end,
‘principal’: principal,
‘is_left’: is_left
}

relays_of_node[principal].append(n)

x_exact_int = model.NewIntervalVar(x, w, x + w,  »)
y_exact_int = model.NewIntervalVar(y, h, y_end,  »)
routing_h_x_ints.append(x_exact_int)
routing_h_y_ints.append(y_exact_int)
routing_v_x_ints.append(x_exact_int)
routing_v_y_ints.append(y_exact_int)

wend = model.NewIntVar(0, grid_size,  »)
model.Add(wend == x + w)
collision_blocks.append({‘x’: x, ‘wend’: wend, ‘y_min’: y, ‘y_max’: y_end, ‘is_relay’: True})
continue

equipment_raw = data.get(‘equipment’,  »)
is_cut = data.get(‘type’) == ‘cut’

if is_cut:
equipment = f’cut_{n}’
min_h = max(1, (G.degree(n) // 2) + 1)
else:
equipment = equipment_raw
min_h = max(1, G.degree(n) + 1)

w = 1
x = model.NewIntVar(0, grid_size – w, f’x_{n}’)
y = model.NewIntVar(0, grid_size, f’y_{n}’)
h = model.NewIntVar(min_h, grid_size, f’h_{n}’)
y_end = model.NewIntVar(0, grid_size, f’y_end_{n}’)
model.Add(y_end == y + h)

node_vars[n] = {
‘is_external’: False,
‘is_splice’: False,
‘is_relay’: False,
‘x’: x,
‘y’: y,
‘h’: h,
‘w’: w,
‘y_end’: y_end,
‘equipment’: equipment,
‘is_cut’: is_cut
}

if equipment not in equipment_groups:
equipment_groups[equipment] = []
equipment_groups[equipment].append(n)

if is_cut:
x_exact_int = model.NewIntervalVar(x, w, x + w,  »)
y_exact_int = model.NewIntervalVar(y, h, y_end,  »)
routing_h_x_ints.append(x_exact_int)
routing_h_y_ints.append(y_exact_int)
routing_v_x_ints.append(x_exact_int)
routing_v_y_ints.append(y_exact_int)

wend = model.NewIntVar(0, grid_size,  »)
model.Add(wend == x + w)
collision_blocks.append({‘x’: x, ‘wend’: wend, ‘y_min’: y, ‘y_max’: y_end})

# ==========================================
# 2. ALIGNEMENT ET BOÎTES VIRTUELLES
# ==========================================

for n, v in node_vars.items():
if v.get(‘is_relay’, False):
p_var = node_vars[v[‘principal’]]
model.Add(v[‘x’] + v[‘w’] == p_var[‘x’]).OnlyEnforceIf(v[‘is_left’])
model.Add(v[‘x’] == p_var[‘x’] + p_var[‘w’]).OnlyEnforceIf(v[‘is_left’].Not())
model.Add(v[‘y’] >= p_var[‘y’] + 1)
model.Add(v[‘y_end’] <= p_var[‘y_end’] – 1)

equipment_side_vars = {}
virtual_boxes_info = {}

for equipment, nodes_in_equipment in equipment_groups.items():
if equipment.startswith(‘cut_’): continue

equipment_side_vars[equipment] = model.NewBoolVar(f’side_{equipment}’)
first_node = nodes_in_equipment[0]

if len(nodes_in_equipment) > 1:
for i in range(1, len(nodes_in_equipment)):
curr_node = nodes_in_equipment[i]
model.Add(node_vars[curr_node][‘x’] == node_vars[first_node][‘x’])

for i in range(len(nodes_in_equipment)):
for j in range(i + 1, len(nodes_in_equipment)):
node_a = nodes_in_equipment[i]
node_b = nodes_in_equipment[j]
a_above_b = model.NewBoolVar(f'{node_a}_above_{node_b}’)
model.Add(node_vars[node_a][‘y_end’] + 1 <= node_vars[node_b][‘y’]).OnlyEnforceIf(a_above_b)
model.Add(node_vars[node_b][‘y_end’] + 1 <= node_vars[node_a][‘y’]).OnlyEnforceIf(a_above_b.Not())

vbox_x = node_vars[first_node][‘x’]
vbox_y_min = model.NewIntVar(0, grid_size, f’vbox_y_min_{equipment}’)
vbox_y_max = model.NewIntVar(0, grid_size, f’vbox_y_max_{equipment}’)

model.AddMinEquality(vbox_y_min, [node_vars[n][‘y’] for n in nodes_in_equipment])
model.AddMaxEquality(vbox_y_max, [node_vars[n][‘y_end’] for n in nodes_in_equipment])
vbox_h = model.NewIntVar(1, grid_size, f’vbox_h_{equipment}’)
model.Add(vbox_h == vbox_y_max – vbox_y_min)

# BLOC RÉINTÉGRÉ : Ajout des intervalles de routage pour la boîte virtuelle
x_exact_int = model.NewIntervalVar(vbox_x, 1, vbox_x + 1,  »)
y_exact_int = model.NewIntervalVar(vbox_y_min, vbox_h, vbox_y_max,  »)
routing_h_x_ints.append(x_exact_int)
routing_h_y_ints.append(y_exact_int)
routing_v_x_ints.append(x_exact_int)
routing_v_y_ints.append(y_exact_int)

virtual_boxes_info[equipment] = {
‘name’: equipment,
‘x’: vbox_x,
‘y_min’: vbox_y_min,
‘h’: vbox_h
}

has_relay_right = model.NewBoolVar(f’vbox_has_relay_right_{equipment}’)
all_relays_in_vbox = []
for child in nodes_in_equipment:
all_relays_in_vbox.extend(relays_of_node[child])

if all_relays_in_vbox:
model.AddMaxEquality(has_relay_right, [node_vars[r][‘is_left’].Not() for r in all_relays_in_vbox])
else:
model.Add(has_relay_right == 0)

w_halo = model.NewIntVar(1, 2, f’vbox_w_halo_{equipment}’)
model.Add(w_halo == 1).OnlyEnforceIf(has_relay_right)
model.Add(w_halo == 2).OnlyEnforceIf(has_relay_right.Not())

x_halo_end = model.NewIntVar(0, grid_size + 5,  »)
model.Add(x_halo_end == vbox_x + w_halo)
halo_x_ints.append(model.NewIntervalVar(vbox_x, w_halo, x_halo_end,  »))

h_halo = model.NewIntVar(1, grid_size + 5,  »)
model.Add(h_halo == vbox_h + 1)
y_halo_end = model.NewIntVar(0, grid_size + 5,  »)
model.Add(y_halo_end == vbox_y_min + h_halo)
halo_y_ints.append(model.NewIntervalVar(vbox_y_min, h_halo, y_halo_end,  »))

vbox_wend = model.NewIntVar(0, grid_size,  »)
model.Add(vbox_wend == vbox_x + 1)
collision_blocks.append({‘x’: vbox_x, ‘wend’: vbox_wend, ‘y_min’: vbox_y_min, ‘y_max’: vbox_y_max})

for n, v in node_vars.items():
if v.get(‘is_relay’):
w, x, y, h = v[‘w’], v[‘x’], v[‘y’], v[‘h’]
is_left = v[‘is_left’]

w_halo = model.NewIntVar(w, w + 1, f’w_halo_{n}’)
model.Add(w_halo == w).OnlyEnforceIf(is_left)
model.Add(w_halo == w + 1).OnlyEnforceIf(is_left.Not())

x_end = model.NewIntVar(0, grid_size + 5,  »)
model.Add(x_end == x + w_halo)
halo_x_ints.append(model.NewIntervalVar(x, w_halo, x_end,  »))

h_padded = model.NewIntVar(2, grid_size + 5,  »)
model.Add(h_padded == h + 1)
y_end = model.NewIntVar(0, grid_size + 5,  »)
model.Add(y_end == y + h_padded)
halo_y_ints.append(model.NewIntervalVar(y, h_padded, y_end,  »))

elif not v.get(‘is_external’) and not v.get(‘is_splice’) and v.get(‘is_cut’):
w, x, y, h = v[‘w’], v[‘x’], v[‘y’], v[‘h’]
has_relay_right = model.NewBoolVar(f’has_relay_right_{n}’)
my_relays = relays_of_node[n]

if my_relays:
model.AddMaxEquality(has_relay_right, [node_vars[r][‘is_left’].Not() for r in my_relays])
else:
model.Add(has_relay_right == 0)

w_halo = model.NewIntVar(w, w + 1, f’w_halo_{n}’)
model.Add(w_halo == w).OnlyEnforceIf(has_relay_right)
model.Add(w_halo == w + 1).OnlyEnforceIf(has_relay_right.Not())

x_end = model.NewIntVar(0, grid_size + 5,  »)
model.Add(x_end == x + w_halo)
halo_x_ints.append(model.NewIntervalVar(x, w_halo, x_end,  »))

h_padded = model.NewIntVar(2, grid_size + 5,  »)
model.Add(h_padded == h + 1)
y_end = model.NewIntVar(0, grid_size + 5,  »)
model.Add(y_end == y + h_padded)
halo_y_ints.append(model.NewIntervalVar(y, h_padded, y_end,  »))

model.AddNoOverlap2D(halo_x_ints, halo_y_ints)

point_nodes = [n for n, v in node_vars.items() if v.get(« is_external ») or v.get(« is_splice »)]
if len(point_nodes) > 1:
MULTI = 1000
point_ids = []
for n in point_nodes:
pid = model.NewIntVar(0, grid_size * MULTI + grid_size, f »pid_{n} »)
model.Add(pid == node_vars[n][« x »] * MULTI + node_vars[n][« y »])
point_ids.append(pid)
model.AddAllDifferent(point_ids)

# ==========================================
# 2.5 ESPACEMENT COLONNE INTER-ÉQUIPEMENTS
# ==========================================

vbox_keys = list(virtual_boxes_info.keys())
for i in range(len(vbox_keys)):
for j in range(i + 1, len(vbox_keys)):
eqA = vbox_keys[i]
eqB = vbox_keys[j]
boxA = virtual_boxes_info[eqA]
boxB = virtual_boxes_info[eqB]

same_x = model.NewBoolVar(f’same_x_{eqA}_{eqB}’)
model.Add(boxA[« x »] == boxB[« x »]).OnlyEnforceIf(same_x)
model.Add(boxA[« x »] != boxB[« x »]).OnlyEnforceIf(same_x.Not())

a_above_b = model.NewBoolVar(f'{eqA}_above_{eqB}’)
b_above_a = model.NewBoolVar(f'{eqB}_above_{eqA}’)

model.Add(boxB[« y_min »] >= boxA[« y_min »] + boxA[« h »] + 2).OnlyEnforceIf(a_above_b)
model.Add(boxA[« y_min »] >= boxB[« y_min »] + boxB[« h »] + 2).OnlyEnforceIf(b_above_a)

model.AddBoolOr([a_above_b, b_above_a]).OnlyEnforceIf(same_x)

for n, v in node_vars.items():
if v.get(‘is_relay’):
principal = v[‘principal’]
p_data = node_vars[principal]
if not p_data.get(‘is_cut’):
p_side_var = equipment_side_vars[p_data[‘equipment’]]
model.Add(v[‘is_left’] != p_side_var)

# ==========================================
# 3. ROUTAGE DES ARÊTES
# ==========================================

node_ports = {n: [] for n in G.nodes()}
cut_side_vars = {n: [] for n in G.nodes() if not node_vars[n].get(‘is_external’)
and not node_vars[n].get(‘is_splice’) and not node_vars[n].get(‘is_relay’)
and node_vars[n].get(‘is_cut’)}
edges_info = []

# CORRECTION ARCHITECTURALE : Initialisation explicite de la liste des pénalités
soft_penalties = []

# CORRECTION ARCHITECTURALE : Injection de key_str et edge_alias pour éviter les fuites de portée
def setup_port(node_id, role, neighbor_id, key_str, edge_alias):
if node_vars[node_id].get(« is_external »):
p_x = node_vars[node_id][« x »]
p_y = node_vars[node_id][« y »]
node_ports[node_id].append({« x »: p_x, « y »: p_y})
return p_x, p_y, None

if node_vars[node_id].get(« is_splice »):
v_splice = node_vars[node_id]
if v_splice.get(« is_backbone »):
p_x = model.NewIntVar(0, grid_size, f »p_x_{node_id}_{key_str}_{role} »)
p_y = model.NewIntVar(0, grid_size, f »p_y_{node_id}_{key_str}_{role} »)
if backbone_orientation == « horizontal »:
model.Add(p_y == v_splice[« y »])
model.Add(p_x >= v_splice[« x »])
model.Add(p_x <= v_splice[« x »] + v_splice[« L »])
else:
model.Add(p_x == v_splice[« x »])
model.Add(p_y >= v_splice[« y »])
model.Add(p_y <= v_splice[« y »] + v_splice[« L »])
node_ports[node_id].append({« x »: p_x, « y »: p_y})
return p_x, p_y, None
else:
p_x, p_y = v_splice[« x »], v_splice[« y »]
node_ports[node_id].append({« x »: p_x, « y »: p_y})
return p_x, p_y, None

if node_vars[node_id].get(‘is_relay’):
v_relay = node_vars[node_id]
p_y = model.NewIntVar(0, grid_size, f’py_{node_id}_{key_str}_{role}’)
model.Add(p_y == v_relay[‘y’] + v_relay[‘h’] // 2)
p_x = model.NewIntVar(0, grid_size, f’px_{node_id}_{key_str}_{role}’)
model.Add(p_x == v_relay[‘x’]).OnlyEnforceIf(v_relay[‘is_left’])
model.Add(p_x == v_relay[‘x’] + v_relay[‘w’]).OnlyEnforceIf(v_relay[‘is_left’].Not())
side_var = v_relay[‘is_left’].Not()
node_ports[node_id].append({‘x’: p_x, ‘y’: p_y, ‘side’: side_var, ‘neighbor’: neighbor_id, ‘alias’: edge_alias})
return p_x, p_y, side_var

is_cut = node_vars[node_id].get(‘is_cut’, False)
if is_cut:
side_var = model.NewBoolVar(f’side_{node_id}_{key_str}_{role}’)
cut_side_vars[node_id].append(side_var)
else:
side_var = equipment_side_vars[node_vars[node_id][‘equipment’]]

p_y = model.NewIntVar(0, grid_size, f’py_{node_id}_{key_str}_{role}’)
model.Add(p_y >= node_vars[node_id][‘y’] + 1)
model.Add(p_y <= node_vars[node_id][‘y_end’] – 1)
p_x = model.NewIntVar(0, grid_size, f’px_{node_id}_{key_str}_{role}’)
model.Add(p_x == node_vars[node_id][‘x’]).OnlyEnforceIf(side_var)
model.Add(p_x == node_vars[node_id][‘x’] + node_vars[node_id][‘w’]).OnlyEnforceIf(side_var.Not())
node_ports[node_id].append({‘x’: p_x, ‘y’: p_y, ‘side’: side_var, ‘neighbor’: neighbor_id, ‘alias’: edge_alias})
return p_x, p_y, side_var

for u, v, key in G.edges(keys=True):
if node_vars[u].get(‘is_relay’, False) and v == node_vars[u].get(‘principal’): continue
if node_vars[v].get(‘is_relay’, False) and u == node_vars[v].get(‘principal’): continue

key_str = f'{u}_{v}_{key}’
edge_alias = (G.edges[u, v, key].get(‘alias’,  ») + G.edges[u, v, key].get(‘sub_alias’,  »))

p_ux, p_uy, side_u = setup_port(u, ‘u’, v, key_str, edge_alias)
p_vx, p_vy, side_v = setup_port(v, ‘v’, u, key_str, edge_alias)

e_vx = model.NewIntVar(0, grid_size, f’e_vx_{key_str}’)

if G.degree(u) == 1 or G.degree(v) == 1:
is_u_splice = node_vars[u].get(« is_splice », False)
is_v_splice = node_vars[v].get(« is_splice », False)
if not is_u_splice and not is_v_splice:
if node_vars[u].get(« equipment ») != node_vars[v].get(« equipment »):
if not allow_bent_leaves:
model.Add(p_uy == p_vy)
model.Add(p_ux != p_vx)
else:
keep_straight = model.NewBoolVar(f’keep_straight_{key_str}’)
model.Add(p_uy == p_vy).OnlyEnforceIf(keep_straight)
model.Add(p_ux != p_vx)
soft_penalties.append(keep_straight.Not())

if side_u is not None:
model.Add(e_vx >= p_ux + 1).OnlyEnforceIf(side_u)
model.Add(e_vx <= p_ux – 1).OnlyEnforceIf(side_u.Not())
if side_v is not None:
model.Add(e_vx >= p_vx + 1).OnlyEnforceIf(side_v)
model.Add(e_vx <= p_vx – 1).OnlyEnforceIf(side_v.Not())

for block in collision_blocks:
model.Add(e_vx != block[‘x’])
model.Add(e_vx != block[‘wend’])
if not block.get(‘is_relay’, False):
model.Add(p_uy != block[‘y_min’])
model.Add(p_uy != block[‘y_max’])
model.Add(p_vy != block[‘y_min’])
model.Add(p_vy != block[‘y_max’])

if not node_vars[u].get(‘is_external’) and not node_vars[v].get(‘is_external’):
model.Add(e_vx > 0)
model.Add(e_vx < grid_size)
model.Add(p_uy > 0)
model.Add(p_uy < grid_size)
model.Add(p_vy > 0)
model.Add(p_vy < grid_size)

h1_xmin, h1_xmax = model.NewIntVar(0, grid_size,  »), model.NewIntVar(0, grid_size,  »)
model.AddMinEquality(h1_xmin, [p_ux, e_vx])
model.AddMaxEquality(h1_xmax, [p_ux, e_vx])
h1_len = model.NewIntVar(0, grid_size,  »)
model.Add(h1_len == h1_xmax – h1_xmin)
h1_y_end = model.NewIntVar(0, grid_size + 5,  »)
model.Add(h1_y_end == p_uy + 1)
routing_h_x_ints.append(model.NewIntervalVar(h1_xmin, h1_len, h1_xmax,  »))
routing_h_y_ints.append(model.NewIntervalVar(p_uy, 1, h1_y_end,  »))

v_ymin, v_ymax = model.NewIntVar(0, grid_size,  »), model.NewIntVar(0, grid_size,  »)
model.AddMinEquality(v_ymin, [p_uy, p_vy])
model.AddMaxEquality(v_ymax, [p_uy, p_vy])
v_len = model.NewIntVar(0, grid_size,  »)
model.Add(v_len == v_ymax – v_ymin)
v_x_end = model.NewIntVar(0, grid_size + 5,  »)
model.Add(v_x_end == e_vx + 1)
routing_v_x_ints.append(model.NewIntervalVar(e_vx, 1, v_x_end,  »))
routing_v_y_ints.append(model.NewIntervalVar(v_ymin, v_len, v_ymax,  »))

h2_xmin, h2_xmax = model.NewIntVar(0, grid_size,  »), model.NewIntVar(0, grid_size,  »)
model.AddMinEquality(h2_xmin, [e_vx, p_vx])
model.AddMaxEquality(h2_xmax, [e_vx, p_vx])
h2_len = model.NewIntVar(0, grid_size,  »)
model.Add(h2_len == h2_xmax – h2_xmin)
h2_y_end = model.NewIntVar(0, grid_size + 5,  »)
model.Add(h2_y_end == p_vy + 1)
routing_h_x_ints.append(model.NewIntervalVar(h2_xmin, h2_len, h2_xmax,  »))
routing_h_y_ints.append(model.NewIntervalVar(p_vy, 1, h2_y_end,  »))

edges_info.append({
‘u’: u, ‘v’: v, ‘key’: key, ‘p_ux’: p_ux, ‘p_uy’: p_uy, ‘p_vx’: p_vx, ‘p_vy’: p_vy,
‘e_vx’: e_vx, ‘key_str’: key_str,
‘h1_len’: h1_len, ‘v_len’: v_len, ‘h2_len’: h2_len,
‘seg_H1’: {‘y’: p_uy, ‘xmin’: h1_xmin, ‘xmax’: h1_xmax},
‘seg_H2’: {‘y’: p_vy, ‘xmin’: h2_xmin, ‘xmax’: h2_xmax},
‘seg_V’: {‘x’: e_vx, ‘ymin’: v_ymin, ‘ymax’: v_ymax}
})

MULTI_C = 1000
for e in edges_info:
e[‘corner_ids’] = []
pts = [(e[‘p_ux’], e[‘p_uy’]), (e[‘e_vx’], e[‘p_uy’]), (e[‘e_vx’], e[‘p_vy’]), (e[‘p_vx’], e[‘p_vy’])]
for i, (px, py) in enumerate(pts):
cid = model.NewIntVar(0, grid_size * MULTI_C + grid_size, f’cid_{e[« key_str »]}_{i}’)
model.Add(cid == px * MULTI_C + py)
e[‘corner_ids’].append(cid)

for i in range(len(edges_info)):
for j in range(i + 1, len(edges_info)):
eA = edges_info[i]
eB = edges_info[j]
shared_nodes = set([eA[‘u’], eA[‘v’]]).intersection([eB[‘u’], eB[‘v’]])
has_shared_splice = any(node_vars[n].get(‘is_splice’) for n in shared_nodes)
if has_shared_splice:
continue
for cid_A in eA[‘corner_ids’]:
for cid_B in eB[‘corner_ids’]:
model.Add(cid_A != cid_B)

for cut_node, side_vars in cut_side_vars.items():
my_relays = relays_of_node[cut_node]
total_edges = len(side_vars) + sum(node_vars[r][‘h’] for r in my_relays)
if total_edges > 0:
right_count = model.NewIntVar(0, total_edges, f’right_count_{cut_node}’)
right_relay_edge_vars = []
for r in my_relays:
h_r = node_vars[r][‘h’]
r_right = model.NewIntVar(0, h_r, f’r_right_{r}’)
model.Add(r_right == h_r).OnlyEnforceIf(node_vars[r][‘is_left’].Not())
model.Add(r_right == 0).OnlyEnforceIf(node_vars[r][‘is_left’])
right_relay_edge_vars.append(r_right)
model.Add(right_count == sum(side_vars) + sum(right_relay_edge_vars))
model.Add(right_count == total_edges // 2)

for n in G.nodes():
if node_vars[n].get(‘is_splice’) or node_vars[n].get(‘is_relay’): continue
ports = node_ports[n]
if len(ports) <= 1: continue
if node_vars[n].get(‘is_external’):
for i in range(len(ports)):
for j in range(i + 1, len(ports)):
diff_x, diff_y = model.NewBoolVar( »), model.NewBoolVar( »)
model.Add(ports[i][‘x’] != ports[j][‘x’]).OnlyEnforceIf(diff_x)
model.Add(ports[i][‘y’] != ports[j][‘y’]).OnlyEnforceIf(diff_y)
model.AddBoolOr([diff_x, diff_y])
elif not node_vars[n].get(‘is_cut’) and not node_vars[n].get(‘is_splice’):
model.AddAllDifferent([p[‘y’] for p in ports])
else:
for i in range(len(ports)):
for j in range(i + 1, len(ports)):
same_side = model.NewBoolVar(f’same_side_{n}_{i}_{j}’)
model.Add(ports[i][‘side’] == ports[j][‘side’]).OnlyEnforceIf(same_side)
model.Add(ports[i][‘side’] != ports[j][‘side’]).OnlyEnforceIf(same_side.Not())
alias_i = ports[i].get(‘alias’,  »)
alias_j = ports[j].get(‘alias’,  »)
if (alias_i is not None and alias_i !=  ») and alias_i == alias_j:
model.Add(same_side == 0)
model.Add(ports[i][‘y’] != ports[j][‘y’]).OnlyEnforceIf(same_side)

model.AddNoOverlap2D(routing_h_x_ints, routing_h_y_ints)
model.AddNoOverlap2D(routing_v_x_ints, routing_v_y_ints)

# ==========================================
# 4. FONCTION OBJECTIVE & DÉTECTION CROISEMENT
# ==========================================

POIDS_FEUILLE_DROITE = 100
POIDS_CROISEMENT = 100
POIDS_VIRAGE = 10
POIDS_HAUTEUR = 5
POIDS_DISTANCE = 1
POIDS_SPLICE_L = 0.5
POIDS_RATIO = 0
POIDS_ANCRAGES = 30

cost_elements = []
max_x = model.NewIntVar(0, grid_size, ‘max_x’)
max_y = model.NewIntVar(0, grid_size, ‘max_y’)
all_x_vars = []
all_y_vars = []

for n, v in node_vars.items():
if v.get(‘is_external’): continue
all_x_vars.append(v[‘x’] + v.get(‘w’, 0))
if ‘y_end’ in v:
all_y_vars.append(v[‘y_end’])
else:
all_y_vars.append(v[‘y’])

for e in edges_info:
all_x_vars.extend([e[‘seg_H1’][‘xmax’], e[‘seg_H2’][‘xmax’], e[‘seg_V’][‘x’]])
all_y_vars.extend([e[‘seg_H1’][‘y’], e[‘seg_H2’][‘y’], e[‘seg_V’][‘ymax’]])

if all_x_vars and all_y_vars:
model.AddMaxEquality(max_x, all_x_vars)
model.AddMaxEquality(max_y, all_y_vars)

target_y = model.NewIntVar(0, grid_size * 3, ‘target_y’)
model.Add(target_y == 3 * max_x)
ratio_penalty = model.NewIntVar(0, grid_size * 3, ‘ratio_penalty’)
model.AddAbsEquality(ratio_penalty, max_y – target_y)
cost_elements.append(POIDS_RATIO * ratio_penalty)

if allow_bent_leaves and soft_penalties:
for penalty in soft_penalties:
cost_elements.append(POIDS_FEUILLE_DROITE * penalty)

global_min_x = model.NewIntVar(0, grid_size, ‘global_min_x’)
global_min_y = model.NewIntVar(0, grid_size, ‘global_min_y’)
min_candidates_x = []
min_candidates_y = []

for n, v in node_vars.items():
if v.get(‘is_external’): continue
min_candidates_x.append(v[‘x’])
min_candidates_y.append(v[‘y’])

for e in edges_info:
min_candidates_x.extend([e[‘seg_H1’][‘xmin’], e[‘seg_H2’][‘xmin’], e[‘seg_V’][‘x’]])
min_candidates_y.extend([e[‘seg_H1’][‘y’], e[‘seg_H2’][‘y’], e[‘seg_V’][‘ymin’]])

if min_candidates_x and min_candidates_y:
model.AddMinEquality(global_min_x, min_candidates_x)
model.AddMinEquality(global_min_y, min_candidates_y)

for n in G.nodes():
v = node_vars[n]
if v.get(« is_splice ») and v.get(« is_backbone »):
cost_elements.append(POIDS_SPLICE_L * v[« L »])
elif not v.get(« is_external ») and not v.get(« is_splice ») and not v.get(« is_relay »):
cost_elements.append(POIDS_HAUTEUR * v[« h »])

for n, v in node_vars.items():
if v.get(« is_splice ») and v.get(« is_backbone »):
ports = node_ports[n]
if not ports: continue
used_positions = []
for k in range(grid_size + 1):
pos_used = model.NewBoolVar(f »pos_used_{n}_{k} »)
port_at_pos = []
for i, p in enumerate(ports):
coord_var = p[« x »] if backbone_orientation == « horizontal » else p[« y »]
is_here = model.NewBoolVar(f »is_here_{n}_{k}_{i} »)
model.Add(coord_var == k).OnlyEnforceIf(is_here)
model.Add(coord_var != k).OnlyEnforceIf(is_here.Not())
port_at_pos.append(is_here)
model.AddMaxEquality(pos_used, port_at_pos)
used_positions.append(pos_used)
nb_ancrages = model.NewIntVar(1, len(ports), f »nb_ancrages_{n} »)
model.Add(nb_ancrages == sum(used_positions))
cost_elements.append(POIDS_ANCRAGES * nb_ancrages)

for e in edges_info:
cost_elements.append(POIDS_DISTANCE * e[‘h1_len’])
cost_elements.append(POIDS_DISTANCE * e[‘v_len’])
cost_elements.append(POIDS_DISTANCE * e[‘h2_len’])
is_turning = model.NewBoolVar(f’is_turning_{e[« key_str »]}’)
model.Add(e[‘v_len’] > 0).OnlyEnforceIf(is_turning)
model.Add(e[‘v_len’] == 0).OnlyEnforceIf(is_turning.Not())
cost_elements.append(POIDS_VIRAGE * is_turning)

def add_crossing_detection(h_seg, v_seg, prefix):
cross_var = model.NewBoolVar(f’cross_{prefix}’)
c1, c2, c3, c4 = [model.NewBoolVar( ») for _ in range(4)]
model.Add(h_seg[‘xmin’] <= v_seg[‘x’]).OnlyEnforceIf(c1)
model.Add(h_seg[‘xmin’] > v_seg[‘x’]).OnlyEnforceIf(c1.Not())
model.Add(v_seg[‘x’] <= h_seg[‘xmax’]).OnlyEnforceIf(c2)
model.Add(v_seg[‘x’] > h_seg[‘xmax’]).OnlyEnforceIf(c2.Not())
model.Add(v_seg[‘ymin’] <= h_seg[‘y’]).OnlyEnforceIf(c3)
model.Add(v_seg[‘ymin’] > h_seg[‘y’]).OnlyEnforceIf(c3.Not())
model.Add(h_seg[‘y’] <= v_seg[‘ymax’]).OnlyEnforceIf(c4)
model.Add(h_seg[‘y’] > v_seg[‘ymax’]).OnlyEnforceIf(c4.Not())
model.AddBoolAnd([c1, c2, c3, c4]).OnlyEnforceIf(cross_var)
model.AddBoolOr([c1.Not(), c2.Not(), c3.Not(), c4.Not()]).OnlyEnforceIf(cross_var.Not())
return cross_var

for i in range(len(edges_info)):
for j in range(i + 1, len(edges_info)):
eA = edges_info[i]
eB = edges_info[j]
if eA[‘u’] in (eB[‘u’], eB[‘v’]) or eA[‘v’] in (eB[‘u’], eB[‘v’]): continue
cross1 = add_crossing_detection(eA[‘seg_H1’], eB[‘seg_V’], f'{eA[« key_str »]}_{eB[« key_str »]}_H1_V’)
cross2 = add_crossing_detection(eA[‘seg_H2’], eB[‘seg_V’], f'{eA[« key_str »]}_{eB[« key_str »]}_H2_V’)
cross3 = add_crossing_detection(eB[‘seg_H1’], eA[‘seg_V’], f'{eB[« key_str »]}_{eA[« key_str »]}_H1_V’)
cross4 = add_crossing_detection(eB[‘seg_H2’], eA[‘seg_V’], f'{eB[« key_str »]}_{eA[« key_str »]}_H2_V’)
cost_elements.extend([
POIDS_CROISEMENT * cross1,
POIDS_CROISEMENT * cross2,
POIDS_CROISEMENT * cross3,
POIDS_CROISEMENT * cross4
])

model.Minimize(sum(cost_elements))

# ==========================================
# 5. RÉSOLUTION ET SAUVEGARDE
# ==========================================

solver = cp_model.CpSolver()
solver.parameters.num_search_workers = 14
number_of_calculated_edges = 0.0

for u, v, data in G.edges(data=True):
t_u, t_v = G.nodes[u].get(‘type’), G.nodes[v].get(‘type’)
if t_u == ‘relay’ or t_v == ‘relay’:
if t_u == t_v or t_u in [‘external’, ‘splice’] or t_v in [‘external’, ‘splice’]:
number_of_calculated_edges += 1.0
else:
number_of_calculated_edges += 1.0

print(number_of_calculated_edges)
solver.parameters.max_time_in_seconds = max(5, 5 * number_of_calculated_edges)
status = solver.Solve(model)

if status == cp_model.OPTIMAL or status == cp_model.FEASIBLE:
for n, v in node_vars.items():
if v.get(‘is_external’):
G.nodes[n][‘type’] = ‘external’
elif v.get(‘is_splice’):
G.nodes[n][« type »] = « splice »
G.nodes[n][« x »] = solver.Value(v[« x »])
G.nodes[n][« y »] = solver.Value(v[« y »])
G.nodes[n][« w »] = 0
G.nodes[n][« h »] = 0
if v.get(« is_backbone »):
G.nodes[n][« is_backbone »] = True
G.nodes[n][« L »] = solver.Value(v[« L »])
G.nodes[n][« backbone_orientation »] = backbone_orientation
else:
G.nodes[n][« is_backbone »] = False
elif v.get(‘is_relay’):
G.nodes[n][‘type’] = ‘relay’
G.nodes[n][‘x’] = solver.Value(v[‘x’])
G.nodes[n][‘y’] = solver.Value(v[‘y’])
G.nodes[n][‘w’] = v[‘w’]
G.nodes[n][‘h’] = v[‘h’]
G.nodes[n][‘is_left’] = solver.Value(v[‘is_left’]) == 1
else:
G.nodes[n][‘x’] = solver.Value(v[‘x’])
G.nodes[n][‘y’] = solver.Value(v[‘y’])
G.nodes[n][‘w’] = v[‘w’]
G.nodes[n][‘h’] = solver.Value(v[‘h’])
ports = node_ports.get(n, [])
equipment = v[‘equipment’]
side_var_val = solver.Value(equipment_side_vars[equipment])
G.nodes[n][‘is_left’] = (side_var_val == 0)

for e in edges_info:
start_x = solver.Value(e[‘p_ux’])
start_y = solver.Value(e[‘p_uy’])
vx = solver.Value(e[‘e_vx’])
end_x = solver.Value(e[‘p_vx’])
end_y = solver.Value(e[‘p_vy’])
route = [(start_x, start_y), (vx, start_y), (vx, end_y), (end_x, end_y)]
G.edges[e[‘u’], e[‘v’], e[‘key’]][‘route’] = route

if node_vars[e[‘u’]].get(‘is_external’):
G.nodes[e[‘u’]][‘port_x’] = start_x
G.nodes[e[‘u’]][‘port_y’] = start_y
if node_vars[e[‘v’]].get(‘is_external’):
G.nodes[e[‘v’]][‘port_x’] = end_x
G.nodes[e[‘v’]][‘port_y’] = end_y

virtual_boxes_results = {}
for vbox in virtual_boxes_info.keys():
virtual_boxes_results[vbox] = {
‘name’: virtual_boxes_info[vbox][‘name’],
‘x’: solver.Value(virtual_boxes_info[vbox][‘x’]),
‘y’: solver.Value(virtual_boxes_info[vbox][‘y_min’]),
‘w’: 1,
‘h’: solver.Value(virtual_boxes_info[vbox][‘h’])
}
G.graph[‘virtual_boxes’] = virtual_boxes_results

relays = [n for n, v in node_vars.items() if v.get(‘is_relay’)]
for relay in relays:
r_data = G.nodes[relay]
r_x, r_y, r_h = r_data[‘x’], r_data[‘y’], r_data[‘h’]
is_left = r_data[‘is_left’]
principal = node_vars[relay][‘principal’]
internal_edges = []
outgoing_edges = []

for u, v, k in G.edges(relay, keys=True):
if v == principal or u == principal:
internal_edges.append((u, v, k))
else:
outgoing_edges.append((u, v, k))

if not outgoing_edges: continue

out_u, out_v, out_k = outgoing_edges[0]
out_route = G.edges[out_u, out_v, out_k][‘route’]
JX = r_x if is_left else r_x + 1
JY = r_y + r_h // 2
PX = r_x + 1 if is_left else r_x
y_slots = [r_y + i for i in range(r_h)]

for i, (u, v, k) in enumerate(internal_edges):
PY = y_slots[i]
if u == principal:
final_route = [(PX, PY), (JX, PY), (JX, JY)]
else:
final_route = [(JX, JY), (JX, PY), (PX, PY)]
G.edges[u, v, k][‘route’] = final_route
if ‘ports’ not in G.nodes[principal]:
G.nodes[principal][‘ports’] = []
G.nodes[principal][‘ports’].append((PX, PY))

G.nodes[relay][‘x’] = JX
G.nodes[relay][‘y’] = JY
G.nodes[relay][‘w’] = 0
G.nodes[relay][‘h’] = 0
G.nodes[relay][‘principal’] = principal

def extract_backbones_to_chains(G: nx.MultiGraph) -> nx.MultiGraph:
«  » »Transforme les splices étendus (backbones) en chaînes de splices ponctuels. » » »
backbones = {n: d for n, d in G.nodes(data=True) if d.get(« type ») == « splice » and d.get(« is_backbone »)}
if not backbones: return G

bb_points = {n: set() for n in backbones}
edges_to_rebuild = []

for u, v, k, d in G.edges(keys=True, data=True):
route = d.get(« route », [])
if not route: continue
if u in backbones: bb_points[u].add(route[0])
if v in backbones: bb_points[v].add(route[-1])
edges_to_rebuild.append((u, v, k, d))

bb_mapping = {n: {} for n in backbones}
for b_node, b_data in backbones.items():
is_horizontal = b_data.get(« backbone_orientation ») == « horizontal »
b_alias, b_sub_alias = None, None
for original_u, original_v, k, d in edges_to_rebuild:
if original_u == b_node or original_v == b_node:
b_alias = d.get(« alias »)
b_sub_alias = d.get(« sub_alias »)
break

if is_horizontal:
sorted_pts = sorted(list(bb_points[b_node]), key=lambda p: p[0])
else:
sorted_pts = sorted(list(bb_points[b_node]), key=lambda p: p[1])
sub_nodes = []
for i, pt in enumerate(sorted_pts):
sub_node_id = f »{b_node}_sub_{i} »
sub_nodes.append(sub_node_id)
bb_mapping[b_node][pt] = sub_node_id
G.add_node(sub_node_id, type= »splice », x=pt[0], y=pt[1], w=0, h=0,
is_backbone=False, alias=b_alias, sub_alias=b_sub_alias)

for i in range(len(sub_nodes) – 1):
node_a = sub_nodes[i]
node_b = sub_nodes[i + 1]
pt_a = sorted_pts[i]
pt_b = sorted_pts[i + 1]
chain_route = [pt_a, pt_b]
G.add_edge(node_a, node_b, type= »connection », route=chain_route,
alias=b_alias, sub_alias=b_sub_alias)

for b_node in backbones: G.remove_node(b_node)

for original_u, original_v, k, d in edges_to_rebuild:
new_u = original_u
new_v = original_v
if original_u in backbones:
pt_u = d[« route »][0]
new_u = bb_mapping[original_u][pt_u]
if original_v in backbones:
pt_v = d[« route »][-1]
new_v = bb_mapping[original_v][pt_v]
G.add_edge(new_u, new_v, key=k, **d)
return G

for u, v, k in G.edges(keys=True):
if ‘route’ in G.edges[u, v, k]:
G.edges[u, v, k][‘route’] = clean_and_merge_route(G.edges[u, v, k][‘route’])

G = extract_backbones_to_chains(G)
return G
else:
# CORRECTION ARCHITECTURALE : alignement de la signature lors de l’appel récursif
if not allow_bent_leaves:
print(‘Placement strict impossible (INFEASIBLE). Relance avec relâchement des arêtes droites’)
return solve_nx_placement(G, backbone_orientation=backbone_orientation, allow_bent_leaves=True)
print(‘Placement totalement non réussi (même avec relâchement)’)
return nx.MultiGraph()