netbox/netbox/dcim/models/cables.py

import itertools
import logging
import threading
from collections import Counter

from django.contrib.contenttypes.fields import GenericForeignKey
from django.contrib.contenttypes.models import ContentType
from django.contrib.postgres.fields import ArrayField
from django.contrib.postgres.indexes import GinIndex
from django.core.exceptions import ValidationError
from django.core.validators import MaxValueValidator, MinValueValidator
from django.db import models, router, transaction
from django.dispatch import Signal
from django.urls import reverse
from django.utils.translation import gettext_lazy as _

from core.models import ObjectType
from dcim.choices import *
from dcim.constants import *
from dcim.exceptions import UnsupportedCablePath
from dcim.fields import PathField
from dcim.utils import decompile_path_node, object_to_path_node, rebuild_cable_paths
from netbox.choices import ColorChoices
from netbox.models import ChangeLoggedModel, PrimaryModel
from utilities.conversion import to_meters
from utilities.data import normalize_update_fields
from utilities.exceptions import AbortRequest
from utilities.fields import ColorField, GenericArrayForeignKey
from utilities.querysets import RestrictedQuerySet, chunked_update
from utilities.serialization import deserialize_object, serialize_object
from wireless.models import WirelessLink

from .device_components import FrontPort, Interface, PathEndpoint, PortMapping, RearPort

__all__ = (
    'Cable',
    'CableBundle',
    'CablePath',
    'CableTermination',
)

logger = logging.getLogger(f'netbox.{__name__}')

trace_paths = Signal()


#
# Cable bundles
#

class CableBundle(PrimaryModel):
    """
    A logical grouping of individual cables.
    """
    name = models.CharField(
        verbose_name=_('name'),
        max_length=100,
        unique=True,
    )

    class Meta:
        ordering = ('name',)
        verbose_name = _('cable bundle')
        verbose_name_plural = _('cable bundles')

    def __str__(self):
        return self.name

    def get_absolute_url(self):
        return reverse('dcim:cablebundle', args=[self.pk])


#
# Cables
#

class CableQuerySet(RestrictedQuerySet):

    def delete(self):
        # Track these Cables as being deleted for the duration, as Cable.delete() does for a single
        # instance: a queryset delete never calls it. Between them the two cover every deletion, as a
        # Cable is never itself cascade-deleted (nothing points at it with on_delete=CASCADE).
        # Resolve the PKs on the DB the delete will use, so read routing can't miss a lagging replica.
        using = self._db or router.db_for_write(self.model, **self._hints)
        pks = list(self.using(using).values_list('pk', flat=True))
        for pk in pks:
            Cable._track_deletion(pk)
        try:
            return super().delete()
        finally:
            for pk in pks:
                Cable._untrack_deletion(pk)


