Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _parameters(self):
g_phy = self.anm['phy']
settings = autonetkit.config.settings
to_memory = settings['Compiler']['Cisco']['to memory']
use_mgmt_interfaces = g_phy.data.mgmt_interfaces_enabled
now = datetime.now()
if settings['Compiler']['Cisco']['timestamp']:
timestamp = now.strftime("%Y%m%d_%H%M%S_%f")
dst_folder = os.path.join(
"rendered", self.host, timestamp, "cisco")
else:
dst_folder = os.path.join("rendered", self.host, "cisco")
# TODO: use a namedtuple
return to_memory, use_mgmt_interfaces, dst_folder
# set defaults
if not g_in.data.specified_int_names:
# if not specified then automatically assign interface names
g_in.data.specified_int_names = False
#import autonetkit.plugins.graph_product as graph_product
# graph_product.expand(g_in) # apply graph products if relevant
expand_fqdn = False
# TODO: make this set from config and also in the input file
if expand_fqdn and len(ank_utils.unique_attr(g_in, "asn")) > 1:
# Multiple ASNs set, use label format device.asn
anm.set_node_label(".", ['label', 'asn'])
g_in.update(g_in.routers(platform="junosphere"), syntax="junos")
if autonetkit.config.settings['Graphml']['Node Defaults']['syntax'] == 'ios':
g_in.update(g_in.routers(platform="dynagen"), syntax="ios")
elif autonetkit.config.settings['Graphml']['Node Defaults']['syntax'] == 'nx_os':
g_in.update(g_in.routers(platform="dynagen"), syntax="nx_os")
g_in.update(g_in.routers(platform="netkit"), syntax="quagga")
# TODO: is this used?
g_in.update(g_in.servers(platform="netkit"), syntax="quagga")
#TODO: check this is needed
#autonetkit.ank.set_node_default(g_in, specified_int_names=None)
g_graphics = anm.add_overlay("graphics") # plotting data
g_graphics.add_nodes_from(g_in, retain=['x', 'y', 'device_type',
'label', 'device_subtype', 'asn'])
return anm
"""Module to build overlay graphs for network design"""
import autonetkit
import autonetkit.ank as ank_utils
import autonetkit.anm
import autonetkit.config
import autonetkit.exception
import autonetkit.log as log
import networkx as nx
SETTINGS = autonetkit.config.settings
# TODO: revisit phy_neighbors for eg ASN and use layer3 instead
__all__ = ['build']
def load(input_graph_string, defaults=True):
# TODO: look at XML header for file type
import autonetkit.load.graphml as graphml
import autonetkit.load.load_json as load_json
try:
input_graph = graphml.load_graphml(
input_graph_string, defaults=defaults)
except autonetkit.exception.AnkIncorrectFileFormat:
input_graph = load_json.load_json(
def __init__(self, host = None):
if use_http_post:
host = config.settings['Http Post']['server']
port = config.settings['Http Post']['port']
self.http_url = "http://%s:%s/publish" % (host, port)
self.publish = self.publish_http_post
self.publish_compressed = self.publish_http_post
class CustomAdapter(logging.LoggerAdapter):
def process(self, msg, kwargs):
return '[%s]: %s' % (self.extra['item'], msg), kwargs
ank_logger = logging.getLogger("ANK")
if not ank_logger.handlers:
console_formatter = logging.Formatter("%(levelname)-1s %(message)s")
ch = logging.StreamHandler()
#ch.setLevel(logging.INFO)
ch.setFormatter(console_formatter)
ch.setLevel(logging.DEBUG)
ank_logger.addHandler(ch)
file_logging = config.settings['Logging']['file']
if file_logging:
LOG_FILENAME = "autonetkit.log"
#fh = logging.FileHandler(LOG_FILENAME)
LOG_SIZE = 2097152 # 2 MB
fh = logging.handlers.RotatingFileHandler(
LOG_FILENAME, maxBytes=LOG_SIZE, backupCount=5)
fh.setLevel(logging.DEBUG)
formatter = logging.Formatter("%(asctime)s %(levelname)s "
"%(funcName)s %(message)s")
fh.setFormatter(formatter)
ank_logger.addHandler(fh)
ank_logger.setLevel(logging.INFO)
# Reference for external access
logger = ank_logger
# Use approach of Pika, allows for autonetkit.log.debug("message")
import autonetkit.ank
import itertools
import netaddr
import os
import pprint
from collections import defaultdict
import string
from datetime import datetime
import autonetkit.log as log
import autonetkit.plugins.naming as naming
import autonetkit.config
settings = autonetkit.config.settings
from autonetkit.ank_utils import alphabetical_sort as alpha_sort
#TODO: rename compiler to build
#TODO: tidy up the dict to list, and sorting formats
#TODO: don't pass lists/dictionaries around: set directly, and then sort in-place later if needed
def dot_to_underscore(instring):
return instring.replace(".", "_")
class RouterCompiler(object):
lo_interface = "lo0" #make this clear distinction between interface id and lo IP
# and set per platform
"""Base Router compiler"""
def __init__(self, nidb, anm):
# set defaults
if not g_in.data.specified_int_names:
# if not specified then automatically assign interface names
g_in.data.specified_int_names = False
#import autonetkit.plugins.graph_product as graph_product
# graph_product.expand(g_in) # apply graph products if relevant
expand_fqdn = False
# TODO: make this set from config and also in the input file
if expand_fqdn and len(ank_utils.unique_attr(g_in, "asn")) > 1:
# Multiple ASNs set, use label format device.asn
anm.set_node_label(".", ['label', 'asn'])
g_in.update(g_in.routers(platform="junosphere"), syntax="junos")
if autonetkit.config.settings['Graphml']['Node Defaults']['syntax'] == 'ios':
g_in.update(g_in.routers(platform="dynagen"), syntax="ios")
elif autonetkit.config.settings['Graphml']['Node Defaults']['syntax'] == 'nx_os':
g_in.update(g_in.routers(platform="dynagen"), syntax="nx_os")
g_in.update(g_in.routers(platform="netkit"), syntax="quagga")
# TODO: is this used?
g_in.update(g_in.servers(platform="netkit"), syntax="quagga")
#TODO: check this is needed
#autonetkit.ank.set_node_default(g_in, specified_int_names=None)
g_graphics = anm.add_overlay("graphics") # plotting data
g_graphics.add_nodes_from(g_in, retain=['x', 'y', 'device_type',
'label', 'device_subtype', 'asn'])
return anm
# if not specified then automatically assign interface names
g_in.data.specified_int_names = False
#import autonetkit.plugins.graph_product as graph_product
# graph_product.expand(g_in) # apply graph products if relevant
expand_fqdn = False
# TODO: make this set from config and also in the input file
if expand_fqdn and len(ank_utils.unique_attr(g_in, "asn")) > 1:
# Multiple ASNs set, use label format device.asn
anm.set_node_label(".", ['label', 'asn'])
g_in.update(g_in.routers(platform="junosphere"), syntax="junos")
if autonetkit.config.settings['Graphml']['Node Defaults']['syntax'] == 'ios':
g_in.update(g_in.routers(platform="dynagen"), syntax="ios")
elif autonetkit.config.settings['Graphml']['Node Defaults']['syntax'] == 'nx_os':
g_in.update(g_in.routers(platform="dynagen"), syntax="nx_os")
g_in.update(g_in.routers(platform="netkit"), syntax="quagga")
# TODO: is this used?
g_in.update(g_in.servers(platform="netkit"), syntax="quagga")
#TODO: check this is needed
#autonetkit.ank.set_node_default(g_in, specified_int_names=None)
g_graphics = anm.add_overlay("graphics") # plotting data
g_graphics.add_nodes_from(g_in, retain=['x', 'y', 'device_type',
'label', 'device_subtype', 'asn'])
return anm