Johannes Kreuzer
7f0dcd97f7
When a cable has a color set, use this color for the edge which repesents the cable, so that a cabling type or maybe a purpose of the cable could be seen in the topology graph.
199 lines
8.5 KiB
Python
199 lines
8.5 KiB
Python
from rest_framework.viewsets import ModelViewSet, ViewSet, ReadOnlyModelViewSet, GenericViewSet
|
|
from rest_framework.decorators import action
|
|
from rest_framework.response import Response
|
|
from django.contrib.contenttypes.models import ContentType
|
|
|
|
from .serializers import PreDeviceRoleSerializer, TopologyDummySerializer, PreTagSerializer
|
|
from django.conf import settings
|
|
|
|
from utilities.api import IsAuthenticatedOrLoginNotRequired
|
|
|
|
from dcim.models import DeviceRole, Device, Cable
|
|
from extras.models import Tag, CustomField, CustomFieldValue
|
|
|
|
ignore_cable_type_raw = settings.PLUGINS_CONFIG["netbox_topology_views"]["ignore_cable_type"]
|
|
ignore_cable_type = ignore_cable_type_raw.split(",")
|
|
|
|
preselected_device_roles_raw = settings.PLUGINS_CONFIG["netbox_topology_views"]["preselected_device_roles"]
|
|
preselected_device_roles = preselected_device_roles_raw.split(",")
|
|
|
|
preselected_tags_raw = settings.PLUGINS_CONFIG["netbox_topology_views"]["preselected_tags"]
|
|
preselected_tags = preselected_tags_raw.split(",")
|
|
|
|
|
|
class PreSelectDeviceRolesViewSet(ReadOnlyModelViewSet):
|
|
queryset = DeviceRole.objects.filter(name__in=preselected_device_roles)
|
|
serializer_class = PreDeviceRoleSerializer
|
|
|
|
class PreSelectTagsViewSet(ReadOnlyModelViewSet):
|
|
queryset = Tag.objects.filter(name__in=preselected_tags)
|
|
serializer_class = PreTagSerializer
|
|
|
|
class SaveCoordsViewSet(GenericViewSet):
|
|
queryset = Device.objects.all()
|
|
serializer_class = TopologyDummySerializer
|
|
|
|
@action(detail=False, methods=['post'])
|
|
def save_coords(self, request):
|
|
results = {}
|
|
if settings.PLUGINS_CONFIG["netbox_topology_views"]["allow_coordinates_saving"]:
|
|
try:
|
|
cfCoords = CustomField.objects.get(name='coordinates')
|
|
except CustomField.DoesNotExist:
|
|
results["status"] = "coords custom field not created"
|
|
return Response(status=500)
|
|
|
|
obj_type = ContentType.objects.get_for_model(Device)
|
|
|
|
device_id = None
|
|
x_coord = None
|
|
y_coord = None
|
|
if "node_id" in request.data:
|
|
if request.data["node_id"]:
|
|
device_id = request.data["node_id"]
|
|
if "x" in request.data:
|
|
if request.data["x"]:
|
|
x_coord = request.data["x"]
|
|
if "y" in request.data:
|
|
if request.data["y"]:
|
|
y_coord = request.data["y"]
|
|
|
|
actual_device= Device.objects.get(id=device_id)
|
|
try:
|
|
cfvCoords = CustomFieldValue.objects.get(obj_type=obj_type, obj_id=actual_device.pk, field__name='coordinates')
|
|
cfvCoords.value = "%s;%s" % (x_coord,y_coord)
|
|
cfvCoords.save()
|
|
results["status"] = "coords set"
|
|
except CustomFieldValue.DoesNotExist:
|
|
cfvCoords = CustomFieldValue(field=cfCoords, obj_type=obj_type, obj_id=actual_device.id)
|
|
cfvCoords.value = "%s;%s" % (x_coord,y_coord)
|
|
cfvCoords.save()
|
|
results["status"] = "coords set for first time"
|
|
|
|
return Response(results)
|
|
else:
|
|
results["status"] = "not allowed to save coords"
|
|
return Response(results, status=500)
|
|
|
|
class SearchViewSet(GenericViewSet):
|
|
#_ignore_model_permissions = True
|
|
#permission_classes = [IsAuthenticatedOrLoginNotRequired]
|
|
|
|
queryset = Device.objects.all()
|
|
serializer_class = TopologyDummySerializer
|
|
|
|
def _filter(self, site, role, name, tags):
|
|
filter_devices = Device.objects.all()
|
|
if name is not None:
|
|
filter_devices = filter_devices.filter(name__contains=name)
|
|
if site is not None:
|
|
filter_devices = filter_devices.filter(site__id__in=site)
|
|
if role is not None:
|
|
filter_devices = filter_devices.filter(device_role__id__in=role)
|
|
if tags is not None:
|
|
filter_devices = filter_devices.filter(tags__id__in=tags)
|
|
return filter_devices
|
|
|
|
|
|
@action(detail=False, methods=['post'])
|
|
def search(self, request):
|
|
name = None
|
|
sites = None
|
|
devicerole = None
|
|
tags = None
|
|
if "devicerole" in request.data:
|
|
if request.data["devicerole"]:
|
|
devicerole = request.data["devicerole"]
|
|
if "sites" in request.data:
|
|
if request.data["sites"]:
|
|
sites = request.data["sites"]
|
|
if "name" in request.data:
|
|
if request.data["name"]:
|
|
name = request.data["name"]
|
|
if "tags" in request.data:
|
|
if request.data["tags"]:
|
|
tags = request.data["tags"]
|
|
|
|
devices = self._filter(sites, devicerole, name, tags)
|
|
|
|
nodes = []
|
|
edges = []
|
|
edge_ids = 0
|
|
cable_ids = []
|
|
for device in devices:
|
|
cables = device.get_cables()
|
|
for cable in cables:
|
|
if cable.termination_a_type.name != "circuit termination" and cable.termination_b_type.name != "circuit termination":
|
|
if cable.id not in cable_ids:
|
|
if cable.termination_a_type.name not in ignore_cable_type and cable.termination_b_type.name not in ignore_cable_type:
|
|
cable_ids.append(cable.id)
|
|
edge_ids += 1
|
|
|
|
cable_a_dev_name = cable.termination_a.device.name
|
|
if cable_a_dev_name is None:
|
|
cable_a_dev_name = "device A name unknown"
|
|
cable_a_name = cable.termination_a.name
|
|
if cable_a_name is None:
|
|
cable_a_name = "cable A name unknown"
|
|
cable_b_dev_name = cable.termination_b.device.name
|
|
if cable_b_dev_name is None:
|
|
cable_b_dev_name = "device B name unknown"
|
|
cable_b_name = cable.termination_b.name
|
|
if cable_b_name is None:
|
|
cable_b_name = "cable B name unknown"
|
|
|
|
edge = {}
|
|
edge["id"] = edge_ids
|
|
edge["from"] = cable.termination_a.device.id
|
|
edge["to"] = cable.termination_b.device.id
|
|
edge["title"] = "Connection between <br> " + cable_a_dev_name + " [" + cable_a_name + "]<br>" + cable_b_dev_name + " [" + cable_b_name + "]"
|
|
if cable.color != "":
|
|
edge["color"] = "#" + cable.color
|
|
edges.append(edge)
|
|
else:
|
|
pass
|
|
#circuittermination not supported for now
|
|
|
|
dev_name = device.name
|
|
if dev_name is None:
|
|
dev_name = "device name unknown"
|
|
|
|
node_content = ""
|
|
|
|
if device.device_type.display_name is not None:
|
|
node_content += "<tr><th>Type: </th><td>" + device.device_type.display_name + "</td></tr>"
|
|
if device.device_role.name is not None:
|
|
node_content += "<tr><th>Role: </th><td>" + device.device_role.name + "</td></tr>"
|
|
if device.serial != "":
|
|
node_content += "<tr><th>Serial: </th><td>" + device.serial + "</td></tr>"
|
|
if device.primary_ip is not None:
|
|
node_content += "<tr><th>IP Address: </th><td>" + str(device.primary_ip.address) + "</td></tr>"
|
|
|
|
dev_title = "<table> %s </table>" % (node_content)
|
|
|
|
node = {}
|
|
node["id"] = device.id
|
|
node["name"] = dev_name
|
|
node["label"] = dev_name
|
|
node["title"] = dev_title
|
|
node["shape"] = 'image'
|
|
if device.device_role.slug in settings.PLUGINS_CONFIG["netbox_topology_views"]["device_img"]:
|
|
node["image"] = '../../static/netbox_topology_views/img/' + device.device_role.slug + ".png"
|
|
else:
|
|
node["image"] = "../../static/netbox_topology_views/img/role-unknown.png"
|
|
|
|
for device_custom_field in device.custom_field_values.all():
|
|
if device_custom_field.field.name == "coordinates":
|
|
cords = device_custom_field.serialized_value.split(";")
|
|
node["x"] = int(cords[0])
|
|
node["y"] = int(cords[1])
|
|
node["physics"] = False
|
|
|
|
nodes.append(node)
|
|
|
|
results = {}
|
|
results["nodes"] = nodes
|
|
results["edges"] = edges
|
|
|
|
return Response(results)
|