class Cable(PrimaryModel):
    """
    A physical connection between two endpoints.
    """
    # Per-thread tracking of Cable PKs currently being deleted; referenced by
    # dcim.signals.nullify_connected_endpoints to record the disconnect on each terminating object and
    # to skip per-CableTermination path retracing during the cascade (retrace_cable_paths does it once).
    _deletion_tracking = threading.local()

    type = models.CharField(
        verbose_name=_('type'),
        max_length=50,
        choices=CableTypeChoices,
        blank=True,
        null=True
    )
    status = models.CharField(
        verbose_name=_('status'),
        max_length=50,
        choices=LinkStatusChoices,
        default=LinkStatusChoices.STATUS_CONNECTED
    )
    profile = models.CharField(
        verbose_name=_('profile'),
        max_length=50,
        choices=CableProfileChoices,
        blank=True,
    )
    tenant = models.ForeignKey(
        to='tenancy.Tenant',
        on_delete=models.PROTECT,
        related_name='cables',
        blank=True,
        null=True
    )
    label = models.CharField(
        verbose_name=_('label'),
        max_length=100,
        blank=True
    )
    color = ColorField(
        verbose_name=_('color'),
        blank=True
    )
    length = models.DecimalField(
        verbose_name=_('length'),
        max_digits=8,
        decimal_places=2,
        blank=True,
        null=True
    )
    length_unit = models.CharField(
        verbose_name=_('length unit'),
        max_length=50,
        choices=CableLengthUnitChoices,
        blank=True,
        null=True
    )
    # Stores the normalized length (in meters) for database ordering
    _abs_length = models.DecimalField(
        max_digits=14,
        decimal_places=4,
        blank=True,
        null=True
    )
    bundle = models.ForeignKey(
        to='dcim.CableBundle',
        on_delete=models.SET_NULL,
        related_name='cables',
        blank=True,
        null=True,
        verbose_name=_('bundle'),
    )

    clone_fields = ('tenant', 'type', 'profile', 'bundle')

    objects = CableQuerySet.as_manager()

    class Meta:
        ordering = ('pk',)
        verbose_name = _('cable')
        verbose_name_plural = _('cables')

    def __init__(self, *args, a_terminations=None, b_terminations=None, **kwargs):
        super().__init__(*args, **kwargs)

        # A copy of the PK to be used by __str__ in case the object is deleted
        self._pk = self.__dict__.get('id')

        # Cache the original profile & status so we can check later whether either has been changed
        self._orig_status = self.__dict__.get('status')
        self._orig_profile = self.__dict__.get('profile')

        self._terminations_modified = False

        # Assign or retrieve A/B terminations
        if a_terminations:
            self.a_terminations = a_terminations
        if b_terminations:
            self.b_terminations = b_terminations

    def __str__(self):
        pk = self.pk or self._pk
        return self.label or f'#{pk}'

    def get_status_color(self):
        return LinkStatusChoices.colors.get(self.status)

    @property
    def profile_class(self):
        from dcim import cable_profiles
        return {
            CableProfileChoices.SINGLE_1C1P: cable_profiles.Single1C1PCableProfile,
            CableProfileChoices.SINGLE_1C2P: cable_profiles.Single1C2PCableProfile,
            CableProfileChoices.SINGLE_1C4P: cable_profiles.Single1C4PCableProfile,
            CableProfileChoices.SINGLE_1C6P: cable_profiles.Single1C6PCableProfile,
            CableProfileChoices.SINGLE_1C8P: cable_profiles.Single1C8PCableProfile,
            CableProfileChoices.SINGLE_1C12P: cable_profiles.Single1C12PCableProfile,
            CableProfileChoices.SINGLE_1C16P: cable_profiles.Single1C16PCableProfile,
            CableProfileChoices.TRUNK_2C1P: cable_profiles.Trunk2C1PCableProfile,
            CableProfileChoices.TRUNK_2C2P: cable_profiles.Trunk2C2PCableProfile,
            CableProfileChoices.TRUNK_2C4P: cable_profiles.Trunk2C4PCableProfile,
            CableProfileChoices.TRUNK_2C4P_SHUFFLE: cable_profiles.Trunk2C4PShuffleCableProfile,
            CableProfileChoices.TRUNK_2C6P: cable_profiles.Trunk2C6PCableProfile,
            CableProfileChoices.TRUNK_2C8P: cable_profiles.Trunk2C8PCableProfile,
            CableProfileChoices.TRUNK_2C12P: cable_profiles.Trunk2C12PCableProfile,
            CableProfileChoices.TRUNK_4C1P: cable_profiles.Trunk4C1PCableProfile,
            CableProfileChoices.TRUNK_4C2P: cable_profiles.Trunk4C2PCableProfile,
            CableProfileChoices.TRUNK_4C4P: cable_profiles.Trunk4C4PCableProfile,
            CableProfileChoices.TRUNK_4C4P_SHUFFLE: cable_profiles.Trunk4C4PShuffleCableProfile,
            CableProfileChoices.TRUNK_4C6P: cable_profiles.Trunk4C6PCableProfile,
            CableProfileChoices.TRUNK_4C8P: cable_profiles.Trunk4C8PCableProfile,
            CableProfileChoices.TRUNK_8C4P: cable_profiles.Trunk8C4PCableProfile,
            CableProfileChoices.BREAKOUT_1C2P_2C1P: cable_profiles.Breakout1C2Px2C1PCableProfile,
            CableProfileChoices.BREAKOUT_1C4P_4C1P: cable_profiles.Breakout1C4Px4C1PCableProfile,
            CableProfileChoices.BREAKOUT_1C6P_6C1P: cable_profiles.Breakout1C6Px6C1PCableProfile,
            CableProfileChoices.BREAKOUT_1C8P_8C1P: cable_profiles.Breakout1C8Px8C1PCableProfile,
            CableProfileChoices.BREAKOUT_2C4P_8C1P_SHUFFLE: cable_profiles.Breakout2C4Px8C1PShuffleCableProfile,
        }.get(self.profile)

    def _get_x_terminations(self, side):
        """
        Return the terminating objects for the given cable end (A or B).
        """
        if side not in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B):
            raise ValueError(f"Unknown cable side: {side}")
        attr = f'_{side.lower()}_terminations'

        if hasattr(self, attr):
            return getattr(self, attr)
        if not self.pk:
            return []
        return [
            # Query self.terminations.all() to leverage cached results
            ct.termination for ct in self.terminations.all() if ct.cable_end == side
        ]

    def _set_x_terminations(self, side, value):
        """
        Set the terminating objects for the given cable end (A or B).
        """
        if side not in (CableEndChoices.SIDE_A, CableEndChoices.SIDE_B):
            raise ValueError(f"Unknown cable side: {side}")
        _attr = f'_{side.lower()}_terminations'

        # If the provided value is a list of CableTermination IDs, resolve them
        # to their corresponding termination objects.
        if all(isinstance(item, int) for item in value):
            value = [
                ct.termination for ct in CableTermination.objects.filter(pk__in=value).prefetch_related('termination')
            ]

        if not self.pk or getattr(self, _attr, []) != list(value):
            self._terminations_modified = True

        setattr(self, _attr, value)

    @property
    def a_terminations(self):
        return self._get_x_terminations(CableEndChoices.SIDE_A)

    @a_terminations.setter
    def a_terminations(self, value):
        self._set_x_terminations(CableEndChoices.SIDE_A, value)

    @property
    def b_terminations(self):
        return self._get_x_terminations(CableEndChoices.SIDE_B)

    @b_terminations.setter
    def b_terminations(self, value):
        self._set_x_terminations(CableEndChoices.SIDE_B, value)

    @property
    def color_name(self):
        color_name = ""
        for hex_code, label in ColorChoices.CHOICES:
            if hex_code.lower() == self.color.lower():
                color_name = str(label)

        return color_name

    def clean(self):
        super().clean()

        # Validate length and length_unit
        if self.length is not None and not self.length_unit:
            raise ValidationError(_("Must specify a unit when setting a cable length"))

        if self._state.adding and self.pk is None and (not self.a_terminations or not self.b_terminations):
            raise ValidationError(_("Must define A and B terminations when creating a new cable."))

        # Validate terminations against the assigned cable profile (if any)
        if self.profile:
            self.profile_class().clean(self)

        if self._terminations_modified:

            # Check that all termination objects for either end are of the same type
            for terms in (self.a_terminations, self.b_terminations):
                if len(terms) > 1 and not all(isinstance(t, type(terms[0])) for t in terms[1:]):
                    raise ValidationError(_("Cannot connect different termination types to same end of cable."))

            # Check that termination types are compatible
            if self.a_terminations and self.b_terminations:
                a_type = self.a_terminations[0]._meta.model_name
                b_type = self.b_terminations[0]._meta.model_name
                if b_type not in COMPATIBLE_TERMINATION_TYPES.get(a_type):
                    raise ValidationError(
                        _("Incompatible termination types: {type_a} and {type_b}").format(type_a=a_type, type_b=b_type)
                    )
                if a_type == b_type:
                    # can't directly use self.a_terminations here as possible they
                    # don't have pk yet
                    a_pks = set(obj.pk for obj in self.a_terminations if obj.pk)
                    b_pks = set(obj.pk for obj in self.b_terminations if obj.pk)

                    if (a_pks & b_pks):
                        raise ValidationError(
                            _("A and B terminations cannot connect to the same object.")
                        )

            # Run clean() on any new CableTerminations
            for termination in self.a_terminations:
                CableTermination(cable=self, cable_end='A', termination=termination).clean()
            for termination in self.b_terminations:
                CableTermination(cable=self, cable_end='B', termination=termination).clean()

    def save(self, *args, force_insert=False, force_update=False, using=None, update_fields=None):
        _created = self.pk is None
        save_kwargs = {
            'using': using,
            'update_fields': update_fields,
        }
        update_fields = normalize_update_fields(save_kwargs)

        length_written = update_fields is None or 'length' in update_fields
        length_unit_written = update_fields is None or 'length_unit' in update_fields

        if length_written or length_unit_written:
            if length_written and length_unit_written:
                stored = {}
            else:
                # Read from the database this save will write, so a router cannot split the two
                db = using or router.db_for_write(Cable, instance=self)
                stored = Cable.objects.using(db).filter(pk=self.pk).values('length', 'length_unit').first() or {}
            length = self.length if length_written else stored.get('length')
            length_unit = self.length_unit if length_unit_written else stored.get('length_unit')

            # Clear length_unit if no length is defined
            if length is None and length_unit_written:
                self.length_unit = None

            # Store the given length (if any) in meters for use in database ordering
            if length is not None and length_unit:
                self._abs_length = to_meters(length, length_unit)
            else:
                self._abs_length = None

            # _abs_length is a denormalized cache of length and length_unit, so persist them together
            if update_fields is not None:
                save_kwargs['update_fields'] = update_fields | {'_abs_length'}

        # A field counts as changed only when this save actually writes it
        status_written = update_fields is None or 'status' in update_fields
        profile_written = update_fields is None or 'profile' in update_fields

        # If this is a new Cable, save it before attempting to create its CableTerminations
        if self._state.adding:
            super().save(*args, force_insert=True, **save_kwargs)
            # Update the private PK used in __str__()
            self._pk = self.pk

        if profile_written and self._orig_profile != self.profile:
            self.update_terminations(force=True)
        elif self._terminations_modified:
            self.update_terminations()

        super().save(*args, force_update=True, **save_kwargs)

        try:
            trace_paths.send(Cable, instance=self, created=_created)
        except UnsupportedCablePath as e:
            raise AbortRequest(e)

        # Reset change tracking for the next save of this instance
        if status_written:
            self._orig_status = self.status
        if profile_written:
            self._orig_profile = self.profile
        self._terminations_modified = False

    def delete(self, *args, **kwargs):
        # Cache the PK locally because super().delete() clears self.pk before the finally block runs; the
        # finally also guarantees the PK is discarded if the delete raises. CableQuerySet.delete() tracks
        # the same way for a queryset delete, which never calls this.
        pk = self.pk
        Cable._track_deletion(pk)
        try:
            return super().delete(*args, **kwargs)
        finally:
            Cable._untrack_deletion(pk)

    @classmethod
    def _track_deletion(cls, pk):
        """
        Track a Cable as being deleted, so that the post_delete handler for its cascaded CableTerminations can
        record the disconnect on each terminating object and skip redundant path retracing (retrace_cable_paths()
        retraces each affected path once, after the Cable itself is deleted). The tracking set lives on a
        threading.local() to isolate concurrent deletions across threads.
        """
        if not hasattr(cls._deletion_tracking, 'pks'):
            cls._deletion_tracking.pks = set()
        cls._deletion_tracking.pks.add(pk)

    @classmethod
    def _untrack_deletion(cls, pk):
        if hasattr(cls._deletion_tracking, 'pks'):
            cls._deletion_tracking.pks.discard(pk)

    @classmethod
    def _is_being_deleted(cls, pk):
        return pk in getattr(cls._deletion_tracking, 'pks', ())

    def clone(self):
        """
        Return attributes suitable for cloning this cable.

        In addition to the fields defined in `clone_fields`, include the termination
        type and parent selector fields used by dcim.forms.connections.get_cable_form().
        """
        attrs = super().clone()

        # Mirror dcim.forms.connections.get_cable_form() parent-field logic
        for cable_end, terminations in (('a', self.a_terminations), ('b', self.b_terminations)):
            if not terminations:
                continue

            term_cls = type(terminations[0])
            term_label = term_cls._meta.label_lower

            # Matches CableForm choices: "<app_label>.<model>"
            attrs[f'{cable_end}_terminations_type'] = term_label

            # Device component
            if hasattr(term_cls, 'device'):
                device_ids = sorted({t.device_id for t in terminations if t.device_id})
                if device_ids:
                    attrs[f'termination_{cable_end}_device'] = device_ids

            # PowerFeed
            elif term_label == 'dcim.powerfeed':
                powerpanel_ids = sorted({t.power_panel_id for t in terminations if t.power_panel_id})
                if powerpanel_ids:
                    attrs[f'termination_{cable_end}_powerpanel'] = powerpanel_ids

            # CircuitTermination
            elif term_label == 'circuits.circuittermination':
                circuit_ids = sorted({t.circuit_id for t in terminations if t.circuit_id})
                if circuit_ids:
                    attrs[f'termination_{cable_end}_circuit'] = circuit_ids

        # Never clone the actual terminations, as they are already occupied
        attrs.pop('a_terminations', None)
        attrs.pop('b_terminations', None)

        return attrs

    def serialize_object(self, exclude=None):
        data = serialize_object(self, exclude=exclude or [])

        # Add A & B terminations to the serialized data
        a_terminations, b_terminations = self.get_terminations()
        data['a_terminations'] = sorted([ct.pk for ct in a_terminations.values()])
        data['b_terminations'] = sorted([ct.pk for ct in b_terminations.values()])

        return data

    @classmethod
    def deserialize_object(cls, data, pk=None):
        a_terminations = data.pop('a_terminations', [])
        b_terminations = data.pop('b_terminations', [])

        instance = deserialize_object(cls, data, pk=pk)

        # Assign A & B termination objects to the Cable instance
        queryset = CableTermination.objects.prefetch_related('termination')
        instance.a_terminations = [
            ct.termination for ct in queryset.filter(pk__in=a_terminations)
        ]
        instance.b_terminations = [
            ct.termination for ct in queryset.filter(pk__in=b_terminations)
        ]

        return instance

    def update_dependent_objects(self):
        """
        Recreate the CablePaths traversing this Cable from its current terminations.
        """
        with transaction.atomic(using=router.db_for_write(CablePath)):

            # Restore channel cable attributes omitted by bulk-update change logging.
            for ct in CableTermination.objects.filter(cable=self).prefetch_related('termination'):
                if isinstance(ct.termination, Interface) and ct.termination.channels:
                    ct.termination.propagate_channel_cables()

            rebuild_cable_paths(self)
    update_dependent_objects.alters_data = True

    def get_terminations(self):
        """
        Return two dictionaries mapping A & B side terminating objects to their corresponding CableTerminations
        for this Cable.
        """
        a_terminations = {}
        b_terminations = {}

        for ct in CableTermination.objects.filter(cable=self).prefetch_related('termination'):
            if ct.cable_end == CableEndChoices.SIDE_A:
                a_terminations[ct.termination] = ct
            else:
                b_terminations[ct.termination] = ct

        return a_terminations, b_terminations

    def _connectors_reassigned(self, existing, terminations):
        """
        Return True if any of the given terminating objects already terminates this Cable, but would be
        assigned to a different connector than the one it currently occupies.

        Args:
            existing: Mapping of terminating objects to their current CableTerminations, as returned by
                get_terminations()
            terminations: The ordered list of terminating objects to be assigned to this end of the Cable
        """
        if not self.profile:
            # Connectors are assigned only for a Cable which has a profile
            return False
        for connector, termination in enumerate(terminations, start=1):
            if (ct := existing.get(termination)) and ct.connector != connector:
                return True
        return False

    def update_terminations(self, force=False):
        """
        Create/delete CableTerminations for this Cable to reflect its current state.

        Args:
            force: Force the recreation of all CableTerminations, even if no changes have been made. Needed e.g. when
                altering a Cable's assigned profile.
        """
        a_terminations, b_terminations = self.get_terminations()

        # A CableTermination's connector is derived from its position within its end's list of terminating
        # objects, so reordering that list (or removing an object from the middle of it) rewires the Cable
        # without changing which objects it connects. Recreate the affected end's CableTerminations so that
        # each is reassigned to its new connector.
        force_a = force or self._connectors_reassigned(a_terminations, self.a_terminations)
        force_b = force or self._connectors_reassigned(b_terminations, self.b_terminations)

        # When force-recreating terminations (e.g. after a profile change), cache the termination objects
        # from the database before deleting, so they are available for recreation. Without this, the
        # a_terminations/b_terminations properties would query the DB after deletion and return empty lists.
        if force_a and not hasattr(self, '_a_terminations'):
            self._a_terminations = list(a_terminations.keys())
        if force_b and not hasattr(self, '_b_terminations'):
            self._b_terminations = list(b_terminations.keys())

            # Recreating terminations invalidates existing paths, even when the endpoints are unchanged
            self._terminations_modified = True

        # Delete any stale CableTerminations
        for termination, ct in a_terminations.items():
            if force_a or (termination.pk and termination not in self.a_terminations):
                ct.delete()
        for termination, ct in b_terminations.items():
            if force_b or (termination.pk and termination not in self.b_terminations):
                ct.delete()

        # Save any new CableTerminations
        profile = self.profile_class() if self.profile else None
        for i, termination in enumerate(self.a_terminations, start=1):
            if force_a or not termination.pk or termination not in a_terminations:
                connector = positions = None
                if profile:
                    connector = i
                    positions = profile.get_position_list(profile.a_connectors[i])
                CableTermination(
                    cable=self,
                    cable_end=CableEndChoices.SIDE_A,
                    connector=connector,
                    positions=positions,
                    termination=termination
                ).save()
        for i, termination in enumerate(self.b_terminations, start=1):
            if force_b or not termination.pk or termination not in b_terminations:
                connector = positions = None
                if profile:
                    connector = i
                    positions = profile.get_position_list(profile.b_connectors[i])
                CableTermination(
                    cable=self,
                    cable_end=CableEndChoices.SIDE_B,
                    connector=connector,
                    positions=positions,
                    termination=termination
                ).save()


