# Passive Optical Network (PON) Planning
# Optimizes placement of optical splitters and fiber connections
# to minimize total network cost while satisfying flow requirements
# Demonstrates multi-commodity flow with bidirectional arcs and capacity constraints
model pon
set S = loadSplitters() # Potential splitter locations
set U = loadONUs() # Optical Network Units (end users)
set C = loadCOs() # Central Offices
set I = loadIntermediate() # Intermediate nodes
set V = loadVertices() # All vertices in network
set K = loadCommodities() # Commodities (ONU and splitter flows)
set E = loadEdges() # Undirected edges
set A = loadArcs() # Directed arcs (bidirectional)
# Adjacency structure
set sigma(V) = loadIncomming() # Incoming arcs for each vertex
set delta(V) = loadOutgoing() # Outgoing arcs for each vertex
# Cost parameters
const c = loadNodeCosts(), forall v in V # Cost to install equipment at vertex v
const d = loadArcWeights(), forall a in A # Distance/cost weight for arc a
const t = 10.0 # Cost multiplier for fiber installation
const gamma = 1.0 # Cost per unit flow per distance
const kappa = 32 # Maximum capacity per splitter
const M = loadM() # Big-M constant for logical constraints
# Decision variables
bin y, forall i in S # 1 if splitter is installed at location i
int 0 <= N <= 8, forall i in S # Number of connections at splitter i
var 0 <= f <= 1, forall k in K, forall a in A # Flow of commodity k on arc a
bin x, forall e in E # 1 if edge e is selected for fiber
# Objective: Minimize total network cost (splitters + fiber + flow routing)
min sum_{s in S}{y_{s} * c_{s}} + sum_{e in E}{x_{e} * d_{e} * t} + sum_{k in K}{sum_{a in A}{f_{k, a} * d_{a} * gamma}}
# Flow conservation for ONUs: Each ONU generates -1 unit of its own commodity
constr sum_{k in U}{[sum_{a in delta(v)}{f_{k,a}} - sum_{a in sigma(v)}{f_{k,a}}]} = -1, forall v in U
# Flow conservation for splitters: Splitters aggregate N connections
constr sum_{k in U}{[sum_{a in delta(v)}{f_{k,a}} - sum_{a in sigma(v)}{f_{k,a}}]} = N_{v}, forall v in S
# Flow conservation for COs: Central offices are neutral for ONU flows
constr sum_{k in U}{[sum_{a in delta(v)}{f_{k,a}} - sum_{a in sigma(v)}{f_{k,a}}]} = 0, forall v in C
# Flow conservation for intermediate nodes: Pass-through for ONU flows
constr sum_{k in U}{[sum_{a in delta(v)}{f_{k,a}} - sum_{a in sigma(v)}{f_{k,a}}]} = 0, forall v in I
# Edge capacity constraint: Flow on both directions of edge limited by edge selection
constr sum_{k in K}{f_{k,e} + f_{k,e + |E|}} <= x_{e} * M, forall e in E
# Splitter capacity constraint: Number of connections limited by splitter capacity
constr N_{s} <= kappa * y_{s}, forall s in S
# Flow conservation for splitter commodity at splitters: Each installed splitter generates -1 unit
constr sum_{k in S}{[sum_{a in delta(v)}{f_{k,a}} - sum_{a in sigma(v)}{f_{k,a}}]} = -y_{v}, forall v in S
# Flow conservation for splitter commodity at COs: All splitter flows terminate at COs
constr sum_{k in S}{[sum_{a in delta(v)}{f_{k,a}} - sum_{a in sigma(v)}{f_{k,a}}]} = sum_{s in S}{y_{s}}, forall v in C
# Flow conservation for splitter commodity at ONUs: Pass-through
constr sum_{k in S}{[sum_{a in delta(v)}{f_{k,a}} - sum_{a in sigma(v)}{f_{k,a}}]} = 0, forall v in U
# Flow conservation for splitter commodity at intermediate nodes: Pass-through
constr sum_{k in S}{[sum_{a in delta(v)}{f_{k,a}} - sum_{a in sigma(v)}{f_{k,a}}]} = 0, forall v in I
end
import elytica
import json
# Load network topology from JSON file
file = open('micronet.json')
graph = json.load(file)
# Add node IDs for indexing
edges = [{**item, 'nid': index} for index, item in enumerate(graph["edges"])]
nodes = [{**item, 'nid': index} for index, item in enumerate(graph["nodes"])]
def loadSplitters():
"""Return IDs of potential splitter locations (triangle nodes)"""
return [n['nid'] for n in nodes if n['type'] == 'triangle']
def loadONUs():
"""Return IDs of Optical Network Units (circle nodes)"""
return [n['nid'] for n in nodes if n['type'] == 'circle']
def loadIntermediate():
"""Return IDs of intermediate routing nodes"""
return [n['nid'] for n in nodes if n['type'] is None]
def loadCOs():
"""Return IDs of Central Offices (square nodes)"""
return [n['nid'] for n in nodes if n['type'] == 'square']
def loadCommodities():
"""Commodities are flows from ONUs and splitters"""
return loadONUs() + loadSplitters()
def loadVertices():
"""Return all vertex IDs"""
return [n['nid'] for n in nodes]
def loadEdges():
"""Return all edge IDs"""
return [a['nid'] for a in edges]
def loadM():
"""Big-M constant: large enough to not restrict flow"""
return len(edges) * 2
def generateArcs(edges):
"""
Generate bidirectional arcs from undirected edges.
Each edge becomes two arcs: forward and backward.
Arc IDs: forward arc keeps original nid, backward arc gets nid + |E|
"""
edge_len = len(edges)
return [
{
'weight': edge['weight'],
'source': direction[0],
'target': direction[1],
'id': edge['id'] + suffix,
'label': edge['label'],
'nid': edge['nid'] if index == 0 else edge['nid'] + edge_len
}
for edge in edges
for index, (direction, suffix) in enumerate([
((edge['source'], edge['target']), '-1'),
((edge['target'], edge['source']), '-2')
])
]
# Generate directed arcs from undirected edges
arcs = generateArcs(edges)
def loadArcs():
"""Return all arc IDs"""
return [a['nid'] for a in arcs]
def loadIncomming():
"""Return incoming arcs for each vertex"""
return {n['nid']: [a['nid'] for a in arcs if a['target'] == n['id']] for n in nodes}
def loadOutgoing():
"""Return outgoing arcs for each vertex"""
return {n['nid']: [a['nid'] for a in arcs if a['source'] == n['id']] for n in nodes}
def loadArcWeights():
"""Return weight (distance/cost) for each arc"""
return {a['nid']: a['weight'] for a in arcs}
def loadNodeCosts():
"""Return installation cost for each node"""
return {n['nid']: 0 if not n['modules'] else n['modules'][0]['cost'] for n in nodes}
def get_color(node):
"""Assign colors based on node type"""
if node["type"] == 'triangle':
return '#0000ff' # Blue for splitters
if node["type"] == 'square':
return '#00ff00' # Green for COs
if node["type"] == 'circle':
return '#ff0000' # Red for ONUs
return '#000000' # Black for intermediate
def main():
# Calculate network centroid for visualization
mid_x = sum(node["x"] for node in nodes) / len(nodes)
mid_y = sum(node["y"] for node in nodes) / len(nodes)
# Translate nodes relative to centroid
translated = [{"x": node["x"] - mid_x, "y": node["y"] - mid_y} for node in nodes]
# Find scaling range
min_x, max_x = min(node["x"] for node in translated), max(node["x"] for node in translated)
min_y, max_y = min(node["y"] for node in translated), max(node["y"] for node in translated)
# Prepare visual nodes with scaled coordinates
visual_vertices = [{
**node,
"x": ((node["x"] - mid_x - min_x) / (max_x - min_x or 1)) * 500,
"y": ((node["y"] - mid_y - min_y) / (max_y - min_y or 1)) * 500,
"name": node["id"],
"symbol": node["type"],
"symbolSize": node["size"],
"itemStyle": {"color": node["color"] if node["color"] is not None else get_color(node)},
"id": None
} for node in nodes]
# Prepare visual edges
visual_edges = [{
**edge,
"id": None,
"weight": None,
"label": None,
"symbol": None
} for edge in edges]
# Clean up None values for JSON serialization
visual_vertices = [{k: v for k, v in node.items() if v is not None} for node in visual_vertices]
visual_edges = [{k: v for k, v in edge.items() if v is not None} for edge in visual_edges]
# Initialize and solve the optimization model
elytica.init_model("pon")
elytica.run_model("pon")
print("\n" + "="*70)
print("PASSIVE OPTICAL NETWORK PLANNING SOLUTION")
print("="*70)
# Extract solution: selected edges and installed splitters
resulting_edges = [
e for e in elytica.get_model_set("pon", "E")
if elytica.get_variable_value("pon", f"x{e}") > 0.5
]
resulting_splitters = [
s for s in elytica.get_model_set("pon", "S")
if elytica.get_variable_value("pon", f"y{s}") > 0.5
]
print(f"\nInstalled Splitters: {len(resulting_splitters)}")
print(f"Selected Fiber Links: {len(resulting_edges)}")
# Get splitter connections
print("\nSplitter Details:")
print("-"*70)
for s in resulting_splitters:
connections = elytica.get_variable_value("pon", f"N{s}")
node_id = nodes[s]['id']
print(f" Splitter at {node_id}: {int(connections)} connections")
# Get objective value
obj_value = elytica.get_best_primal_bound("pon")
print("\n" + "-"*70)
print(f"Total Network Cost: {obj_value:.2f}")
print("="*70)
# Filter visualization to show only selected components
visual_edges = [e for e in visual_edges if e['nid'] in resulting_edges]
no_splitters = [v for v in visual_vertices if ('type' not in v) or (v['type'] != 'triangle')]
splitters = [v for v in visual_vertices if v['nid'] in resulting_splitters]
visual_vertices = splitters + no_splitters
# Create ECharts graph visualization
chart = {
"title": {"text": "Optimal PON Network Design"},
"tooltip": {},
"series": [{
"roam": 1, # Enable zoom (use 1 instead of 'zoom' for JSON serialization)
"type": 'graph',
"layout": 'none', # Use provided coordinates
"data": visual_vertices,
"edges": visual_edges,
"lineStyle": {
"opacity": 1,
"width": 3
}
}]
}
# Prepare results
results = {
"objective_value": obj_value,
"installed_splitters": resulting_splitters,
"selected_edges": resulting_edges,
"num_splitters": len(resulting_splitters),
"num_fiber_links": len(resulting_edges),
"chart": chart
}
elytica.write_results(json.dumps(results, indent=2))
return 0
# Expected JSON file format (micronet.json):
#
# {
# "nodes": [
# {
# "id": "node1",
# "type": "circle", # "circle" = ONU, "triangle" = splitter, "square" = CO, null = intermediate
# "x": 100.5,
# "y": 200.3,
# "size": 10,
# "color": "#ff0000", # Optional
# "modules": [{"cost": 1500}] # Equipment cost
# },
# ...
# ],
# "edges": [
# {
# "id": "edge1",
# "source": "node1",
# "target": "node2",
# "weight": 150.0, # Distance in meters
# "label": "Link A"
# },
# ...
# ]
# }
#
# Node types:
# - "circle": Optical Network Units (end users/customers)
# - "triangle": Potential splitter locations
# - "square": Central Offices (headend equipment)
# - null: Intermediate routing nodes
Read this through MCP: read with path: "examples/pon-planning", or resource hlpl://examples/pon-planning.