Closes #151: Implement interface end to end connection view (#188)

* Added end to end interface connections, made showing cables and wireless optional

* Fixed 'Hide Unconnected' and other bugs

* rename var + extra check

* Add filter selection image for readme

* Added guidance on using the selection options

* Renaming to selection_options

* Renamed filter_selection to selection_options

* Update README.md

* Bug fix

* fix parameter order

* differentiate line type for interface connections

* allow both cables and logical connections

* Change logical connection line type

Co-authored-by: Mario <github@franzbonenkamp.de>

* Hide redundant connections, configurable options

* updated readme + removed visjs setting from settings

* Update README.md

Co-authored-by: Mario <github@franzbonenkamp.de>

* updated readme for 'Show Logical Connections'

Co-authored-by: tagrusendorf <tagrusendorf@dev-netbox.tag.kbproductions.ca>
Co-authored-by: mattieserver <3049868+mattieserver@users.noreply.github.com>
Co-authored-by: Mario <github@franzbonenkamp.de>
Co-authored-by: Mattijs Vanhaverbeke <mattijs.vanhaverbeke@ebo-enterprises.com>
This commit is contained in:
tgrusendorf
2022-12-27 14:49:34 +01:00
committed by GitHub
co-authored by Mario tagrusendorf mattieserver Mattijs Vanhaverbeke
parent e0bd1d4e86
commit dfb0b532b9
5 changed files with 225 additions and 113 deletions
+1
View File
@@ -29,6 +29,7 @@ class TopologyViewsConfig(PluginConfig):
"always_save_coordinates": False,
"preselected_tags": [],
"draw_default_layout": False,
"hide_single_cable_logical_conns": False,
}
+12
View File
@@ -34,8 +34,11 @@ class DeviceFilterForm(TenancyFilterForm, NetBoxModelFilterSetForm):
"filter_id",
"hide_unconnected",
"save_coords",
"show_cables",
"show_circuit",
"show_logical_connections",
"show_power",
"show_wireless",
),
),
(
@@ -104,6 +107,15 @@ class DeviceFilterForm(TenancyFilterForm, NetBoxModelFilterSetForm):
hide_unconnected = forms.BooleanField(
label=_("Hide Unconnected"), required=False, initial=False
)
show_logical_connections = forms.BooleanField(
label =_("Show Logical Connections"), required=False, initial=False
)
show_cables = forms.BooleanField(
label =_("Show Cables"), required=False, initial=False
)
show_wireless = forms.BooleanField(
label =_("Show Wireless Links"), required=False, initial=False
)
show_circuit = forms.BooleanField(
label=_("Show Circuit Terminations"), required=False, initial=False
)
+184 -113
View File
@@ -9,6 +9,7 @@ from dcim.models import (
Cable,
CableTermination,
Device,
device_components,
DeviceRole,
FrontPort,
Interface,
@@ -177,6 +178,7 @@ def create_edge(
cable: Optional[Cable] = None,
wireless: Optional[Dict] = None,
power: Optional[bool] = None,
interface: Optional[bool] = None,
):
cable_a_name = (
"device A name unknown"
@@ -217,6 +219,13 @@ def create_edge(
edge["dashes"] = [5, 5, 3, 3]
title = "Power Connection"
elif interface is not None:
title = "Interface Connection"
edge["width"] = 3
edge["dashes"] = [1, 10, 1, 10]
edge["arrows"] = {"to": {"enabled": True, "scaleFactor": 0.5}, "from": {"enabled": True, "scaleFactor": 0.5}}
edge["color"] = '#f1c232'
edge[
"title"
] = f"{title} between<br>{cable_a_dev_name} [{cable_a_name}]<br>{cable_b_dev_name} [{cable_b_name}]"
@@ -251,8 +260,11 @@ def get_topology_data(
queryset: QuerySet,
hide_unconnected: bool,
save_coords: bool,
show_cables: bool,
show_circuit: bool,
show_logical_connections: bool,
show_power: bool,
show_wireless: bool,
):
if not queryset:
return None
@@ -266,9 +278,15 @@ def get_topology_data(
nodes_powerfeed: Dict[int, PowerFeed] = {}
nodes_provider_networks = {}
cable_ids = DefaultDict(dict)
interface_ids = DefaultDict(dict)
ignore_cable_type = settings.PLUGINS_CONFIG["netbox_topology_views"][
"ignore_cable_type"
]
hide_single_cable_logical_conns = bool(
settings.PLUGINS_CONFIG["netbox_topology_views"][
"hide_single_cable_logical_conns"
]
)
device_ids = [d.pk for d in queryset]
site_ids = [d.site_id for d in queryset]
@@ -345,15 +363,6 @@ def get_topology_data(
for d in nodes_circuits.values():
nodes.append(create_node(d, save_coords))
links: QuerySet[CableTermination] = CableTermination.objects.filter(
Q(_device_id__in=device_ids)
).select_related("termination_type")
wlan_links: QuerySet[WirelessLink] = WirelessLink.objects.filter(
Q(_interface_a_device_id__in=device_ids)
& Q(_interface_b_device_id__in=device_ids)
)
if show_power:
power_panels_ids = PowerPanel.objects.filter(
Q(site_id__in=site_ids)
@@ -407,112 +416,153 @@ def get_topology_data(
for d in nodes_powerpanel.values():
nodes.append(create_node(d, save_coords))
for link in links:
if link.termination_type.name in ignore_cable_type:
continue
# Normal device cables
if link.termination_type.name in supported_termination_types:
complete_link = False
if link.cable_end == "A":
if link.cable_id not in cable_ids:
cable_ids[link.cable_id] = {}
else:
if "B" in cable_ids[link.cable_id]:
if cable_ids[link.cable_id]["B"] is not None:
complete_link = True
elif link.cable_end == "B":
if link.cable_id not in cable_ids:
cable_ids[link.cable_id] = {}
else:
if "A" in cable_ids[link.cable_id]:
if cable_ids[link.cable_id]["A"] is not None:
complete_link = True
else:
print("Unkown cable end")
cable_ids[link.cable_id][link.cable_end] = link
if complete_link:
edge_ids += 1
if isinstance(cable_ids[link.cable_id]["B"], CableTermination):
if cable_ids[link.cable_id]["B"]._device_id not in nodes_devices:
nodes_devices[
cable_ids[link.cable_id]["B"]._device_id
] = cable_ids[link.cable_id]["B"].termination.device
termination_b = {
"termination_name": cable_ids[link.cable_id][
"B"
].termination.name,
"termination_device_name": cable_ids[link.cable_id][
"B"
].termination.device.name,
"device_id": cable_ids[link.cable_id][
"B"
].termination.device_id,
}
else:
termination_b = cable_ids[link.cable_id]["B"]
if isinstance(cable_ids[link.cable_id]["A"], CableTermination):
if cable_ids[link.cable_id]["A"]._device_id not in nodes_devices:
nodes_devices[
cable_ids[link.cable_id]["A"]._device_id
] = cable_ids[link.cable_id]["A"].termination.device
termination_a = {
"termination_name": cable_ids[link.cable_id][
"A"
].termination.name,
"termination_device_name": cable_ids[link.cable_id][
"A"
].termination.device.name,
"device_id": cable_ids[link.cable_id][
"A"
].termination.device_id,
}
else:
termination_a = cable_ids[link.cable_id]["A"]
edges.append(
create_edge(
edge_id=edge_ids,
cable=link.cable,
termination_a=termination_a,
termination_b=termination_b,
)
)
for wlan_link in wlan_links:
if wlan_link.interface_a.device_id not in nodes_devices:
nodes_devices[
wlan_link.interface_a.device.pk
] = wlan_link.interface_a.device
if wlan_link.interface_b.device_id not in nodes_devices:
nodes_devices[
wlan_link.interface_b.device.pk
] = wlan_link.interface_b.device
termination_a = {
"termination_name": wlan_link.interface_a.name,
"termination_device_name": wlan_link.interface_a.device.name,
"device_id": wlan_link.interface_a.device_id,
}
termination_b = {
"termination_name": wlan_link.interface_b.name,
"termination_device_name": wlan_link.interface_b.device.name,
"device_id": wlan_link.interface_b.device_id,
}
wireless = {"ssid": wlan_link.ssid}
edge_ids += 1
edges.append(
create_edge(
edge_id=edge_ids,
termination_a=termination_a,
termination_b=termination_b,
wireless=wireless,
)
if show_logical_connections:
interfaces = Interface.objects.filter(
Q(_path__is_complete=True) & Q(device_id__in=device_ids)
)
for interface in interfaces:
# print('{} {} {} {}'.format(interface.device.name, interface.name, interface._path.destinations[0].device.name, interface._path.destinations[0].name))
for destination in interface._path.destinations:
if isinstance(destination, device_components.Interface):
if destination.device.id not in device_ids:
# print('Destination interface not in device queryset, ignoring')
continue
if destination.id in interface_ids:
# we've already captured the destination interface, ignore this connection
# print('Destination interface already exists, ignoring')
continue
if hide_single_cable_logical_conns and interface.cable_id==destination.cable_id and show_cables:
# interface connection is the same as the cable connection, ignore this connection
continue
interface_ids[interface.id]=interface
edge_ids += 1
termination_a = { "termination_name": interface.name, "termination_device_name": interface.device.name, "device_id": interface.device.id }
termination_b = { "termination_name": destination.name, "termination_device_name": destination.device.name, "device_id": destination.device.id }
edges.append(create_edge(edge_id=edge_ids, termination_a=termination_a, termination_b=termination_b, interface=True))
nodes_devices[interface.device.id] = interface.device
nodes_devices[destination.device.id] = destination.device
if show_cables:
links: QuerySet[CableTermination] = CableTermination.objects.filter(
Q(_device_id__in=device_ids)
).select_related("termination_type")
for link in links:
if link.termination_type.name in ignore_cable_type:
continue
# Normal device cables
if link.termination_type.name in supported_termination_types:
complete_link = False
if link.cable_end == "A":
if link.cable_id not in cable_ids:
cable_ids[link.cable_id] = {}
else:
if "B" in cable_ids[link.cable_id]:
if cable_ids[link.cable_id]["B"] is not None:
complete_link = True
elif link.cable_end == "B":
if link.cable_id not in cable_ids:
cable_ids[link.cable_id] = {}
else:
if "A" in cable_ids[link.cable_id]:
if cable_ids[link.cable_id]["A"] is not None:
complete_link = True
else:
print("Unkown cable end")
cable_ids[link.cable_id][link.cable_end] = link
if complete_link:
edge_ids += 1
if isinstance(cable_ids[link.cable_id]["B"], CableTermination):
if cable_ids[link.cable_id]["B"]._device_id not in nodes_devices:
nodes_devices[
cable_ids[link.cable_id]["B"]._device_id
] = cable_ids[link.cable_id]["B"].termination.device
termination_b = {
"termination_name": cable_ids[link.cable_id][
"B"
].termination.name,
"termination_device_name": cable_ids[link.cable_id][
"B"
].termination.device.name,
"device_id": cable_ids[link.cable_id][
"B"
].termination.device_id,
}
else:
termination_b = cable_ids[link.cable_id]["B"]
if isinstance(cable_ids[link.cable_id]["A"], CableTermination):
if cable_ids[link.cable_id]["A"]._device_id not in nodes_devices:
nodes_devices[
cable_ids[link.cable_id]["A"]._device_id
] = cable_ids[link.cable_id]["A"].termination.device
termination_a = {
"termination_name": cable_ids[link.cable_id][
"A"
].termination.name,
"termination_device_name": cable_ids[link.cable_id][
"A"
].termination.device.name,
"device_id": cable_ids[link.cable_id][
"A"
].termination.device_id,
}
else:
termination_a = cable_ids[link.cable_id]["A"]
edges.append(
create_edge(
edge_id=edge_ids,
cable=link.cable,
termination_a=termination_a,
termination_b=termination_b,
)
)
if show_wireless:
wlan_links: QuerySet[WirelessLink] = WirelessLink.objects.filter(
Q(_interface_a_device_id__in=device_ids)
& Q(_interface_b_device_id__in=device_ids)
)
for wlan_link in wlan_links:
if wlan_link.interface_a.device_id not in nodes_devices:
nodes_devices[
wlan_link.interface_a.device.pk
] = wlan_link.interface_a.device
if wlan_link.interface_b.device_id not in nodes_devices:
nodes_devices[
wlan_link.interface_b.device.pk
] = wlan_link.interface_b.device
termination_a = {
"termination_name": wlan_link.interface_a.name,
"termination_device_name": wlan_link.interface_a.device.name,
"device_id": wlan_link.interface_a.device_id,
}
termination_b = {
"termination_name": wlan_link.interface_b.name,
"termination_device_name": wlan_link.interface_b.device.name,
"device_id": wlan_link.interface_b.device_id,
}
wireless = {"ssid": wlan_link.ssid}
edge_ids += 1
edges.append(
create_edge(
edge_id=edge_ids,
termination_a=termination_a,
termination_b=termination_b,
wireless=wireless,
)
)
for qs_device in queryset:
if qs_device.pk not in nodes_devices and not hide_unconnected:
nodes_devices[qs_device.pk] = qs_device
@@ -564,22 +614,43 @@ class TopologyHomeView(PermissionRequiredMixin, View):
if request.GET["show_circuit"] == "on":
show_circuit = True
show_logical_connections = False
if "show_logical_connections" in request.GET:
if request.GET["show_logical_connections"] == "on" :
show_logical_connections = True
show_cables = False
if "show_cables" in request.GET:
if request.GET["show_cables"] == "on" :
show_cables = True
show_wireless = False
if "show_wireless" in request.GET:
if request.GET["show_wireless"] == "on" :
show_wireless = True
if "draw_init" in request.GET:
if request.GET["draw_init"].lower() == "true":
topo_data = get_topology_data(
self.queryset,
hide_unconnected,
save_coords,
show_cables,
show_circuit,
show_logical_connections,
show_power,
show_wireless,
)
else:
topo_data = get_topology_data(
self.queryset,
hide_unconnected,
save_coords,
show_cables,
show_circuit,
show_logical_connections,
show_power,
show_wireless,
)
else:
preselected_device_roles = settings.PLUGINS_CONFIG["netbox_topology_views"][