class CableTermination(ChangeLoggedModel):
    """
    A mapping between side A or B of a Cable and a terminating object (e.g. an Interface or CircuitTermination).
    """
    cable = models.ForeignKey(
        to='dcim.Cable',
        on_delete=models.CASCADE,
        related_name='terminations'
    )
    cable_end = models.CharField(
        max_length=1,
        choices=CableEndChoices,
        verbose_name=_('end')
    )
    termination_type = models.ForeignKey(
        to='contenttypes.ContentType',
        on_delete=models.PROTECT,
        related_name='+'
    )
    termination_id = models.PositiveBigIntegerField()
    termination = GenericForeignKey(
        ct_field='termination_type',
        fk_field='termination_id'
    )
    connector = models.PositiveSmallIntegerField(
        blank=True,
        null=True,
        validators=(
            MinValueValidator(CABLE_CONNECTOR_MIN),
            MaxValueValidator(CABLE_CONNECTOR_MAX)
        ),
    )
    positions = ArrayField(
        base_field=models.PositiveSmallIntegerField(
            validators=(
                MinValueValidator(CABLE_POSITION_MIN),
                MaxValueValidator(CABLE_POSITION_MAX)
            )
        ),
        blank=True,
        null=True,
    )

    # Cached associations to enable efficient filtering
    _device = models.ForeignKey(
        to='dcim.Device',
        on_delete=models.CASCADE,
        blank=True,
        null=True
    )
    _rack = models.ForeignKey(
        to='dcim.Rack',
        on_delete=models.CASCADE,
        blank=True,
        null=True
    )
    _location = models.ForeignKey(
        to='dcim.Location',
        on_delete=models.CASCADE,
        blank=True,
        null=True
    )
    _site = models.ForeignKey(
        to='dcim.Site',
        on_delete=models.CASCADE,
        blank=True,
        null=True
    )

    objects = RestrictedQuerySet.as_manager()

    class Meta:
        ordering = ('cable', 'cable_end', 'connector', 'pk')
        constraints = (
            models.UniqueConstraint(
                fields=('termination_type', 'termination_id'),
                name='%(app_label)s_%(class)s_unique_termination'
            ),
            models.UniqueConstraint(
                fields=('cable', 'cable_end', 'connector'),
                name='%(app_label)s_%(class)s_unique_connector'
            ),
        )
        verbose_name = _('cable termination')
        verbose_name_plural = _('cable terminations')

    def __str__(self):
        return f'Cable {self.cable} to {self.termination}'

    def clean(self):
        super().clean()

        # Disallow connecting a cable to any termination object that is
        # explicitly flagged as "mark connected".
        termination = getattr(self, 'termination', None)
        if termination is not None and getattr(termination, "mark_connected", False):
            raise ValidationError(
                _("Cannot connect a cable to {obj_parent} > {obj} because it is marked as connected.").format(
                    obj_parent=termination.parent_object,
                    obj=termination,
                )
            )

        # Check for existing termination
        qs = CableTermination.objects.filter(
            termination_type=self.termination_type,
            termination_id=self.termination_id
        )
        if self.cable.pk:
            qs = qs.exclude(cable=self.cable)

        existing_termination = qs.first()
        if existing_termination is not None:
            raise ValidationError(
                _("Duplicate termination found for {app_label}.{model} {termination_id}: cable {cable_pk}").format(
                    app_label=self.termination_type.app_label,
                    model=self.termination_type.model,
                    termination_id=self.termination_id,
                    cable_pk=existing_termination.cable.pk
                )
            )
        # A channel subinterface derives its cable from its parent interface and cannot be cabled directly. Checked
        # ahead of the generic type validation below (channel is a nonconnectable type) to surface the more specific
        # guidance.
        if self.termination_type.model == 'interface' and self.termination.channel_id:
            raise ValidationError(
                _("Cables cannot be terminated directly to a channel subinterface; cable the parent interface instead.")
            )

        # Validate the interface type (if applicable)
        if self.termination_type.model == 'interface' and self.termination.type in NONCONNECTABLE_IFACE_TYPES:
            raise ValidationError(
                _("Cables cannot be terminated to {type_display} interfaces").format(
                    type_display=self.termination.get_type_display()
                )
            )

        # A CircuitTermination attached to a ProviderNetwork cannot have a Cable
        if self.termination_type.model == 'circuittermination' and self.termination._provider_network is not None:
            raise ValidationError(_("Circuit terminations attached to a provider network may not be cabled."))

    def save(self, *args, **kwargs):

        # Cache objects associated with the terminating object (for filtering)
        self.cache_related_objects()

        super().save(*args, **kwargs)

        # Set the cable on the terminating object
        termination = self.termination._meta.model.objects.get(pk=self.termination_id)
        termination.snapshot()
        termination.set_cable_termination(self)
        termination.save()

    def delete(self, *args, **kwargs):

        # Delete the cable association on the terminating object
        termination = self.termination._meta.model.objects.get(pk=self.termination_id)
        termination.snapshot()
        termination.clear_cable_termination(self)
        termination.save()

        super().delete(*args, **kwargs)

    def cache_related_objects(self):
        """
        Cache objects related to the termination (e.g. device, rack, site) directly on the object to
        enable efficient filtering.
        """
        if self.termination is None:
            raise ValueError(
                _("Invalid cable termination: the assigned termination object does not exist.")
            )

        # Device components
        if getattr(self.termination, 'device', None):
            self._device = self.termination.device
            self._rack = self.termination.device.rack
            self._location = self.termination.device.location
            self._site = self.termination.device.site

        # Power feeds
        elif getattr(self.termination, 'rack', None):
            self._rack = self.termination.rack
            self._location = self.termination.rack.location
            self._site = self.termination.rack.site

        # Circuit terminations (which cache their own site/location)
        elif self.termination._meta.label_lower == 'circuits.circuittermination':
            self._site = self.termination._site
            self._location = self.termination._location
    cache_related_objects.alters_data = True

    def to_objectchange(self, action):
        objectchange = super().to_objectchange(action)
        objectchange.related_object = self.termination
        return objectchange


