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
" + cable_a_dev_name + " [" + cable_a_name + "]
" + cable_b_dev_name + " [" + cable_b_name + "]" edges.append(edge) else: pass #circuittermination not supported for now dev_name = device.name if dev_name is None: dev_name = "device name unknown" cable_role_name = device.device_type.display_name if cable_role_name is None: cable_role_name = "device role name unknown" dev_title = "
Type:" + cable_role_name + "
Role:" + device.device_role.name if device.serial != "": dev_title = dev_title + "
Serial:" + device.serial if device.primary_ip is not None: dev_title = dev_title + "
IP Address:" + str(device.primary_ip.address) dev_title = dev_title + "
" 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)