Coverage for dcim/models/cables.py: 18%
670 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1import itertools
2import logging
3import threading
4from collections import Counter
6from django.contrib.contenttypes.fields import GenericForeignKey
7from django.contrib.contenttypes.models import ContentType
8from django.contrib.postgres.fields import ArrayField
9from django.contrib.postgres.indexes import GinIndex
10from django.core.exceptions import ValidationError
11from django.core.validators import MaxValueValidator, MinValueValidator
12from django.db import models, router, transaction
13from django.dispatch import Signal
14from django.urls import reverse
15from django.utils.translation import gettext_lazy as _
17from core.models import ObjectType
18from dcim.choices import *
19from dcim.constants import *
20from dcim.exceptions import UnsupportedCablePath
21from dcim.fields import PathField
22from dcim.utils import decompile_path_node, object_to_path_node, rebuild_cable_paths
23from netbox.choices import ColorChoices
24from netbox.models import ChangeLoggedModel, PrimaryModel
25from utilities.conversion import to_meters
26from utilities.data import normalize_update_fields
27from utilities.exceptions import AbortRequest
28from utilities.fields import ColorField, GenericArrayForeignKey
29from utilities.querysets import RestrictedQuerySet, chunked_update
30from utilities.serialization import deserialize_object, serialize_object
31from wireless.models import WirelessLink
33from .device_components import FrontPort, Interface, PathEndpoint, PortMapping, RearPort
35__all__ = (
36 'Cable',
37 'CableBundle',
38 'CablePath',
39 'CableTermination',
40)
42logger = logging.getLogger(f'netbox.{__name__}')
44trace_paths = Signal()
47#
48# Cable bundles
49#
51class CableBundle(PrimaryModel):
52 """
53 A logical grouping of individual cables.
54 """
55 name = models.CharField(
56 verbose_name=_('name'),
57 max_length=100,
58 unique=True,
59 )
61 class Meta:
62 ordering = ('name',)
63 verbose_name = _('cable bundle')
64 verbose_name_plural = _('cable bundles')
66 def __str__(self):
67 return self.name
69 def get_absolute_url(self):
70 return reverse('dcim:cablebundle', args=[self.pk])
73#
74# Cables
75#
77class CableQuerySet(RestrictedQuerySet):
79 def delete(self):
80 # Track these Cables as being deleted for the duration, as Cable.delete() does for a single
81 # instance: a queryset delete never calls it. Between them the two cover every deletion, as a
82 # Cable is never itself cascade-deleted (nothing points at it with on_delete=CASCADE).
83 # Resolve the PKs on the DB the delete will use, so read routing can't miss a lagging replica.
84 using = self._db or router.db_for_write(self.model, **self._hints)
85 pks = list(self.using(using).values_list('pk', flat=True))
86 for pk in pks:
87 Cable._track_deletion(pk)
88 try:
89 return super().delete()
90 finally:
91 for pk in pks:
92 Cable._untrack_deletion(pk)
95class Cable(PrimaryModel):
96 """
97 A physical connection between two endpoints.
98 """
99 # Per-thread tracking of Cable PKs currently being deleted; referenced by
100 # dcim.signals.nullify_connected_endpoints to record the disconnect on each terminating object and
101 # to skip per-CableTermination path retracing during the cascade (retrace_cable_paths does it once).
102 _deletion_tracking = threading.local()
104 type = models.CharField(
105 verbose_name=_('type'),
106 max_length=50,
107 choices=CableTypeChoices,
108 blank=True,
109 null=True
110 )
111 status = models.CharField(
112 verbose_name=_('status'),
113 max_length=50,
114 choices=LinkStatusChoices,
115 default=LinkStatusChoices.STATUS_CONNECTED
116 )
117 profile = models.CharField(
118 verbose_name=_('profile'),
119 max_length=50,
120 choices=CableProfileChoices,
121 blank=True,
122 )
123 tenant = models.ForeignKey(
124 to='tenancy.Tenant',
125 on_delete=models.PROTECT,
126 related_name='cables',
127 blank=True,
128 null=True
129 )
130 label = models.CharField(
131 verbose_name=_('label'),
132 max_length=100,
133 blank=True
134 )
135 color = ColorField(
136 verbose_name=_('color'),
137 blank=True
138 )
139 length = models.DecimalField(
140 verbose_name=_('length'),
141 max_digits=8,
142 decimal_places=2,
143 blank=True,
144 null=True
145 )
146 length_unit = models.CharField(
147 verbose_name=_('length unit'),
148 max_length=50,
149 choices=CableLengthUnitChoices,
150 blank=True,
151 null=True
152 )
153 # Stores the normalized length (in meters) for database ordering
154 _abs_length = models.DecimalField(
155 max_digits=14,
156 decimal_places=4,
157 blank=True,
158 null=True
159 )
160 bundle = models.ForeignKey(
161 to='dcim.CableBundle',
162 on_delete=models.SET_NULL,
163 related_name='cables',
164 blank=True,
165 null=True,
166 verbose_name=_('bundle'),
167 )
169 clone_fields = ('tenant', 'type', 'profile', 'bundle')
171 objects = CableQuerySet.as_manager()
173 class Meta:
174 ordering = ('pk',)
175 verbose_name = _('cable')
176 verbose_name_plural = _('cables')
178 def __init__(self, *args, a_terminations=None, b_terminations=None, **kwargs):
179 super().__init__(*args, **kwargs)
181 # A copy of the PK to be used by __str__ in case the object is deleted
182 self._pk = self.__dict__.get('id')
184 # Cache the original profile & status so we can check later whether either has been changed
185 self._orig_status = self.__dict__.get('status')
186 self._orig_profile = self.__dict__.get('profile')
188 self._terminations_modified = False
190 # Assign or retrieve A/B terminations
191 if a_terminations: 191 ↛ 192line 191 didn't jump to line 192 because the condition on line 191 was never true
192 self.a_terminations = a_terminations
193 if b_terminations: 193 ↛ 194line 193 didn't jump to line 194 because the condition on line 193 was never true
194 self.b_terminations = b_terminations
196 def __str__(self):
197 pk = self.pk or self._pk
198 return self.label or f'#{pk}'
200 def get_status_color(self):
201 return LinkStatusChoices.colors.get(self.status)
203 @property
204 def profile_class(self):
205 from dcim import cable_profiles
206 return {
207 CableProfileChoices.SINGLE_1C1P: cable_profiles.Single1C1PCableProfile,
208 CableProfileChoices.SINGLE_1C2P: cable_profiles.Single1C2PCableProfile,
209 CableProfileChoices.SINGLE_1C4P: cable_profiles.Single1C4PCableProfile,
210 CableProfileChoices.SINGLE_1C6P: cable_profiles.Single1C6PCableProfile,
211 CableProfileChoices.SINGLE_1C8P: cable_profiles.Single1C8PCableProfile,
212 CableProfileChoices.SINGLE_1C12P: cable_profiles.Single1C12PCableProfile,
213 CableProfileChoices.SINGLE_1C16P: cable_profiles.Single1C16PCableProfile,
214 CableProfileChoices.TRUNK_2C1P: cable_profiles.Trunk2C1PCableProfile,
215 CableProfileChoices.TRUNK_2C2P: cable_profiles.Trunk2C2PCableProfile,
216 CableProfileChoices.TRUNK_2C4P: cable_profiles.Trunk2C4PCableProfile,
217 CableProfileChoices.TRUNK_2C4P_SHUFFLE: cable_profiles.Trunk2C4PShuffleCableProfile,
218 CableProfileChoices.TRUNK_2C6P: cable_profiles.Trunk2C6PCableProfile,
219 CableProfileChoices.TRUNK_2C8P: cable_profiles.Trunk2C8PCableProfile,
220 CableProfileChoices.TRUNK_2C12P: cable_profiles.Trunk2C12PCableProfile,
221 CableProfileChoices.TRUNK_4C1P: cable_profiles.Trunk4C1PCableProfile,
222 CableProfileChoices.TRUNK_4C2P: cable_profiles.Trunk4C2PCableProfile,
223 CableProfileChoices.TRUNK_4C4P: cable_profiles.Trunk4C4PCableProfile,
224 CableProfileChoices.TRUNK_4C4P_SHUFFLE: cable_profiles.Trunk4C4PShuffleCableProfile,
225 CableProfileChoices.TRUNK_4C6P: cable_profiles.Trunk4C6PCableProfile,
226 CableProfileChoices.TRUNK_4C8P: cable_profiles.Trunk4C8PCableProfile,
227 CableProfileChoices.TRUNK_8C4P: cable_profiles.Trunk8C4PCableProfile,
228 CableProfileChoices.BREAKOUT_1C2P_2C1P: cable_profiles.Breakout1C2Px2C1PCableProfile,
229 CableProfileChoices.BREAKOUT_1C4P_4C1P: cable_profiles.Breakout1C4Px4C1PCableProfile,
230 CableProfileChoices.BREAKOUT_1C6P_6C1P: cable_profiles.Breakout1C6Px6C1PCableProfile,
231 CableProfileChoices.BREAKOUT_1C8P_8C1P: cable_profiles.Breakout1C8Px8C1PCableProfile,
232 CableProfileChoices.BREAKOUT_2C4P_8C1P_SHUFFLE: cable_profiles.Breakout2C4Px8C1PShuffleCableProfile,
233 }.get(self.profile)
235 def _get_x_terminations(self, side):
236 """
237 Return the terminating objects for the given cable end (A or B).
238 """
239 if side not in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B): 239 ↛ 240line 239 didn't jump to line 240 because the condition on line 239 was never true
240 raise ValueError(f"Unknown cable side: {side}")
241 attr = f'_{side.lower()}_terminations'
243 if hasattr(self, attr): 243 ↛ 244line 243 didn't jump to line 244 because the condition on line 243 was never true
244 return getattr(self, attr)
245 if not self.pk: 245 ↛ 247line 245 didn't jump to line 247 because the condition on line 245 was always true
246 return []
247 return [
248 # Query self.terminations.all() to leverage cached results
249 ct.termination for ct in self.terminations.all() if ct.cable_end == side
250 ]
252 def _set_x_terminations(self, side, value):
253 """
254 Set the terminating objects for the given cable end (A or B).
255 """
256 if side not in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B):
257 raise ValueError(f"Unknown cable side: {side}")
258 _attr = f'_{side.lower()}_terminations'
260 # If the provided value is a list of CableTermination IDs, resolve them
261 # to their corresponding termination objects.
262 if all(isinstance(item, int) for item in value):
263 value = [
264 ct.termination for ct in CableTermination.objects.filter(pk__in=value).prefetch_related('termination')
265 ]
267 if not self.pk or getattr(self, _attr, []) != list(value):
268 self._terminations_modified = True
270 setattr(self, _attr, value)
272 @property
273 def a_terminations(self):
274 return self._get_x_terminations(CableEndChoices.SIDE_A)
276 @a_terminations.setter
277 def a_terminations(self, value):
278 self._set_x_terminations(CableEndChoices.SIDE_A, value)
280 @property
281 def b_terminations(self):
282 return self._get_x_terminations(CableEndChoices.SIDE_B)
284 @b_terminations.setter
285 def b_terminations(self, value):
286 self._set_x_terminations(CableEndChoices.SIDE_B, value)
288 @property
289 def color_name(self):
290 color_name = ""
291 for hex_code, label in ColorChoices.CHOICES:
292 if hex_code.lower() == self.color.lower():
293 color_name = str(label)
295 return color_name
297 def clean(self):
298 super().clean()
300 # Validate length and length_unit
301 if self.length is not None and not self.length_unit: 301 ↛ 302line 301 didn't jump to line 302 because the condition on line 301 was never true
302 raise ValidationError(_("Must specify a unit when setting a cable length"))
304 if self._state.adding and self.pk is None and (not self.a_terminations or not self.b_terminations): 304 ↛ 308line 304 didn't jump to line 308 because the condition on line 304 was always true
305 raise ValidationError(_("Must define A and B terminations when creating a new cable."))
307 # Validate terminations against the assigned cable profile (if any)
308 if self.profile:
309 self.profile_class().clean(self)
311 if self._terminations_modified:
313 # Check that all termination objects for either end are of the same type
314 for terms in (self.a_terminations, self.b_terminations):
315 if len(terms) > 1 and not all(isinstance(t, type(terms[0])) for t in terms[1:]):
316 raise ValidationError(_("Cannot connect different termination types to same end of cable."))
318 # Check that termination types are compatible
319 if self.a_terminations and self.b_terminations:
320 a_type = self.a_terminations[0]._meta.model_name
321 b_type = self.b_terminations[0]._meta.model_name
322 if b_type not in COMPATIBLE_TERMINATION_TYPES.get(a_type):
323 raise ValidationError(
324 _("Incompatible termination types: {type_a} and {type_b}").format(type_a=a_type, type_b=b_type)
325 )
326 if a_type == b_type:
327 # can't directly use self.a_terminations here as possible they
328 # don't have pk yet
329 a_pks = set(obj.pk for obj in self.a_terminations if obj.pk)
330 b_pks = set(obj.pk for obj in self.b_terminations if obj.pk)
332 if (a_pks & b_pks):
333 raise ValidationError(
334 _("A and B terminations cannot connect to the same object.")
335 )
337 # Run clean() on any new CableTerminations
338 for termination in self.a_terminations:
339 CableTermination(cable=self, cable_end='A', termination=termination).clean()
340 for termination in self.b_terminations:
341 CableTermination(cable=self, cable_end='B', termination=termination).clean()
343 def save(self, *args, force_insert=False, force_update=False, using=None, update_fields=None):
344 _created = self.pk is None
345 save_kwargs = {
346 'using': using,
347 'update_fields': update_fields,
348 }
349 update_fields = normalize_update_fields(save_kwargs)
351 length_written = update_fields is None or 'length' in update_fields
352 length_unit_written = update_fields is None or 'length_unit' in update_fields
354 if length_written or length_unit_written:
355 if length_written and length_unit_written:
356 stored = {}
357 else:
358 # Read from the database this save will write, so a router cannot split the two
359 db = using or router.db_for_write(Cable, instance=self)
360 stored = Cable.objects.using(db).filter(pk=self.pk).values('length', 'length_unit').first() or {}
361 length = self.length if length_written else stored.get('length')
362 length_unit = self.length_unit if length_unit_written else stored.get('length_unit')
364 # Clear length_unit if no length is defined
365 if length is None and length_unit_written:
366 self.length_unit = None
368 # Store the given length (if any) in meters for use in database ordering
369 if length is not None and length_unit:
370 self._abs_length = to_meters(length, length_unit)
371 else:
372 self._abs_length = None
374 # _abs_length is a denormalized cache of length and length_unit, so persist them together
375 if update_fields is not None:
376 save_kwargs['update_fields'] = update_fields | {'_abs_length'}
378 # A field counts as changed only when this save actually writes it
379 status_written = update_fields is None or 'status' in update_fields
380 profile_written = update_fields is None or 'profile' in update_fields
382 # If this is a new Cable, save it before attempting to create its CableTerminations
383 if self._state.adding:
384 super().save(*args, force_insert=True, **save_kwargs)
385 # Update the private PK used in __str__()
386 self._pk = self.pk
388 if profile_written and self._orig_profile != self.profile:
389 self.update_terminations(force=True)
390 elif self._terminations_modified:
391 self.update_terminations()
393 super().save(*args, force_update=True, **save_kwargs)
395 try:
396 trace_paths.send(Cable, instance=self, created=_created)
397 except UnsupportedCablePath as e:
398 raise AbortRequest(e)
400 # Reset change tracking for the next save of this instance
401 if status_written:
402 self._orig_status = self.status
403 if profile_written:
404 self._orig_profile = self.profile
405 self._terminations_modified = False
407 def delete(self, *args, **kwargs):
408 # Cache the PK locally because super().delete() clears self.pk before the finally block runs; the
409 # finally also guarantees the PK is discarded if the delete raises. CableQuerySet.delete() tracks
410 # the same way for a queryset delete, which never calls this.
411 pk = self.pk
412 Cable._track_deletion(pk)
413 try:
414 return super().delete(*args, **kwargs)
415 finally:
416 Cable._untrack_deletion(pk)
418 @classmethod
419 def _track_deletion(cls, pk):
420 """
421 Track a Cable as being deleted, so that the post_delete handler for its cascaded CableTerminations can
422 record the disconnect on each terminating object and skip redundant path retracing (retrace_cable_paths()
423 retraces each affected path once, after the Cable itself is deleted). The tracking set lives on a
424 threading.local() to isolate concurrent deletions across threads.
425 """
426 if not hasattr(cls._deletion_tracking, 'pks'):
427 cls._deletion_tracking.pks = set()
428 cls._deletion_tracking.pks.add(pk)
430 @classmethod
431 def _untrack_deletion(cls, pk):
432 if hasattr(cls._deletion_tracking, 'pks'):
433 cls._deletion_tracking.pks.discard(pk)
435 @classmethod
436 def _is_being_deleted(cls, pk):
437 return pk in getattr(cls._deletion_tracking, 'pks', ())
439 def clone(self):
440 """
441 Return attributes suitable for cloning this cable.
443 In addition to the fields defined in `clone_fields`, include the termination
444 type and parent selector fields used by dcim.forms.connections.get_cable_form().
445 """
446 attrs = super().clone()
448 # Mirror dcim.forms.connections.get_cable_form() parent-field logic
449 for cable_end, terminations in (('a', self.a_terminations), ('b', self.b_terminations)):
450 if not terminations:
451 continue
453 term_cls = type(terminations[0])
454 term_label = term_cls._meta.label_lower
456 # Matches CableForm choices: "<app_label>.<model>"
457 attrs[f'{cable_end}_terminations_type'] = term_label
459 # Device component
460 if hasattr(term_cls, 'device'):
461 device_ids = sorted({t.device_id for t in terminations if t.device_id})
462 if device_ids:
463 attrs[f'termination_{cable_end}_device'] = device_ids
465 # PowerFeed
466 elif term_label == 'dcim.powerfeed':
467 powerpanel_ids = sorted({t.power_panel_id for t in terminations if t.power_panel_id})
468 if powerpanel_ids:
469 attrs[f'termination_{cable_end}_powerpanel'] = powerpanel_ids
471 # CircuitTermination
472 elif term_label == 'circuits.circuittermination':
473 circuit_ids = sorted({t.circuit_id for t in terminations if t.circuit_id})
474 if circuit_ids:
475 attrs[f'termination_{cable_end}_circuit'] = circuit_ids
477 # Never clone the actual terminations, as they are already occupied
478 attrs.pop('a_terminations', None)
479 attrs.pop('b_terminations', None)
481 return attrs
483 def serialize_object(self, exclude=None):
484 data = serialize_object(self, exclude=exclude or [])
486 # Add A & B terminations to the serialized data
487 a_terminations, b_terminations = self.get_terminations()
488 data['a_terminations'] = sorted([ct.pk for ct in a_terminations.values()])
489 data['b_terminations'] = sorted([ct.pk for ct in b_terminations.values()])
491 return data
493 @classmethod
494 def deserialize_object(cls, data, pk=None):
495 a_terminations = data.pop('a_terminations', [])
496 b_terminations = data.pop('b_terminations', [])
498 instance = deserialize_object(cls, data, pk=pk)
500 # Assign A & B termination objects to the Cable instance
501 queryset = CableTermination.objects.prefetch_related('termination')
502 instance.a_terminations = [
503 ct.termination for ct in queryset.filter(pk__in=a_terminations)
504 ]
505 instance.b_terminations = [
506 ct.termination for ct in queryset.filter(pk__in=b_terminations)
507 ]
509 return instance
511 def update_dependent_objects(self):
512 """
513 Recreate the CablePaths traversing this Cable from its current terminations.
514 """
515 with transaction.atomic(using=router.db_for_write(CablePath)):
517 # Restore channel cable attributes omitted by bulk-update change logging.
518 for ct in CableTermination.objects.filter(cable=self).prefetch_related('termination'):
519 if isinstance(ct.termination, Interface) and ct.termination.channels:
520 ct.termination.propagate_channel_cables()
522 rebuild_cable_paths(self)
523 update_dependent_objects.alters_data = True
525 def get_terminations(self):
526 """
527 Return two dictionaries mapping A & B side terminating objects to their corresponding CableTerminations
528 for this Cable.
529 """
530 a_terminations = {}
531 b_terminations = {}
533 for ct in CableTermination.objects.filter(cable=self).prefetch_related('termination'):
534 if ct.cable_end == CableEndChoices.SIDE_A:
535 a_terminations[ct.termination] = ct
536 else:
537 b_terminations[ct.termination] = ct
539 return a_terminations, b_terminations
541 def _connectors_reassigned(self, existing, terminations):
542 """
543 Return True if any of the given terminating objects already terminates this Cable, but would be
544 assigned to a different connector than the one it currently occupies.
546 Args:
547 existing: Mapping of terminating objects to their current CableTerminations, as returned by
548 get_terminations()
549 terminations: The ordered list of terminating objects to be assigned to this end of the Cable
550 """
551 if not self.profile:
552 # Connectors are assigned only for a Cable which has a profile
553 return False
554 for connector, termination in enumerate(terminations, start=1):
555 if (ct := existing.get(termination)) and ct.connector != connector:
556 return True
557 return False
559 def update_terminations(self, force=False):
560 """
561 Create/delete CableTerminations for this Cable to reflect its current state.
563 Args:
564 force: Force the recreation of all CableTerminations, even if no changes have been made. Needed e.g. when
565 altering a Cable's assigned profile.
566 """
567 a_terminations, b_terminations = self.get_terminations()
569 # A CableTermination's connector is derived from its position within its end's list of terminating
570 # objects, so reordering that list (or removing an object from the middle of it) rewires the Cable
571 # without changing which objects it connects. Recreate the affected end's CableTerminations so that
572 # each is reassigned to its new connector.
573 force_a = force or self._connectors_reassigned(a_terminations, self.a_terminations)
574 force_b = force or self._connectors_reassigned(b_terminations, self.b_terminations)
576 # When force-recreating terminations (e.g. after a profile change), cache the termination objects
577 # from the database before deleting, so they are available for recreation. Without this, the
578 # a_terminations/b_terminations properties would query the DB after deletion and return empty lists.
579 if force_a and not hasattr(self, '_a_terminations'):
580 self._a_terminations = list(a_terminations.keys())
581 if force_b and not hasattr(self, '_b_terminations'):
582 self._b_terminations = list(b_terminations.keys())
584 # Recreating terminations invalidates existing paths, even when the endpoints are unchanged
585 self._terminations_modified = True
587 # Delete any stale CableTerminations
588 for termination, ct in a_terminations.items():
589 if force_a or (termination.pk and termination not in self.a_terminations):
590 ct.delete()
591 for termination, ct in b_terminations.items():
592 if force_b or (termination.pk and termination not in self.b_terminations):
593 ct.delete()
595 # Save any new CableTerminations
596 profile = self.profile_class() if self.profile else None
597 for i, termination in enumerate(self.a_terminations, start=1):
598 if force_a or not termination.pk or termination not in a_terminations:
599 connector = positions = None
600 if profile:
601 connector = i
602 positions = profile.get_position_list(profile.a_connectors[i])
603 CableTermination(
604 cable=self,
605 cable_end=CableEndChoices.SIDE_A,
606 connector=connector,
607 positions=positions,
608 termination=termination
609 ).save()
610 for i, termination in enumerate(self.b_terminations, start=1):
611 if force_b or not termination.pk or termination not in b_terminations:
612 connector = positions = None
613 if profile:
614 connector = i
615 positions = profile.get_position_list(profile.b_connectors[i])
616 CableTermination(
617 cable=self,
618 cable_end=CableEndChoices.SIDE_B,
619 connector=connector,
620 positions=positions,
621 termination=termination
622 ).save()
625class CableTermination(ChangeLoggedModel):
626 """
627 A mapping between side A or B of a Cable and a terminating object (e.g. an Interface or CircuitTermination).
628 """
629 cable = models.ForeignKey(
630 to='dcim.Cable',
631 on_delete=models.CASCADE,
632 related_name='terminations'
633 )
634 cable_end = models.CharField(
635 max_length=1,
636 choices=CableEndChoices,
637 verbose_name=_('end')
638 )
639 termination_type = models.ForeignKey(
640 to='contenttypes.ContentType',
641 on_delete=models.PROTECT,
642 related_name='+'
643 )
644 termination_id = models.PositiveBigIntegerField()
645 termination = GenericForeignKey(
646 ct_field='termination_type',
647 fk_field='termination_id'
648 )
649 connector = models.PositiveSmallIntegerField(
650 blank=True,
651 null=True,
652 validators=(
653 MinValueValidator(CABLE_CONNECTOR_MIN),
654 MaxValueValidator(CABLE_CONNECTOR_MAX)
655 ),
656 )
657 positions = ArrayField(
658 base_field=models.PositiveSmallIntegerField(
659 validators=(
660 MinValueValidator(CABLE_POSITION_MIN),
661 MaxValueValidator(CABLE_POSITION_MAX)
662 )
663 ),
664 blank=True,
665 null=True,
666 )
668 # Cached associations to enable efficient filtering
669 _device = models.ForeignKey(
670 to='dcim.Device',
671 on_delete=models.CASCADE,
672 blank=True,
673 null=True
674 )
675 _rack = models.ForeignKey(
676 to='dcim.Rack',
677 on_delete=models.CASCADE,
678 blank=True,
679 null=True
680 )
681 _location = models.ForeignKey(
682 to='dcim.Location',
683 on_delete=models.CASCADE,
684 blank=True,
685 null=True
686 )
687 _site = models.ForeignKey(
688 to='dcim.Site',
689 on_delete=models.CASCADE,
690 blank=True,
691 null=True
692 )
694 objects = RestrictedQuerySet.as_manager()
696 class Meta:
697 ordering = ('cable', 'cable_end', 'connector', 'pk')
698 constraints = (
699 models.UniqueConstraint(
700 fields=('termination_type', 'termination_id'),
701 name='%(app_label)s_%(class)s_unique_termination'
702 ),
703 models.UniqueConstraint(
704 fields=('cable', 'cable_end', 'connector'),
705 name='%(app_label)s_%(class)s_unique_connector'
706 ),
707 )
708 verbose_name = _('cable termination')
709 verbose_name_plural = _('cable terminations')
711 def __str__(self):
712 return f'Cable {self.cable} to {self.termination}'
714 def clean(self):
715 super().clean()
717 # Disallow connecting a cable to any termination object that is
718 # explicitly flagged as "mark connected".
719 termination = getattr(self, 'termination', None)
720 if termination is not None and getattr(termination, "mark_connected", False):
721 raise ValidationError(
722 _("Cannot connect a cable to {obj_parent} > {obj} because it is marked as connected.").format(
723 obj_parent=termination.parent_object,
724 obj=termination,
725 )
726 )
728 # Check for existing termination
729 qs = CableTermination.objects.filter(
730 termination_type=self.termination_type,
731 termination_id=self.termination_id
732 )
733 if self.cable.pk:
734 qs = qs.exclude(cable=self.cable)
736 existing_termination = qs.first()
737 if existing_termination is not None:
738 raise ValidationError(
739 _("Duplicate termination found for {app_label}.{model} {termination_id}: cable {cable_pk}").format(
740 app_label=self.termination_type.app_label,
741 model=self.termination_type.model,
742 termination_id=self.termination_id,
743 cable_pk=existing_termination.cable.pk
744 )
745 )
746 # A channel subinterface derives its cable from its parent interface and cannot be cabled directly. Checked
747 # ahead of the generic type validation below (channel is a nonconnectable type) to surface the more specific
748 # guidance.
749 if self.termination_type.model == 'interface' and self.termination.channel_id:
750 raise ValidationError(
751 _("Cables cannot be terminated directly to a channel subinterface; cable the parent interface instead.")
752 )
754 # Validate the interface type (if applicable)
755 if self.termination_type.model == 'interface' and self.termination.type in NONCONNECTABLE_IFACE_TYPES:
756 raise ValidationError(
757 _("Cables cannot be terminated to {type_display} interfaces").format(
758 type_display=self.termination.get_type_display()
759 )
760 )
762 # A CircuitTermination attached to a ProviderNetwork cannot have a Cable
763 if self.termination_type.model == 'circuittermination' and self.termination._provider_network is not None:
764 raise ValidationError(_("Circuit terminations attached to a provider network may not be cabled."))
766 def save(self, *args, **kwargs):
768 # Cache objects associated with the terminating object (for filtering)
769 self.cache_related_objects()
771 super().save(*args, **kwargs)
773 # Set the cable on the terminating object
774 termination = self.termination._meta.model.objects.get(pk=self.termination_id)
775 termination.snapshot()
776 termination.set_cable_termination(self)
777 termination.save()
779 def delete(self, *args, **kwargs):
781 # Delete the cable association on the terminating object
782 termination = self.termination._meta.model.objects.get(pk=self.termination_id)
783 termination.snapshot()
784 termination.clear_cable_termination(self)
785 termination.save()
787 super().delete(*args, **kwargs)
789 def cache_related_objects(self):
790 """
791 Cache objects related to the termination (e.g. device, rack, site) directly on the object to
792 enable efficient filtering.
793 """
794 if self.termination is None:
795 raise ValueError(
796 _("Invalid cable termination: the assigned termination object does not exist.")
797 )
799 # Device components
800 if getattr(self.termination, 'device', None):
801 self._device = self.termination.device
802 self._rack = self.termination.device.rack
803 self._location = self.termination.device.location
804 self._site = self.termination.device.site
806 # Power feeds
807 elif getattr(self.termination, 'rack', None):
808 self._rack = self.termination.rack
809 self._location = self.termination.rack.location
810 self._site = self.termination.rack.site
812 # Circuit terminations (which cache their own site/location)
813 elif self.termination._meta.label_lower == 'circuits.circuittermination':
814 self._site = self.termination._site
815 self._location = self.termination._location
816 cache_related_objects.alters_data = True
818 def to_objectchange(self, action):
819 objectchange = super().to_objectchange(action)
820 objectchange.related_object = self.termination
821 return objectchange
824class CablePath(models.Model):
825 """
826 A CablePath instance represents the physical path from a set of origin nodes to a set of destination nodes,
827 including all intermediate elements.
829 `path` contains the ordered set of nodes, arranged in lists of (type, ID) tuples. (Each cable in the path can
830 terminate to one or more objects.) For example, consider the following
831 topology:
833 A B C
834 Interface 1 --- Front Port 1 | Rear Port 1 --- Rear Port 2 | Front Port 3 --- Interface 2
835 Front Port 2 Front Port 4
837 This path would be expressed as:
839 CablePath(
840 path = [
841 [Interface 1],
842 [Cable A],
843 [Front Port 1, Front Port 2],
844 [Rear Port 1],
845 [Cable B],
846 [Rear Port 2],
847 [Front Port 3, Front Port 4],
848 [Cable C],
849 [Interface 2],
850 ]
851 )
853 `is_active` is set to True only if every Cable within the path has a status of "connected". `is_complete` is True
854 if the instance represents a complete end-to-end path from origin(s) to destination(s). `is_split` is True if the
855 path diverges across multiple cables.
857 `_nodes` retains a flattened list of all nodes within the path to enable simple filtering.
858 """
859 path = models.JSONField(
860 verbose_name=_('path'),
861 default=list
862 )
863 is_active = models.BooleanField(
864 verbose_name=_('is active'),
865 default=False
866 )
867 is_complete = models.BooleanField(
868 verbose_name=_('is complete'),
869 default=False
870 )
871 is_split = models.BooleanField(
872 verbose_name=_('is split'),
873 default=False
874 )
875 _nodes = PathField()
877 _netbox_private = True
879 class Meta:
880 indexes = (
881 # GIN index supports @> operator used by `_nodes__contains` lookups,
882 # which fire on every cable/termination delete and path retrace.
883 GinIndex(fields=('_nodes',)),
884 )
885 verbose_name = _('cable path')
886 verbose_name_plural = _('cable paths')
888 def __str__(self):
889 return f"Path #{self.pk}: {len(self.path)} hops"
891 def save(self, *args, **kwargs):
893 # Save the flattened nodes list
894 self._nodes = list(itertools.chain(*self.path))
896 super().save(*args, **kwargs)
898 # Record a direct reference to this CablePath on its originating object(s). Only PathEndpoint
899 # subclasses carry the denormalized `_path` back-reference; other valid origins (e.g.
900 # CircuitTermination) do not, so skip the update for them.
901 origin_model = self.origin_type.model_class()
902 if issubclass(origin_model, PathEndpoint):
903 origin_ids = [decompile_path_node(node)[1] for node in self.path[0]]
904 chunked_update(origin_model.objects.filter(pk__in=origin_ids), _path=self.pk)
906 def delete(self, *args, **kwargs):
907 # Mirror save() - clear _path on origins to prevent stale references
908 # in table views that render _path.destinations. Only PathEndpoint subclasses carry `_path`.
909 if self.path:
910 origin_model = self.origin_type.model_class()
911 if issubclass(origin_model, PathEndpoint):
912 origin_ids = [decompile_path_node(node)[1] for node in self.path[0]]
913 chunked_update(origin_model.objects.filter(pk__in=origin_ids, _path=self.pk), _path=None)
915 super().delete(*args, **kwargs)
917 @property
918 def origin_type(self):
919 if self.path:
920 ct_id, _ = decompile_path_node(self.path[0][0])
921 return ContentType.objects.get_for_id(ct_id)
922 return None
924 @property
925 def destination_type(self):
926 if self.is_complete:
927 ct_id, _ = decompile_path_node(self.path[-1][0])
928 return ContentType.objects.get_for_id(ct_id)
929 return None
931 @property
932 def _path_decompiled(self):
933 res = []
934 for step in self.path:
935 nodes = []
936 for node in step:
937 nodes.append(decompile_path_node(node))
938 res.append(nodes)
939 return res
941 path_objects = GenericArrayForeignKey("_path_decompiled")
943 @property
944 def origins(self):
945 """
946 Return the list of originating objects.
947 """
948 return self.path_objects[0]
950 @property
951 def destinations(self):
952 """
953 Return the list of destination objects, if the path is complete.
954 """
955 if not self.is_complete:
956 return []
957 return self.path_objects[-1]
959 @property
960 def segment_count(self):
961 return int(len(self.path) / 3)
963 @classmethod
964 def from_origin(cls, terminations):
965 """
966 Create a new CablePath instance as traced from the given termination objects. These can be any object to which a
967 Cable or WirelessLink connects (interfaces, console ports, circuit termination, etc.). All terminations must be
968 of the same type and must belong to the same parent object.
969 """
970 from circuits.models import Circuit, CircuitTermination
972 if not terminations:
973 return None
975 # Ensure all originating terminations are attached to the same link
976 if len(terminations) > 1 and not all(t.link == terminations[0].link for t in terminations[1:]):
977 raise UnsupportedCablePath(_("All originating terminations must be attached to the same link"))
979 path = []
980 position_stack = []
981 is_complete = False
982 is_active = True
983 is_split = False
985 logger.debug(f'Tracing cable path from {terminations}...')
987 segment = 0
988 while terminations:
989 segment += 1
990 logger.debug(f'[Path segment #{segment}] Position stack: {position_stack}')
991 logger.debug(f'[Path segment #{segment}] Local terminations: {terminations}')
993 # Terminations must all be of the same type
994 if not all(isinstance(t, type(terminations[0])) for t in terminations[1:]):
995 raise UnsupportedCablePath(_("All mid-span terminations must have the same termination type"))
997 # All mid-span terminations must all be attached to the same device
998 if (
999 not isinstance(terminations[0], PathEndpoint) and
1000 not isinstance(terminations[0].parent_object, Circuit) and
1001 not all(t.parent_object == terminations[0].parent_object for t in terminations[1:])
1002 ):
1003 raise UnsupportedCablePath(_("All mid-span terminations must have the same parent object"))
1005 # Check for a split path (e.g. rear port fanning out to multiple front ports with
1006 # different cables attached)
1007 if len(set(t.link for t in terminations)) > 1 and (
1008 position_stack and len(terminations) != len(position_stack[-1])
1009 ):
1010 is_split = True
1011 break
1013 # Step 1: Record the near-end termination object(s)
1014 path.append([
1015 object_to_path_node(t) for t in terminations
1016 ])
1017 # If not null, push cable positions onto the stack
1018 if isinstance(terminations[0], PathEndpoint) and terminations[0].cable_positions:
1019 position_stack.append(list(terminations[0].cable_positions))
1021 # Step 2: Determine the attached links (Cable or WirelessLink), if any
1022 links = list(dict.fromkeys(
1023 termination.link for termination in terminations if termination.link is not None
1024 ))
1025 logger.debug(f'[Path segment #{segment}] Links: {links}')
1026 if len(links) == 0:
1027 if len(path) == 1:
1028 # If this is the start of the path and no link exists, return None
1029 return None
1030 # Otherwise, halt the trace if no link exists
1031 break
1032 if not all(type(link) in (Cable, WirelessLink) for link in links):
1033 raise UnsupportedCablePath(_("All links must be cable or wireless"))
1034 if not all(isinstance(link, type(links[0])) for link in links):
1035 raise UnsupportedCablePath(_("All links must match first link type"))
1037 # Step 3: Record asymmetric paths as split
1038 not_connected_terminations = [termination.link for termination in terminations if termination.link is None]
1039 if len(not_connected_terminations) > 0:
1040 is_complete = False
1041 is_split = True
1043 # Step 4: Record the links, keeping cables in order to allow for SVG rendering
1044 cables = []
1045 for link in links:
1046 if object_to_path_node(link) not in cables:
1047 cables.append(object_to_path_node(link))
1048 path.append(cables)
1050 # Step 5: Update the path status if a link is not connected
1051 links_status = [link.status for link in links if link.status != LinkStatusChoices.STATUS_CONNECTED]
1052 if any([status != LinkStatusChoices.STATUS_CONNECTED for status in links_status]):
1053 is_active = False
1055 # Step 6: Determine the far-end terminations
1056 if isinstance(links[0], Cable):
1057 # Profile-based tracing
1058 if links[0].profile:
1059 cable_profile = links[0].profile_class()
1060 if position_stack:
1061 positions = position_stack.pop()
1062 else:
1063 # When the position stack is empty (e.g. the trace reached this
1064 # profiled cable after crossing single-position pass-through ports
1065 # which don't push onto the stack), derive positions from each
1066 # termination's own cable_positions — which were set by this
1067 # profiled cable when it was saved.
1068 positions = [
1069 pos for term in terminations for pos in (term.cable_positions or [])
1070 ]
1071 remote_terminations = []
1072 new_positions = []
1074 # Build (termination, position) pairs by matching stacked positions
1075 # to each termination's cable_positions. This correctly handles
1076 # multiple terminations on different connectors of the same cable.
1077 remaining = Counter(positions)
1078 term_position_pairs = []
1079 for term in terminations:
1080 if term.cable_positions:
1081 for cp in term.cable_positions:
1082 if remaining[cp]:
1083 term_position_pairs.append((term, cp))
1084 remaining[cp] -= 1
1086 # Fallback for when positions don't match cable_positions
1087 if not term_position_pairs:
1088 term_position_pairs = [(terminations[0], pos) for pos in positions or [None]]
1090 peer_results = cable_profile.get_peer_terminations(term_position_pairs)
1091 seen = set()
1092 for peer, new_pos in peer_results:
1093 # If the far-end termination is a channelized interface, resolve to the specific channel
1094 # subinterface bound to the mapped connector position (the far end is channelized on the same
1095 # physical connector, so the peer lookup returns the parent rather than the channel). A
1096 # channelized parent is never itself a path endpoint, so an unoccupied position yields no peer.
1097 if new_pos is not None and getattr(peer, 'channels', None):
1098 peer = peer.child_interfaces.filter(channel_id=new_pos).first()
1099 # Deduplicate peer terminations by model type & PK.
1100 key = None if peer is None else (peer._meta.concrete_model, peer.pk)
1101 if key not in seen:
1102 seen.add(key)
1103 remote_terminations.append(peer)
1104 new_positions.append(new_pos)
1106 # If all peers resolved to None (no far-end terminations exist),
1107 # treat as an empty result so the path is recorded as incomplete
1108 # rather than falling through to the endpoint check with a stale
1109 # None entry.
1110 if remote_terminations and all(peer is None for peer in remote_terminations):
1111 remote_terminations = []
1113 position_stack.append(new_positions)
1115 # Legacy (positionless) behavior
1116 else:
1117 termination_type = ObjectType.objects.get_for_model(terminations[0])
1118 local_cable_terminations = CableTermination.objects.filter(
1119 termination_type=termination_type,
1120 termination_id__in=[t.pk for t in terminations]
1121 )
1123 q_filter = Q()
1124 for lct in local_cable_terminations:
1125 cable_end = 'A' if lct.cable_end == 'B' else 'B'
1126 q_filter |= Q(cable=lct.cable, cable_end=cable_end)
1128 # Make sure this filter has been populated; if not, we have probably been given invalid data
1129 if not q_filter:
1130 break
1132 remote_cable_terminations = CableTermination.objects.filter(q_filter).prefetch_related(
1133 'termination'
1134 )
1135 remote_terminations = [ct.termination for ct in remote_cable_terminations]
1136 else:
1137 # WirelessLink
1138 remote_terminations = [
1139 link.interface_b if link.interface_a is terminations[0] else link.interface_a for link in links
1140 ]
1142 logger.debug(f'[Path segment #{segment}] Remote terminations: {remote_terminations}')
1144 # Remote Terminations must all be of the same type, otherwise return a split path
1145 if not all(isinstance(t, type(remote_terminations[0])) for t in remote_terminations[1:]):
1146 is_complete = False
1147 is_split = True
1148 logger.debug('Remote termination types differ; aborting trace.')
1149 break
1151 # Step 7: Record the far-end termination object(s)
1152 path.append([
1153 object_to_path_node(t) for t in remote_terminations if t is not None
1154 ])
1156 # Step 8: Determine the "next hop" terminations, if applicable
1157 if not remote_terminations:
1158 break
1160 if isinstance(remote_terminations[0], FrontPort):
1161 # Follow FrontPorts to their corresponding RearPorts
1162 if remote_terminations[0].positions > 1 and position_stack:
1163 positions = position_stack.pop()
1164 q_filter = Q()
1165 for rt in remote_terminations:
1166 q_filter |= Q(front_port=rt, front_port_position__in=positions)
1167 port_mappings = PortMapping.objects.filter(q_filter)
1168 elif remote_terminations[0].positions > 1:
1169 is_split = True
1170 logger.debug(
1171 'Encountered front port mapped to multiple rear ports but position stack is empty; aborting '
1172 'trace.'
1173 )
1174 break
1175 else:
1176 port_mappings = PortMapping.objects.filter(front_port__in=remote_terminations)
1177 if not port_mappings:
1178 break
1180 # Compile the list of RearPorts without duplication or altering their ordering
1181 terminations = list(dict.fromkeys(mapping.rear_port for mapping in port_mappings))
1182 if any(t.positions > 1 for t in terminations):
1183 position_stack.append([mapping.rear_port_position for mapping in port_mappings])
1185 elif isinstance(remote_terminations[0], RearPort):
1186 # Follow RearPorts to their corresponding FrontPorts
1187 if remote_terminations[0].positions > 1 and position_stack:
1188 positions = position_stack.pop()
1189 q_filter = Q()
1190 for rt in remote_terminations:
1191 q_filter |= Q(rear_port=rt, rear_port_position__in=positions)
1192 port_mappings = PortMapping.objects.filter(q_filter)
1193 elif remote_terminations[0].positions > 1:
1194 is_split = True
1195 logger.debug(
1196 'Encountered rear port mapped to multiple front ports but position stack is empty; aborting '
1197 'trace.'
1198 )
1199 break
1200 else:
1201 port_mappings = PortMapping.objects.filter(rear_port__in=remote_terminations)
1202 if not port_mappings:
1203 break
1205 # Compile the list of FrontPorts without duplication or altering their ordering
1206 terminations = list(dict.fromkeys(mapping.front_port for mapping in port_mappings))
1207 if any(t.positions > 1 for t in terminations):
1208 position_stack.append([mapping.front_port_position for mapping in port_mappings])
1210 elif isinstance(remote_terminations[0], CircuitTermination):
1211 # Follow a CircuitTermination to its corresponding CircuitTermination (A to Z or vice versa)
1212 qs = Q()
1213 for remote_termination in remote_terminations:
1214 qs |= Q(
1215 circuit=remote_termination.circuit,
1216 term_side='Z' if remote_termination.term_side == 'A' else 'A'
1217 )
1219 # Get all circuit terminations
1220 circuit_terminations = CircuitTermination.objects.filter(qs)
1222 if not circuit_terminations.exists():
1223 break
1224 if all([ct._provider_network for ct in circuit_terminations]):
1225 # Circuit terminates to a ProviderNetwork
1226 path.extend([
1227 [object_to_path_node(ct) for ct in circuit_terminations],
1228 [object_to_path_node(ct._provider_network) for ct in circuit_terminations],
1229 ])
1230 is_complete = True
1231 break
1232 if all([ct.termination and not ct.cable for ct in circuit_terminations]):
1233 # Circuit terminates to a Region/Site/etc.
1234 path.extend([
1235 [object_to_path_node(ct) for ct in circuit_terminations],
1236 [object_to_path_node(ct.termination) for ct in circuit_terminations],
1237 ])
1238 break
1239 if any([ct.cable in links for ct in circuit_terminations]):
1240 # No valid path
1241 is_split = True
1242 break
1244 terminations = circuit_terminations
1246 else:
1247 # Check for non-symmetric path
1248 if all(isinstance(t, type(remote_terminations[0])) for t in remote_terminations[1:]):
1249 is_complete = True
1250 elif len(remote_terminations) == 0:
1251 is_complete = False
1252 else:
1253 # Unsupported topology, mark as split and exit
1254 is_complete = False
1255 is_split = True
1256 logger.warning('Encountered an unsupported topology; aborting trace.')
1257 break
1259 return cls(
1260 path=path,
1261 is_complete=is_complete,
1262 is_active=is_active,
1263 is_split=is_split
1264 )
1266 def retrace(self):
1267 """
1268 Retrace the path from the currently-defined originating termination(s)
1269 """
1270 _new = self.from_origin(self.origins)
1271 if _new:
1272 self.path = _new.path
1273 self.is_complete = _new.is_complete
1274 self.is_active = _new.is_active
1275 self.is_split = _new.is_split
1276 self.save()
1277 else:
1278 self.delete()
1279 retrace.alters_data = True
1281 def get_cable_ids(self):
1282 """
1283 Return all Cable IDs within the path.
1284 """
1285 cable_ct = ObjectType.objects.get_for_model(Cable).pk
1286 cable_ids = []
1288 for node in self._nodes:
1289 ct, id = decompile_path_node(node)
1290 if ct == cable_ct:
1291 cable_ids.append(id)
1293 return cable_ids
1295 def get_total_length(self):
1296 """
1297 Return a tuple containing the sum of the length of each cable and the distance of each circuit
1298 crossed by the path, and a flag indicating whether the length is definitive.
1299 """
1300 from circuits.models import CircuitTermination
1302 object_types = ObjectType.objects.get_for_models(Cable, CircuitTermination)
1303 cable_ct = object_types[Cable].pk
1304 circuit_termination_ct = object_types[CircuitTermination].pk
1306 # Pre-cache cable lengths by ID
1307 cable_ids = self.get_cable_ids()
1308 cables = {
1309 cable['pk']: cable['_abs_length']
1310 for cable in Cable.objects.filter(id__in=cable_ids, _abs_length__isnull=False).values('pk', '_abs_length')
1311 }
1313 # Pre-cache the circuit terminations within the path, along with their circuits
1314 circuit_termination_ids = []
1315 for node in self._nodes:
1316 ct, pk = decompile_path_node(node)
1317 if ct == circuit_termination_ct:
1318 circuit_termination_ids.append(pk)
1319 circuit_terminations = CircuitTermination.objects.select_related('circuit').in_bulk(circuit_termination_ids)
1321 # Iterate through each set of nodes in the path. For cables, add the length of the longest cable to the total
1322 # length of the path. Also map each set of nodes to its circuit terminations, keyed by circuit ID.
1323 total_length = 0
1324 circuit_hops = []
1325 for node_set in self.path:
1326 hop_length = 0
1327 hop_terminations = {}
1328 for node in node_set:
1329 ct, pk = decompile_path_node(node)
1330 if ct == cable_ct:
1331 if pk in cables and cables[pk] > hop_length:
1332 hop_length = cables[pk]
1333 elif ct == circuit_termination_ct:
1334 termination = circuit_terminations.get(pk)
1335 if termination is not None:
1336 hop_terminations[termination.circuit_id] = termination
1337 else:
1338 break # Neither a cable nor a circuit termination
1339 total_length += hop_length
1340 circuit_hops.append(hop_terminations)
1342 # Unresolvable circuit terminations may conceal a crossing, so they render the total non-definitive
1343 is_definitive = len(cables) == len(cable_ids) and len(circuit_terminations) == len(set(circuit_termination_ids))
1345 # A circuit crossing appears as two adjacent sets of opposing terminations of the same circuit. For each
1346 # crossing, add the longest distance among the circuits crossed, mirroring the handling of parallel cables.
1347 for near_hop, far_hop in itertools.pairwise(circuit_hops):
1348 crossing_distance = 0
1349 for circuit_id in near_hop.keys() & far_hop.keys():
1350 if near_hop[circuit_id].term_side == far_hop[circuit_id].term_side:
1351 continue
1352 distance = near_hop[circuit_id].circuit._abs_distance
1353 if distance is None:
1354 is_definitive = False
1355 elif distance > crossing_distance:
1356 crossing_distance = distance
1357 total_length += crossing_distance
1359 return total_length, is_definitive
1361 def get_split_nodes(self):
1362 """
1363 Return all available next segments in a split cable path.
1364 """
1365 from circuits.models import CircuitTermination
1366 nodes = self.path_objects[-1]
1368 # RearPort splitting to multiple FrontPorts with no stack position
1369 if type(nodes[0]) is RearPort:
1370 return [
1371 mapping.front_port for mapping in
1372 PortMapping.objects.filter(rear_port__in=nodes).prefetch_related('front_port')
1373 ]
1374 # Cable terminating to multiple FrontPorts mapped to different
1375 # RearPorts connected to different cables
1376 if type(nodes[0]) is FrontPort:
1377 return [
1378 mapping.rear_port for mapping in
1379 PortMapping.objects.filter(front_port__in=nodes).prefetch_related('rear_port')
1380 ]
1381 # Cable terminating to multiple CircuitTerminations
1382 if type(nodes[0]) is CircuitTermination:
1383 return [
1384 ct.get_peer_termination() for ct in nodes
1385 ]
1386 return []
1388 def get_asymmetric_nodes(self):
1389 """
1390 Return all available next segments in a split cable path.
1391 """
1392 from circuits.models import CircuitTermination
1393 asymmetric_nodes = []
1394 for nodes in self.path_objects:
1395 if type(nodes[0]) in [RearPort, FrontPort, CircuitTermination]:
1396 asymmetric_nodes.extend([node for node in nodes if node.link is None])
1398 return asymmetric_nodes