class CablePath(models.Model):
    """
    A CablePath instance represents the physical path from a set of origin nodes to a set of destination nodes,
    including all intermediate elements.

    `path` contains the ordered set of nodes, arranged in lists of (type, ID) tuples. (Each cable in the path can
    terminate to one or more objects.)  For example, consider the following
    topology:

                     A                              B                              C
        Interface 1 --- Front Port 1 | Rear Port 1 --- Rear Port 2 | Front Port 3 --- Interface 2
                        Front Port 2                                 Front Port 4

    This path would be expressed as:

    CablePath(
        path = [
            [Interface 1],
            [Cable A],
            [Front Port 1, Front Port 2],
            [Rear Port 1],
            [Cable B],
            [Rear Port 2],
            [Front Port 3, Front Port 4],
            [Cable C],
            [Interface 2],
        ]
    )

    `is_active` is set to True only if every Cable within the path has a status of "connected". `is_complete` is True
    if the instance represents a complete end-to-end path from origin(s) to destination(s). `is_split` is True if the
    path diverges across multiple cables.

    `_nodes` retains a flattened list of all nodes within the path to enable simple filtering.
    """
    path = models.JSONField(
        verbose_name=_('path'),
        default=list
    )
    is_active = models.BooleanField(
        verbose_name=_('is active'),
        default=False
    )
    is_complete = models.BooleanField(
        verbose_name=_('is complete'),
        default=False
    )
    is_split = models.BooleanField(
        verbose_name=_('is split'),
        default=False
    )
    _nodes = PathField()

    _netbox_private = True

    class Meta:
        indexes = (
            # GIN index supports @> operator used by `_nodes__contains` lookups,
            # which fire on every cable/termination delete and path retrace.
            GinIndex(fields=('_nodes',)),
        )
        verbose_name = _('cable path')
        verbose_name_plural = _('cable paths')

    def __str__(self):
        return f"Path #{self.pk}: {len(self.path)} hops"

    def save(self, *args, **kwargs):

        # Save the flattened nodes list
        self._nodes = list(itertools.chain(*self.path))

        super().save(*args, **kwargs)

        # Record a direct reference to this CablePath on its originating object(s). Only PathEndpoint
        # subclasses carry the denormalized `_path` back-reference; other valid origins (e.g.
        # CircuitTermination) do not, so skip the update for them.
        origin_model = self.origin_type.model_class()
        if issubclass(origin_model, PathEndpoint):
            origin_ids = [decompile_path_node(node)[1] for node in self.path[0]]
            chunked_update(origin_model.objects.filter(pk__in=origin_ids), _path=self.pk)

    def delete(self, *args, **kwargs):
        # Mirror save() - clear _path on origins to prevent stale references
        # in table views that render _path.destinations. Only PathEndpoint subclasses carry `_path`.
        if self.path:
            origin_model = self.origin_type.model_class()
            if issubclass(origin_model, PathEndpoint):
                origin_ids = [decompile_path_node(node)[1] for node in self.path[0]]
                chunked_update(origin_model.objects.filter(pk__in=origin_ids, _path=self.pk), _path=None)

        super().delete(*args, **kwargs)

    @property
    def origin_type(self):
        if self.path:
            ct_id, _ = decompile_path_node(self.path[0][0])
            return ContentType.objects.get_for_id(ct_id)
        return None

    @property
    def destination_type(self):
        if self.is_complete:
            ct_id, _ = decompile_path_node(self.path[-1][0])
            return ContentType.objects.get_for_id(ct_id)
        return None

    @property
    def _path_decompiled(self):
        res = []
        for step in self.path:
            nodes = []
            for node in step:
                nodes.append(decompile_path_node(node))
            res.append(nodes)
        return res

    path_objects = GenericArrayForeignKey("_path_decompiled")

    @property
    def origins(self):
        """
        Return the list of originating objects.
        """
        return self.path_objects[0]

    @property
    def destinations(self):
        """
        Return the list of destination objects, if the path is complete.
        """
        if not self.is_complete:
            return []
        return self.path_objects[-1]

    @property
    def segment_count(self):
        return int(len(self.path) / 3)

    @classmethod
    def from_origin(cls, terminations):
        """
        Create a new CablePath instance as traced from the given termination objects. These can be any object to which a
        Cable or WirelessLink connects (interfaces, console ports, circuit termination, etc.). All terminations must be
        of the same type and must belong to the same parent object.
        """
        from circuits.models import Circuit, CircuitTermination

        if not terminations:
            return None

        # Ensure all originating terminations are attached to the same link
        if len(terminations) > 1 and not all(t.link == terminations[0].link for t in terminations[1:]):
            raise UnsupportedCablePath(_("All originating terminations must be attached to the same link"))

        path = []
        position_stack = []
        is_complete = False
        is_active = True
        is_split = False

        logger.debug(f'Tracing cable path from {terminations}...')

        segment = 0
        while terminations:
            segment += 1
            logger.debug(f'[Path segment #{segment}] Position stack: {position_stack}')
            logger.debug(f'[Path segment #{segment}] Local terminations: {terminations}')

            # Terminations must all be of the same type
            if not all(isinstance(t, type(terminations[0])) for t in terminations[1:]):
                raise UnsupportedCablePath(_("All mid-span terminations must have the same termination type"))

            # All mid-span terminations must all be attached to the same device
            if (
                not isinstance(terminations[0], PathEndpoint) and
                not isinstance(terminations[0].parent_object, Circuit) and
                not all(t.parent_object == terminations[0].parent_object for t in terminations[1:])
            ):
                raise UnsupportedCablePath(_("All mid-span terminations must have the same parent object"))

            # Check for a split path (e.g. rear port fanning out to multiple front ports with
            # different cables attached)
            if len(set(t.link for t in terminations)) > 1 and (
                    position_stack and len(terminations) != len(position_stack[-1])
            ):
                is_split = True
                break

            # Step 1: Record the near-end termination object(s)
            path.append([
                object_to_path_node(t) for t in terminations
            ])
            # If not null, push cable positions onto the stack
            if isinstance(terminations[0], PathEndpoint) and terminations[0].cable_positions:
                position_stack.append(list(terminations[0].cable_positions))

            # Step 2: Determine the attached links (Cable or WirelessLink), if any
            links = list(dict.fromkeys(
                termination.link for termination in terminations if termination.link is not None
            ))
            logger.debug(f'[Path segment #{segment}] Links: {links}')
            if len(links) == 0:
                if len(path) == 1:
                    # If this is the start of the path and no link exists, return None
                    return None
                # Otherwise, halt the trace if no link exists
                break
            if not all(type(link) in (Cable, WirelessLink) for link in links):
                raise UnsupportedCablePath(_("All links must be cable or wireless"))
            if not all(isinstance(link, type(links[0])) for link in links):
                raise UnsupportedCablePath(_("All links must match first link type"))

            # Step 3: Record asymmetric paths as split
            not_connected_terminations = [termination.link for termination in terminations if termination.link is None]
            if len(not_connected_terminations) > 0:
                is_complete = False
                is_split = True

            # Step 4: Record the links, keeping cables in order to allow for SVG rendering
            cables = []
            for link in links:
                if object_to_path_node(link) not in cables:
                    cables.append(object_to_path_node(link))
            path.append(cables)

            # Step 5: Update the path status if a link is not connected
            links_status = [link.status for link in links if link.status != LinkStatusChoices.STATUS_CONNECTED]
            if any([status != LinkStatusChoices.STATUS_CONNECTED for status in links_status]):
                is_active = False

            # Step 6: Determine the far-end terminations
            if isinstance(links[0], Cable):
                # Profile-based tracing
                if links[0].profile:
                    cable_profile = links[0].profile_class()
                    if position_stack:
                        positions = position_stack.pop()
                    else:
                        # When the position stack is empty (e.g. the trace reached this
                        # profiled cable after crossing single-position pass-through ports
                        # which don't push onto the stack), derive positions from each
                        # termination's own cable_positions — which were set by this
                        # profiled cable when it was saved.
                        positions = [
                            pos for term in terminations for pos in (term.cable_positions or [])
                        ]
                    remote_terminations = []
                    new_positions = []

                    # Build (termination, position) pairs by matching stacked positions
                    # to each termination's cable_positions. This correctly handles
                    # multiple terminations on different connectors of the same cable.
                    remaining = Counter(positions)
                    term_position_pairs = []
                    for term in terminations:
                        if term.cable_positions:
                            for cp in term.cable_positions:
                                if remaining[cp]:
                                    term_position_pairs.append((term, cp))
                                    remaining[cp] -= 1

                    # Fallback for when positions don't match cable_positions
                    if not term_position_pairs:
                        term_position_pairs = [(terminations[0], pos) for pos in positions or [None]]

                    peer_results = cable_profile.get_peer_terminations(term_position_pairs)
                    seen = set()
                    for peer, new_pos in peer_results:
                        # If the far-end termination is a channelized interface, resolve to the specific channel
                        # subinterface bound to the mapped connector position (the far end is channelized on the same
                        # physical connector, so the peer lookup returns the parent rather than the channel). A
                        # channelized parent is never itself a path endpoint, so an unoccupied position yields no peer.
                        if new_pos is not None and getattr(peer, 'channels', None):
                            peer = peer.child_interfaces.filter(channel_id=new_pos).first()
                        # Deduplicate peer terminations by model type & PK.
                        key = None if peer is None else (peer._meta.concrete_model, peer.pk)
                        if key not in seen:
                            seen.add(key)
                            remote_terminations.append(peer)
                        new_positions.append(new_pos)

                    # If all peers resolved to None (no far-end terminations exist),
                    # treat as an empty result so the path is recorded as incomplete
                    # rather than falling through to the endpoint check with a stale
                    # None entry.
                    if remote_terminations and all(peer is None for peer in remote_terminations):
                        remote_terminations = []

                    position_stack.append(new_positions)

                # Legacy (positionless) behavior
                else:
                    termination_type = ObjectType.objects.get_for_model(terminations[0])
                    local_cable_terminations = CableTermination.objects.filter(
                        termination_type=termination_type,
                        termination_id__in=[t.pk for t in terminations]
                    )

                    q_filter = Q()
                    for lct in local_cable_terminations:
                        cable_end = 'A' if lct.cable_end == 'B' else 'B'
                        q_filter |= Q(cable=lct.cable, cable_end=cable_end)

                    # Make sure this filter has been populated; if not, we have probably been given invalid data
                    if not q_filter:
                        break

                    remote_cable_terminations = CableTermination.objects.filter(q_filter).prefetch_related(
                        'termination'
                    )
                    remote_terminations = [ct.termination for ct in remote_cable_terminations]
            else:
                # WirelessLink
                remote_terminations = [
                    link.interface_b if link.interface_a is terminations[0] else link.interface_a for link in links
                ]

            logger.debug(f'[Path segment #{segment}] Remote terminations: {remote_terminations}')

            # Remote Terminations must all be of the same type, otherwise return a split path
            if not all(isinstance(t, type(remote_terminations[0])) for t in remote_terminations[1:]):
                is_complete = False
                is_split = True
                logger.debug('Remote termination types differ; aborting trace.')
                break

            # Step 7: Record the far-end termination object(s)
            path.append([
                object_to_path_node(t) for t in remote_terminations if t is not None
            ])

            # Step 8: Determine the "next hop" terminations, if applicable
            if not remote_terminations:
                break

            if isinstance(remote_terminations[0], FrontPort):
                # Follow FrontPorts to their corresponding RearPorts
                if remote_terminations[0].positions > 1 and position_stack:
                    positions = position_stack.pop()
                    q_filter = Q()
                    for rt in remote_terminations:
                        q_filter |= Q(front_port=rt, front_port_position__in=positions)
                    port_mappings = PortMapping.objects.filter(q_filter)
                elif remote_terminations[0].positions > 1:
                    is_split = True
                    logger.debug(
                        'Encountered front port mapped to multiple rear ports but position stack is empty; aborting '
                        'trace.'
                    )
                    break
                else:
                    port_mappings = PortMapping.objects.filter(front_port__in=remote_terminations)
                if not port_mappings:
                    break

                # Compile the list of RearPorts without duplication or altering their ordering
                terminations = list(dict.fromkeys(mapping.rear_port for mapping in port_mappings))
                if any(t.positions > 1 for t in terminations):
                    position_stack.append([mapping.rear_port_position for mapping in port_mappings])

            elif isinstance(remote_terminations[0], RearPort):
                # Follow RearPorts to their corresponding FrontPorts
                if remote_terminations[0].positions > 1 and position_stack:
                    positions = position_stack.pop()
                    q_filter = Q()
                    for rt in remote_terminations:
                        q_filter |= Q(rear_port=rt, rear_port_position__in=positions)
                    port_mappings = PortMapping.objects.filter(q_filter)
                elif remote_terminations[0].positions > 1:
                    is_split = True
                    logger.debug(
                        'Encountered rear port mapped to multiple front ports but position stack is empty; aborting '
                        'trace.'
                    )
                    break
                else:
                    port_mappings = PortMapping.objects.filter(rear_port__in=remote_terminations)
                if not port_mappings:
                    break

                # Compile the list of FrontPorts without duplication or altering their ordering
                terminations = list(dict.fromkeys(mapping.front_port for mapping in port_mappings))
                if any(t.positions > 1 for t in terminations):
                    position_stack.append([mapping.front_port_position for mapping in port_mappings])

            elif isinstance(remote_terminations[0], CircuitTermination):
                # Follow a CircuitTermination to its corresponding CircuitTermination (A to Z or vice versa)
                qs = Q()
                for remote_termination in remote_terminations:
                    qs |= Q(
                        circuit=remote_termination.circuit,
                        term_side='Z' if remote_termination.term_side == 'A' else 'A'
                    )

                # Get all circuit terminations
                circuit_terminations = CircuitTermination.objects.filter(qs)

                if not circuit_terminations.exists():
                    break
                if all([ct._provider_network for ct in circuit_terminations]):
                    # Circuit terminates to a ProviderNetwork
                    path.extend([
                        [object_to_path_node(ct) for ct in circuit_terminations],
                        [object_to_path_node(ct._provider_network) for ct in circuit_terminations],
                    ])
                    is_complete = True
                    break
                if all([ct.termination and not ct.cable for ct in circuit_terminations]):
                    # Circuit terminates to a Region/Site/etc.
                    path.extend([
                        [object_to_path_node(ct) for ct in circuit_terminations],
                        [object_to_path_node(ct.termination) for ct in circuit_terminations],
                    ])
                    break
                if any([ct.cable in links for ct in circuit_terminations]):
                    # No valid path
                    is_split = True
                    break

                terminations = circuit_terminations

            else:
                # Check for non-symmetric path
                if all(isinstance(t, type(remote_terminations[0])) for t in remote_terminations[1:]):
                    is_complete = True
                elif len(remote_terminations) == 0:
                    is_complete = False
                else:
                    # Unsupported topology, mark as split and exit
                    is_complete = False
                    is_split = True
                    logger.warning('Encountered an unsupported topology; aborting trace.')
                break

        return cls(
            path=path,
            is_complete=is_complete,
            is_active=is_active,
            is_split=is_split
        )

    def retrace(self):
        """
        Retrace the path from the currently-defined originating termination(s)
        """
        _new = self.from_origin(self.origins)
        if _new:
            self.path = _new.path
            self.is_complete = _new.is_complete
            self.is_active = _new.is_active
            self.is_split = _new.is_split
            self.save()
        else:
            self.delete()
    retrace.alters_data = True

    def get_cable_ids(self):
        """
        Return all Cable IDs within the path.
        """
        cable_ct = ObjectType.objects.get_for_model(Cable).pk
        cable_ids = []

        for node in self._nodes:
            ct, id = decompile_path_node(node)
            if ct == cable_ct:
                cable_ids.append(id)

        return cable_ids

    def get_total_length(self):
        """
        Return a tuple containing the sum of the length of each cable and the distance of each circuit
        crossed by the path, and a flag indicating whether the length is definitive.
        """
        from circuits.models import CircuitTermination

        object_types = ObjectType.objects.get_for_models(Cable, CircuitTermination)
        cable_ct = object_types[Cable].pk
        circuit_termination_ct = object_types[CircuitTermination].pk

        # Pre-cache cable lengths by ID
        cable_ids = self.get_cable_ids()
        cables = {
            cable['pk']: cable['_abs_length']
            for cable in Cable.objects.filter(id__in=cable_ids, _abs_length__isnull=False).values('pk', '_abs_length')
        }

        # Pre-cache the circuit terminations within the path, along with their circuits
        circuit_termination_ids = []
        for node in self._nodes:
            ct, pk = decompile_path_node(node)
            if ct == circuit_termination_ct:
                circuit_termination_ids.append(pk)
        circuit_terminations = CircuitTermination.objects.select_related('circuit').in_bulk(circuit_termination_ids)

        # Iterate through each set of nodes in the path. For cables, add the length of the longest cable to the total
        # length of the path. Also map each set of nodes to its circuit terminations, keyed by circuit ID.
        total_length = 0
        circuit_hops = []
        for node_set in self.path:
            hop_length = 0
            hop_terminations = {}
            for node in node_set:
                ct, pk = decompile_path_node(node)
                if ct == cable_ct:
                    if pk in cables and cables[pk] > hop_length:
                        hop_length = cables[pk]
                elif ct == circuit_termination_ct:
                    termination = circuit_terminations.get(pk)
                    if termination is not None:
                        hop_terminations[termination.circuit_id] = termination
                else:
                    break  # Neither a cable nor a circuit termination
            total_length += hop_length
            circuit_hops.append(hop_terminations)

        # Unresolvable circuit terminations may conceal a crossing, so they render the total non-definitive
        is_definitive = len(cables) == len(cable_ids) and len(circuit_terminations) == len(set(circuit_termination_ids))

        # A circuit crossing appears as two adjacent sets of opposing terminations of the same circuit. For each
        # crossing, add the longest distance among the circuits crossed, mirroring the handling of parallel cables.
        for near_hop, far_hop in itertools.pairwise(circuit_hops):
            crossing_distance = 0
            for circuit_id in near_hop.keys() & far_hop.keys():
                if near_hop[circuit_id].term_side == far_hop[circuit_id].term_side:
                    continue
                distance = near_hop[circuit_id].circuit._abs_distance
                if distance is None:
                    is_definitive = False
                elif distance > crossing_distance:
                    crossing_distance = distance
            total_length += crossing_distance

        return total_length, is_definitive

    def get_split_nodes(self):
        """
        Return all available next segments in a split cable path.
        """
        from circuits.models import CircuitTermination
        nodes = self.path_objects[-1]

        # RearPort splitting to multiple FrontPorts with no stack position
        if type(nodes[0]) is RearPort:
            return [
                mapping.front_port for mapping in
                PortMapping.objects.filter(rear_port__in=nodes).prefetch_related('front_port')
            ]
        # Cable terminating to multiple FrontPorts mapped to different
        # RearPorts connected to different cables
        if type(nodes[0]) is FrontPort:
            return [
                mapping.rear_port for mapping in
                PortMapping.objects.filter(front_port__in=nodes).prefetch_related('rear_port')
            ]
        # Cable terminating to multiple CircuitTerminations
        if type(nodes[0]) is CircuitTermination:
            return [
                ct.get_peer_termination() for ct in nodes
            ]
        return []

    def get_asymmetric_nodes(self):
        """
        Return all available next segments in a split cable path.
        """
        from circuits.models import CircuitTermination
        asymmetric_nodes = []
        for nodes in self.path_objects:
            if type(nodes[0]) in [RearPort, FrontPort, CircuitTermination]:
                asymmetric_nodes.extend([node for node in nodes if node.link is None])

        return asymmetric_nodes