Coverage for dcim/cable_profiles.py: 0%
166 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-10-10 18:35 +0000
1from collections import defaultdict, namedtuple
3from django.core.exceptions import ValidationError
4from django.db.models import Q
5from django.utils.translation import gettext_lazy as _
7from dcim.choices import CableEndChoices
8from dcim.models import CableTermination
10PeerLookupKey = namedtuple('PeerLookupKey', ['cable_id', 'cable_end', 'connector', 'position'])
13class BaseCableProfile:
14 """Base class for representing a cable profile."""
16 # Mappings of connectors to the number of positions presented by each, at either end of the cable. For example, a
17 # 12-strand MPO fiber cable would have one connector at either end with six positions (six bidirectional fiber
18 # pairs).
19 a_connectors = {}
20 b_connectors = {}
22 # Defined a mapping of A/B connector & position pairings. If not defined, all positions are presumed to be
23 # symmetrical (i.e. 1:1 on side A maps to 1:1 on side B). If defined, it must be constructed as a dictionary of
24 # two-item tuples, e.g. {(1, 1): (1, 1)}.
25 _mapping = None
27 def clean(self, cable):
28 # Enforce maximum terminations limits
29 a_terminations_count = len(cable.a_terminations)
30 b_terminations_count = len(cable.b_terminations)
31 max_a_terminations = len(self.a_connectors)
32 max_b_terminations = len(self.b_connectors)
33 if a_terminations_count > max_a_terminations:
34 raise ValidationError({
35 'a_terminations': _(
36 'A side of cable has {count} terminations but only {max} are permitted for profile {profile}'
37 ).format(
38 count=a_terminations_count,
39 profile=cable.get_profile_display(),
40 max=max_a_terminations,
41 )
42 })
43 if b_terminations_count > max_b_terminations:
44 raise ValidationError({
45 'b_terminations': _(
46 'B side of cable has {count} terminations but only {max} are permitted for profile {profile}'
47 ).format(
48 count=b_terminations_count,
49 profile=cable.get_profile_display(),
50 max=max_b_terminations,
51 )
52 })
54 def get_mapped_position(self, side, connector, position):
55 """
56 Return the mapped far-end connector & position for a given cable end the local connector & position.
57 """
58 # By default, assume all positions are symmetrical.
59 if self._mapping:
60 return self._mapping.get((connector, position))
61 return connector, position
63 #
64 # Peer termination resolution
65 #
67 def _get_peer_lookup_key(self, termination, position):
68 """
69 Given a cabled object and a local position, return the (cable_id, cable_end, connector, position)
70 tuple identifying the corresponding peer CableTermination on the opposite end of the cable.
71 """
72 try:
73 connector, position = self.get_mapped_position(
74 termination.cable_end,
75 termination.cable_connector,
76 position
77 )
78 except TypeError:
79 raise ValueError(
80 f"Could not map connector {termination.cable_connector} position {position} on side "
81 f"{termination.cable_end}"
82 )
83 return PeerLookupKey(termination.cable_id, termination.opposite_cable_end, connector, position)
85 def get_peer_termination(self, termination, position):
86 """
87 Given a terminating object, return the peer terminating object (if any) on the opposite end of the cable.
88 """
89 cable_id, cable_end, connector, position = self._get_peer_lookup_key(termination, position)
90 try:
91 ct = CableTermination.objects.get(
92 cable_id=cable_id,
93 cable_end=cable_end,
94 connector=connector,
95 positions__contains=[position],
96 )
97 return ct.termination, position
98 except CableTermination.DoesNotExist:
99 return None, None
101 def get_peer_terminations(self, term_position_pairs):
102 """
103 Resolve a batch of (termination, position) pairs to peer terminations.
104 """
105 if not term_position_pairs:
106 return []
108 # Fast path: a single pair doesn't benefit from batching
109 if len(term_position_pairs) == 1:
110 return [self.get_peer_termination(*term_position_pairs[0])]
112 lookup_keys = [
113 self._get_peer_lookup_key(termination, position) for termination, position in term_position_pairs
114 ]
116 # Group requested positions by (cable_id, cable_end, connector) so we can
117 # build one overlap clause per group instead of one clause per position.
118 positions_by_group = defaultdict(set)
119 for key in lookup_keys:
120 positions_by_group[(key.cable_id, key.cable_end, key.connector)].add(key.position)
122 q_filter = Q()
123 for (cable_id, cable_end, connector), positions in positions_by_group.items():
124 q_filter |= Q(
125 cable_id=cable_id,
126 cable_end=cable_end,
127 connector=connector,
128 positions__overlap=list(positions),
129 )
131 peer_ct_by_key = {}
132 term_ids_by_type = defaultdict(set)
133 model_by_type = {}
135 for ct in CableTermination.objects.filter(q_filter).select_related('termination_type'):
136 model_by_type[ct.termination_type_id] = ct.termination_type.model_class()
137 term_ids_by_type[ct.termination_type_id].add(ct.termination_id)
139 group_key = (ct.cable_id, ct.cable_end, ct.connector)
140 requested_positions = positions_by_group.get(group_key, ())
142 # The overlap query may return a termination carrying additional
143 # positions, so only index the positions we actually requested.
144 for pos in ct.positions or []:
145 if pos not in requested_positions:
146 continue
148 key = PeerLookupKey(ct.cable_id, ct.cable_end, ct.connector, pos)
149 if key in peer_ct_by_key and peer_ct_by_key[key].pk != ct.pk:
150 raise CableTermination.MultipleObjectsReturned(
151 f"Multiple peer terminations found for cable {ct.cable_id} end {ct.cable_end} "
152 f"connector {ct.connector} position {pos}"
153 )
154 peer_ct_by_key[key] = ct
156 # Use _base_manager to ensure all termination objects are loaded
157 # regardless of any custom default manager filters (e.g. soft-delete).
158 # Note: Django's GenericForeignKey / ContentType.get_object_for_this_type()
159 # uses _default_manager, but _base_manager is safer for bulk resolution
160 # during path tracing.
161 term_by_type = {
162 type_id: model_by_type[type_id]._base_manager.in_bulk(ids) for type_id, ids in term_ids_by_type.items()
163 }
165 results = []
166 for key in lookup_keys:
167 ct = peer_ct_by_key.get(key)
168 if ct is None:
169 results.append((None, None))
170 else:
171 termination = term_by_type[ct.termination_type_id].get(ct.termination_id)
172 results.append((termination, key.position if termination is not None else None))
174 return results
176 @staticmethod
177 def get_position_list(n):
178 """Return a list of integers from 1 to n, inclusive."""
179 return list(range(1, n + 1))
182# Profile naming:
183# - Single: One connector per side, with one or more positions
184# - Trunk: Two or more connectors per side, with one or more positions per connector
185# - Breakout: One or more connectors on the A side which map to a greater number of B side connectors
186# - Shuffle: A cable with nonlinear position mappings between sides
188class Single1C1PCableProfile(BaseCableProfile):
189 a_connectors = {
190 1: 1,
191 }
192 b_connectors = a_connectors
195class Single1C2PCableProfile(BaseCableProfile):
196 a_connectors = {
197 1: 2,
198 }
199 b_connectors = a_connectors
202class Single1C4PCableProfile(BaseCableProfile):
203 a_connectors = {
204 1: 4,
205 }
206 b_connectors = a_connectors
209class Single1C6PCableProfile(BaseCableProfile):
210 a_connectors = {
211 1: 6,
212 }
213 b_connectors = a_connectors
216class Single1C8PCableProfile(BaseCableProfile):
217 a_connectors = {
218 1: 8,
219 }
220 b_connectors = a_connectors
223class Single1C12PCableProfile(BaseCableProfile):
224 a_connectors = {
225 1: 12,
226 }
227 b_connectors = a_connectors
230class Single1C16PCableProfile(BaseCableProfile):
231 a_connectors = {
232 1: 16,
233 }
234 b_connectors = a_connectors
237class Trunk2C1PCableProfile(BaseCableProfile):
238 a_connectors = {
239 1: 1,
240 2: 1,
241 }
242 b_connectors = a_connectors
245class Trunk2C2PCableProfile(BaseCableProfile):
246 a_connectors = {
247 1: 2,
248 2: 2,
249 }
250 b_connectors = a_connectors
253class Trunk2C4PCableProfile(BaseCableProfile):
254 a_connectors = {
255 1: 4,
256 2: 4,
257 }
258 b_connectors = a_connectors
261class Trunk2C6PCableProfile(BaseCableProfile):
262 a_connectors = {
263 1: 6,
264 2: 6,
265 }
266 b_connectors = a_connectors
269class Trunk2C8PCableProfile(BaseCableProfile):
270 a_connectors = {
271 1: 8,
272 2: 8,
273 }
274 b_connectors = a_connectors
277class Trunk2C12PCableProfile(BaseCableProfile):
278 a_connectors = {
279 1: 12,
280 2: 12,
281 }
282 b_connectors = a_connectors
285class Trunk4C1PCableProfile(BaseCableProfile):
286 a_connectors = {
287 1: 1,
288 2: 1,
289 3: 1,
290 4: 1,
291 }
292 b_connectors = a_connectors
295class Trunk4C2PCableProfile(BaseCableProfile):
296 a_connectors = {
297 1: 2,
298 2: 2,
299 3: 2,
300 4: 2,
301 }
302 b_connectors = a_connectors
305class Trunk4C4PCableProfile(BaseCableProfile):
306 a_connectors = {
307 1: 4,
308 2: 4,
309 3: 4,
310 4: 4,
311 }
312 b_connectors = a_connectors
315class Trunk4C6PCableProfile(BaseCableProfile):
316 a_connectors = {
317 1: 6,
318 2: 6,
319 3: 6,
320 4: 6,
321 }
322 b_connectors = a_connectors
325class Trunk4C8PCableProfile(BaseCableProfile):
326 a_connectors = {
327 1: 8,
328 2: 8,
329 3: 8,
330 4: 8,
331 }
332 b_connectors = a_connectors
335class Trunk8C4PCableProfile(BaseCableProfile):
336 a_connectors = {
337 1: 4,
338 2: 4,
339 3: 4,
340 4: 4,
341 5: 4,
342 6: 4,
343 7: 4,
344 8: 4,
345 }
346 b_connectors = a_connectors
349class Breakout1C2Px2C1PCableProfile(BaseCableProfile):
350 a_connectors = {
351 1: 2,
352 }
353 b_connectors = {
354 1: 1,
355 2: 1,
356 }
357 _mapping = {
358 (1, 1): (1, 1),
359 (1, 2): (2, 1),
360 (2, 1): (1, 2),
361 }
364class Breakout1C4Px4C1PCableProfile(BaseCableProfile):
365 a_connectors = {
366 1: 4,
367 }
368 b_connectors = {
369 1: 1,
370 2: 1,
371 3: 1,
372 4: 1,
373 }
374 _mapping = {
375 (1, 1): (1, 1),
376 (1, 2): (2, 1),
377 (1, 3): (3, 1),
378 (1, 4): (4, 1),
379 (2, 1): (1, 2),
380 (3, 1): (1, 3),
381 (4, 1): (1, 4),
382 }
385class Breakout1C6Px6C1PCableProfile(BaseCableProfile):
386 a_connectors = {
387 1: 6,
388 }
389 b_connectors = {
390 1: 1,
391 2: 1,
392 3: 1,
393 4: 1,
394 5: 1,
395 6: 1,
396 }
397 _mapping = {
398 (1, 1): (1, 1),
399 (1, 2): (2, 1),
400 (1, 3): (3, 1),
401 (1, 4): (4, 1),
402 (1, 5): (5, 1),
403 (1, 6): (6, 1),
404 (2, 1): (1, 2),
405 (3, 1): (1, 3),
406 (4, 1): (1, 4),
407 (5, 1): (1, 5),
408 (6, 1): (1, 6),
409 }
412class Breakout1C8Px8C1PCableProfile(BaseCableProfile):
413 a_connectors = {
414 1: 8,
415 }
416 b_connectors = {
417 1: 1,
418 2: 1,
419 3: 1,
420 4: 1,
421 5: 1,
422 6: 1,
423 7: 1,
424 8: 1,
425 }
426 _mapping = {
427 (1, 1): (1, 1),
428 (1, 2): (2, 1),
429 (1, 3): (3, 1),
430 (1, 4): (4, 1),
431 (1, 5): (5, 1),
432 (1, 6): (6, 1),
433 (1, 7): (7, 1),
434 (1, 8): (8, 1),
435 (2, 1): (1, 2),
436 (3, 1): (1, 3),
437 (4, 1): (1, 4),
438 (5, 1): (1, 5),
439 (6, 1): (1, 6),
440 (7, 1): (1, 7),
441 (8, 1): (1, 8),
442 }
445class Trunk2C4PShuffleCableProfile(BaseCableProfile):
446 a_connectors = {
447 1: 4,
448 2: 4,
449 }
450 b_connectors = a_connectors
451 _mapping = {
452 (1, 1): (1, 1),
453 (1, 2): (1, 2),
454 (1, 3): (2, 1),
455 (1, 4): (2, 2),
456 (2, 1): (1, 3),
457 (2, 2): (1, 4),
458 (2, 3): (2, 3),
459 (2, 4): (2, 4),
460 }
463class Trunk4C4PShuffleCableProfile(BaseCableProfile):
464 a_connectors = {
465 1: 4,
466 2: 4,
467 3: 4,
468 4: 4,
469 }
470 b_connectors = a_connectors
471 _mapping = {
472 (1, 1): (1, 1),
473 (1, 2): (2, 1),
474 (1, 3): (3, 1),
475 (1, 4): (4, 1),
476 (2, 1): (1, 2),
477 (2, 2): (2, 2),
478 (2, 3): (3, 2),
479 (2, 4): (4, 2),
480 (3, 1): (1, 3),
481 (3, 2): (2, 3),
482 (3, 3): (3, 3),
483 (3, 4): (4, 3),
484 (4, 1): (1, 4),
485 (4, 2): (2, 4),
486 (4, 3): (3, 4),
487 (4, 4): (4, 4),
488 }
491class Breakout2C4Px8C1PShuffleCableProfile(BaseCableProfile):
492 a_connectors = {
493 1: 4,
494 2: 4,
495 }
496 b_connectors = {
497 1: 1,
498 2: 1,
499 3: 1,
500 4: 1,
501 5: 1,
502 6: 1,
503 7: 1,
504 8: 1,
505 }
506 _a_mapping = {
507 (1, 1): (1, 1),
508 (1, 2): (2, 1),
509 (1, 3): (5, 1),
510 (1, 4): (6, 1),
511 (2, 1): (3, 1),
512 (2, 2): (4, 1),
513 (2, 3): (7, 1),
514 (2, 4): (8, 1),
515 }
516 _b_mapping = {
517 (1, 1): (1, 1),
518 (2, 1): (1, 2),
519 (3, 1): (2, 1),
520 (4, 1): (2, 2),
521 (5, 1): (1, 3),
522 (6, 1): (1, 4),
523 (7, 1): (2, 3),
524 (8, 1): (2, 4),
525 }
527 def get_mapped_position(self, side, connector, position):
528 if side.upper() == CableEndChoices.SIDE_A:
529 return self._a_mapping.get((connector, position))
530 return self._b_mapping.get((connector, position))