Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def mock_max_flow_app(topo):
"""
Return a simple mock maxflow app with a single traffic class
The traffic class will be between two randomly chosen nodes
:return:
"""
appconfig = {
'name': u'mockmaxflowapp',
'constraints': [ALLOCATE_FLOW],
'obj': OBJ_MAX_ALL_FLOW,
'predicate': null_predicate,
'resource_cost': {RES_BANDWIDTH: 100}
}
# Create an application based on our config
app = App(_get_pptc(topo, appconfig), **appconfig)
return app
"""
global _topology, _paths
if request.method == 'GET':
if _topology is None:
return
return jsonify(_topology.to_json())
elif request.method == 'POST':
data = request.get_json()
logger.debug(data)
_topology = Topology.from_json(data)
logging.info('Topology read successfully')
_paths = {}
for s in _topology.nodes():
_paths[s] = {}
for t in _topology.nodes():
_paths[s][t] = list(generate_paths_ie(s, t, _topology, null_predicate, 100, 5))
return ""
else:
abort(405) # method not allowed
# SET THIS TO THE TOPOLOGY WE ARE TESTING
#_topology = None
_topology = nx.DiGraph()
_topology.add_node(0, services='switch',resources={})
_topology.add_node(1, services='switch',resources={})
_topology.add_node(2, services='switch',resources={})
_topology.add_edge(0, 1, source=0, target=1, resources={'bw': 10000})
_topology.add_edge(1, 0, source=1, target=0, resources={'bw': 10000})
_topology.add_edge(2, 1, source=2, target=1, resources={'bw': 10000})
_topology.add_edge(1, 2, source=1, target=2, resources={'bw': 10000})
_topology = Topology(u'NoName', _topology)
_predicatedict = {
'null': null_predicate,
'null_predicate': null_predicate,
}
def assign_to_tc(tcs, paths, name):
"""
Assign paths to traffic classes based on the ingress & egress nodes.
To be used only if there is one traffic class (predicate)
per ingress-egress pair.
:param tcs: list of traffic classes
:param paths: dictionary of paths: ..
paths[ingress][egress] = list_of_paths
:return: paths per traffic class
"""
from sol.path.predicates import null_predicate
from sol.topology.generators import complete_topology
from sol.topology.traffic import TrafficClass
from sol.utils.const import *
if __name__ == '__main__':
# Generate a topology:
topo = complete_topology(5) # 5 nodes
# Application configuration
appconfig = {
'name': u'minLatencyApp',
'constraints': [ALLOCATE_FLOW, ROUTE_ALL],
'obj': OBJ_MIN_LATENCY,
'predicate': null_predicate,
'resource_cost': {BANDWIDTH: 100}
}
# Generate a single traffic class:
# TrafficClass (id, name, source node, destination node)
# For now don't worry about IP addresses.
tc = [TrafficClass(1, u'classname', 0, 4)]
# Generate all paths for this traffic class
pptc = generate_paths_tc(topo, tc, appconfig['predicate'],
cutoff=100)
# Create an application based on our config
app = App(pptc, **appconfig)
# Create and solve an optimization based on the app
opt = from_app(topo, app)
opt.solve()
# SET THIS TO THE TOPOLOGY WE ARE TESTING
#_topology = None
_topology = nx.DiGraph()
_topology.add_node(0, services='switch',resources={})
_topology.add_node(1, services='switch',resources={})
_topology.add_node(2, services='switch',resources={})
_topology.add_edge(0, 1, source=0, target=1, resources={'bw': 10000})
_topology.add_edge(1, 0, source=1, target=0, resources={'bw': 10000})
_topology.add_edge(2, 1, source=2, target=1, resources={'bw': 10000})
_topology.add_edge(1, 2, source=1, target=2, resources={'bw': 10000})
_topology = Topology(u'NoName', _topology)
_predicatedict = {
'null': null_predicate,
'null_predicate': null_predicate,
}
def assign_to_tc(tcs, paths, name):
"""
Assign paths to traffic classes based on the ingress & egress nodes.
To be used only if there is one traffic class (predicate)
per ingress-egress pair.
:param tcs: list of traffic classes
:param paths: dictionary of paths: ..
paths[ingress][egress] = list_of_paths
:return: paths per traffic class
from sol.opt.app import App
app = Flask(__name__)
logger = logging.getLogger()
logger.addHandler(logging.StreamHandler())
# REST-specific configuration here
__API_VERSION = 1 # the current api version
_json_pretty = False # whether to pretty print json or not (not == save space)
_gzip = True # whether to gzip the returned responses
_topology = None
_predicatedict = {
'null': null_predicate,
'null_predicate': null_predicate,
}
def assign_to_tc(tcs, paths, name):
"""
Assign paths to traffic classes based on the ingress & egress nodes.
To be used only if there is one traffic class (predicate)
per ingress-egress pair.
:param tcs: list of traffic classes
:param paths: dictionary of paths: ..
paths[ingress][egress] = list_of_paths
:return: paths per traffic class