Skip to content

p4net.topo

Topology description layer: pure data classes, no syscalls.

__all__ module-attribute 

__all__ = ['Host', 'Link', 'LinkEndpoint', 'P4Switch', 'Topology', 'TopologyError']

TopologyError

Bases: P4NetError

Topology validation or construction failure.

Source code in src/p4net/topo/exceptions.py 
8
9
class TopologyError(P4NetError):
    """Topology validation or construction failure."""

Host dataclass

A network host with optional primary IP, MAC, and default route.

IPv4 lives in ip / default_route; IPv6 lives in ip6 / default_route6. Either, neither, or both may be set. Addresses and gateways must be strings or None.

Source code in src/p4net/topo/host.py 
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
@dataclass(frozen=True)
class Host:
    """A network host with optional primary IP, MAC, and default route.

    IPv4 lives in ``ip`` / ``default_route``; IPv6 lives in ``ip6`` /
    ``default_route6``. Either, neither, or both may be set.
    Addresses and gateways must be strings or ``None``.
    """

    name: str
    ip: str | None = None
    mac: str | None = None
    default_route: str | None = None
    ip6: str | None = None
    default_route6: str | None = None

    def __post_init__(self) -> None:
        if not isinstance(self.name, str) or not NAME_RE.fullmatch(self.name):
            raise TopologyError(f"invalid host name {self.name!r}: must match {NAME_RE.pattern}")
        for field in ("ip", "ip6", "default_route", "default_route6"):
            value = getattr(self, field)
            if value is not None and not isinstance(value, str):
                raise TopologyError(
                    f"host {self.name!r}: {field} must be a string or None, got {value!r}"
                )
        if self.ip is not None:
            try:
                ipaddress.IPv4Interface(self.ip)
            except (ValueError, ipaddress.AddressValueError, ipaddress.NetmaskValueError) as exc:
                raise TopologyError(
                    f"invalid host IP {self.ip!r}: must be an IPv4 CIDR (e.g. '10.0.0.1/24')"
                ) from exc
        if self.mac is not None and (
            not isinstance(self.mac, str) or not _MAC_RE.fullmatch(self.mac)
        ):
            raise TopologyError(f"invalid host MAC {self.mac!r}: must be 'XX:XX:XX:XX:XX:XX'")
        if self.default_route is not None:
            if self.ip is None:
                raise TopologyError(f"host {self.name!r}: default_route requires ip to be set")
            try:
                ipaddress.IPv4Address(self.default_route)
            except (ValueError, ipaddress.AddressValueError) as exc:
                raise TopologyError(
                    f"invalid default_route {self.default_route!r}: must be an IPv4 address"
                ) from exc
        if self.ip6 is not None:
            if isinstance(self.ip6, str) and "." in self.ip6:
                raise TopologyError(
                    f"invalid host ip6 {self.ip6!r}: must be an IPv6 CIDR (e.g. 'fd00::1/64')"
                )
            try:
                ipaddress.IPv6Interface(self.ip6)
            except (ValueError, ipaddress.AddressValueError, ipaddress.NetmaskValueError) as exc:
                raise TopologyError(
                    f"invalid host ip6 {self.ip6!r}: must be an IPv6 CIDR (e.g. 'fd00::1/64')"
                ) from exc
        if self.default_route6 is not None:
            if self.ip6 is None:
                raise TopologyError(f"host {self.name!r}: default_route6 requires ip6 to be set")
            try:
                ipaddress.IPv6Address(self.default_route6)
            except (ValueError, ipaddress.AddressValueError) as exc:
                raise TopologyError(
                    f"invalid default_route6 {self.default_route6!r}: must be an IPv6 address"
                ) from exc

A bidirectional veth pair between two nodes, with optional impairments.

Endpoint node references must be non-empty strings.

Symmetric impairments (bandwidth / delay / jitter / loss_pct) apply equally to both veth sides. Per-direction overrides (*_a_to_b / *_b_to_a) shape only one direction; mixing a symmetric field and a matching asymmetric field for the same parameter is rejected.

Per-direction additive fields (*_a_to_b_extra / *_b_to_a_extra) layer on top of the symmetric base. They require the symmetric base to be set and are mutually exclusive with the per-direction override on the same dimension. Bandwidth has no _extra form because additive bandwidth lacks clean physical semantics.

Source code in src/p4net/topo/link.py 
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
@dataclass(frozen=True)
class Link:
    """A bidirectional veth pair between two nodes, with optional impairments.

    Endpoint node references must be non-empty strings.

    Symmetric impairments (``bandwidth`` / ``delay`` / ``jitter`` /
    ``loss_pct``) apply equally to both veth sides. Per-direction overrides
    (``*_a_to_b`` / ``*_b_to_a``) shape only one direction; mixing a symmetric
    field and a matching asymmetric field for the same parameter is rejected.

    Per-direction additive fields (``*_a_to_b_extra`` / ``*_b_to_a_extra``)
    layer on top of the symmetric base. They require the symmetric base to be
    set and are mutually exclusive with the per-direction override on the
    same dimension. Bandwidth has no ``_extra`` form because additive
    bandwidth lacks clean physical semantics.
    """

    a: LinkEndpoint
    b: LinkEndpoint
    bandwidth: str | None = None
    delay: str | None = None
    jitter: str | None = None
    loss_pct: float | None = None
    mtu: int | None = None
    bandwidth_a_to_b: str | None = None
    bandwidth_b_to_a: str | None = None
    delay_a_to_b: str | None = None
    delay_b_to_a: str | None = None
    jitter_a_to_b: str | None = None
    jitter_b_to_a: str | None = None
    loss_pct_a_to_b: float | None = None
    loss_pct_b_to_a: float | None = None
    delay_a_to_b_extra: str | None = None
    delay_b_to_a_extra: str | None = None
    jitter_a_to_b_extra: str | None = None
    jitter_b_to_a_extra: str | None = None
    loss_pct_a_to_b_extra: float | None = None
    loss_pct_b_to_a_extra: float | None = None

    def __post_init__(self) -> None:
        if not isinstance(self.a, LinkEndpoint):
            raise TopologyError("Link.a must be a LinkEndpoint")
        if not isinstance(self.b, LinkEndpoint):
            raise TopologyError("Link.b must be a LinkEndpoint")
        if not isinstance(self.a.node, str) or not self.a.node:
            raise TopologyError("Link.a.node must be a non-empty string")
        if not isinstance(self.b.node, str) or not self.b.node:
            raise TopologyError("Link.b.node must be a non-empty string")
        for field in (
            "loss_pct",
            "loss_pct_a_to_b",
            "loss_pct_b_to_a",
            "loss_pct_a_to_b_extra",
            "loss_pct_b_to_a_extra",
        ):
            value = getattr(self, field)
            if value is None:
                continue
            try:
                valid = (
                    isinstance(value, (int, float))
                    and not isinstance(value, bool)
                    and math.isfinite(value)
                )
            except OverflowError:
                valid = False
            if not valid:
                raise TopologyError(
                    f"link {field} {value!r} must be a finite int or float (bool is not accepted)"
                )
        for param, sym, asym in (
            ("bandwidth", self.bandwidth, (self.bandwidth_a_to_b, self.bandwidth_b_to_a)),
            ("delay", self.delay, (self.delay_a_to_b, self.delay_b_to_a)),
            ("jitter", self.jitter, (self.jitter_a_to_b, self.jitter_b_to_a)),
            ("loss_pct", self.loss_pct, (self.loss_pct_a_to_b, self.loss_pct_b_to_a)),
        ):
            if sym is not None and any(a is not None for a in asym):
                raise TopologyError(f"link sets both {param} and {param}_<dir>; pick one")
        if self.jitter is not None and self.delay is None:
            raise TopologyError("link jitter requires delay to be set")
        if self.jitter_a_to_b is not None and self.delay_a_to_b is None and self.delay is None:
            raise TopologyError("link jitter_a_to_b requires delay_a_to_b or delay to be set")
        if self.jitter_b_to_a is not None and self.delay_b_to_a is None and self.delay is None:
            raise TopologyError("link jitter_b_to_a requires delay_b_to_a or delay to be set")
        if self.loss_pct is not None and not 0.0 <= self.loss_pct <= 100.0:
            raise TopologyError(f"link loss_pct {self.loss_pct} out of range [0.0, 100.0]")
        for direction, value in (
            ("a_to_b", self.loss_pct_a_to_b),
            ("b_to_a", self.loss_pct_b_to_a),
        ):
            if value is not None and not 0.0 <= value <= 100.0:
                raise TopologyError(f"link loss_pct_{direction} {value} out of range [0.0, 100.0]")
        for param, sym, per_dir, extras in (
            (
                "delay",
                self.delay,
                (self.delay_a_to_b, self.delay_b_to_a),
                (self.delay_a_to_b_extra, self.delay_b_to_a_extra),
            ),
            (
                "jitter",
                self.jitter,
                (self.jitter_a_to_b, self.jitter_b_to_a),
                (self.jitter_a_to_b_extra, self.jitter_b_to_a_extra),
            ),
            (
                "loss_pct",
                self.loss_pct,
                (self.loss_pct_a_to_b, self.loss_pct_b_to_a),
                (self.loss_pct_a_to_b_extra, self.loss_pct_b_to_a_extra),
            ),
        ):
            for direction, extra, per in zip(("a_to_b", "b_to_a"), extras, per_dir, strict=True):
                if extra is None:
                    continue
                if per is not None:
                    raise TopologyError(
                        f"link sets both {param}_{direction} and {param}_{direction}_extra; "
                        f"pick one"
                    )
                if sym is None:
                    raise TopologyError(
                        f"link {param}_{direction}_extra requires symmetric {param} to be set"
                    )
        for direction, value in (
            ("a_to_b", self.loss_pct_a_to_b_extra),
            ("b_to_a", self.loss_pct_b_to_a_extra),
        ):
            if value is not None and value < 0.0:
                raise TopologyError(f"link loss_pct_{direction}_extra {value} must be non-negative")
        if self.mtu is not None:
            if isinstance(self.mtu, bool) or not isinstance(self.mtu, int):
                raise TopologyError(f"link mtu {self.mtu!r} must be an int (bool is not accepted)")
            if not _MTU_MIN <= self.mtu <= _MTU_MAX:
                raise TopologyError(f"link mtu {self.mtu} out of range [{_MTU_MIN}, {_MTU_MAX}]")

LinkEndpoint dataclass

One end of a Link, attached to a host or switch.

ip and ip6 are link-level overrides that apply only on host endpoints; switch endpoints don't carry L3 addresses. Address overrides must be strings or None.

Source code in src/p4net/topo/link.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
@dataclass(frozen=True)
class LinkEndpoint:
    """One end of a `Link`, attached to a host or switch.

    ``ip`` and ``ip6`` are link-level overrides that apply only on host
    endpoints; switch endpoints don't carry L3 addresses.
    Address overrides must be strings or ``None``.
    """

    node: str
    port: int | None = None
    iface_name: str | None = None
    ip: str | None = None
    mac: str | None = None
    ip6: str | None = None

    def __post_init__(self) -> None:
        if self.port is not None:
            _validate_port_number(self.port)
        for field in ("ip", "ip6"):
            value = getattr(self, field)
            if value is not None and not isinstance(value, str):
                raise TopologyError(
                    f"LinkEndpoint {self.node!r}: {field} must be a string or None, got {value!r}"
                )
        if self.mac is not None and (
            not isinstance(self.mac, str) or not _MAC_RE.fullmatch(self.mac)
        ):
            raise TopologyError(
                f"invalid LinkEndpoint MAC {self.mac!r}: must be 'XX:XX:XX:XX:XX:XX'"
            )
        if self.ip6 is not None:
            if isinstance(self.ip6, str) and "." in self.ip6:
                raise TopologyError(f"invalid LinkEndpoint ip6 {self.ip6!r}: must be an IPv6 CIDR")
            try:
                ipaddress.IPv6Interface(self.ip6)
            except (ValueError, ipaddress.AddressValueError, ipaddress.NetmaskValueError) as exc:
                raise TopologyError(
                    f"invalid LinkEndpoint ip6 {self.ip6!r}: must be an IPv6 CIDR"
                ) from exc

P4Switch dataclass

A P4-programmable switch backed by BMv2 simple_switch_grpc.

Device IDs and port numbers accept ints, excluding bool, or None.

Source code in src/p4net/topo/switch.py 
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
@dataclass(frozen=True)
class P4Switch:
    """A P4-programmable switch backed by BMv2 simple_switch_grpc.

    Device IDs and port numbers accept ints, excluding bool, or None.
    """

    name: str
    p4_src: Path
    arch: str = "v1model"
    device_id: int | None = None
    grpc_port: int | None = None
    thrift_port: int | None = None
    cpu_port: int | None = None
    log_level: str = "info"
    pcap_enabled: bool = True

    def __post_init__(self) -> None:
        if not isinstance(self.name, str) or not NAME_RE.fullmatch(self.name):
            raise TopologyError(f"invalid switch name {self.name!r}: must match {NAME_RE.pattern}")
        # Coerce p4_src to Path (frozen dataclass: bypass via object.__setattr__).
        if not isinstance(self.p4_src, Path):
            object.__setattr__(self, "p4_src", Path(self.p4_src))
        if self.p4_src.suffix != ".p4":
            raise TopologyError(f"p4_src {str(self.p4_src)!r}: file must have .p4 suffix")
        if self.arch != "v1model":
            raise TopologyError(
                f"unsupported arch {self.arch!r}: only 'v1model' is supported in this release"
            )
        for field_name, value in (
            ("device_id", self.device_id),
            ("grpc_port", self.grpc_port),
            ("thrift_port", self.thrift_port),
            ("cpu_port", self.cpu_port),
        ):
            if value is not None and (isinstance(value, bool) or not isinstance(value, int)):
                raise TopologyError(f"{field_name} {value!r} must be an int (bool is not accepted)")
        if self.device_id is not None and not 0 <= self.device_id < _DEVICE_ID_MAX:
            raise TopologyError(f"device_id {self.device_id} out of range [0, {_DEVICE_ID_MAX})")
        if self.grpc_port is not None and not _PORT_MIN <= self.grpc_port <= _PORT_MAX:
            raise TopologyError(
                f"grpc_port {self.grpc_port} out of range [{_PORT_MIN}, {_PORT_MAX}]"
            )
        if self.thrift_port is not None and not _PORT_MIN <= self.thrift_port <= _PORT_MAX:
            raise TopologyError(
                f"thrift_port {self.thrift_port} out of range [{_PORT_MIN}, {_PORT_MAX}]"
            )
        if self.cpu_port is not None and not _CPU_PORT_MIN <= self.cpu_port <= _CPU_PORT_MAX:
            raise TopologyError(
                f"cpu_port {self.cpu_port} out of range [{_CPU_PORT_MIN}, {_CPU_PORT_MAX}]"
            )
        if not isinstance(self.log_level, str) or self.log_level not in _VALID_LOG_LEVELS:
            raise TopologyError(
                f"invalid log_level {self.log_level!r}: must be one of {sorted(_VALID_LOG_LEVELS)}"
            )
        if not isinstance(self.pcap_enabled, bool):
            raise TopologyError(f"pcap_enabled {self.pcap_enabled!r} must be a bool")

Topology

A mutable builder for a topology description.

Source code in src/p4net/topo/topology.py 
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
class Topology:
    """A mutable builder for a topology description."""

    def __init__(self) -> None:
        self._hosts: dict[str, Host] = {}
        self._switches: dict[str, P4Switch] = {}
        self._links: list[Link] = []

    @property
    def hosts(self) -> Mapping[str, Host]:
        """Map of host name → :class:`Host`."""
        return self._hosts

    @property
    def switches(self) -> Mapping[str, P4Switch]:
        """Map of switch name → :class:`P4Switch`."""
        return self._switches

    @property
    def links(self) -> Sequence[Link]:
        """Tuple of every :class:`Link` in declaration order."""
        return tuple(self._links)

    def node(self, name: str) -> NodeKind:
        """Look up a node by name.

        Returns:
            The :class:`Host` or :class:`P4Switch` named ``name``.

        Raises:
            TopologyError: if no node with that name exists.
        """
        if name in self._hosts:
            return self._hosts[name]
        if name in self._switches:
            return self._switches[name]
        raise TopologyError(f"no node named {name!r}")

    def add_host(
        self,
        name: str,
        *,
        ip: str | None = None,
        mac: str | None = None,
        default_route: str | None = None,
        ip6: str | None = None,
        default_route6: str | None = None,
    ) -> Host:
        """Append a :class:`Host` to the topology.

        Args:
            name: Host name; must be unique across hosts and switches.
            ip: IPv4 CIDR (e.g. ``"10.0.0.1/24"``).
            mac: MAC address (e.g. ``"00:00:00:00:00:01"``).
            default_route: IPv4 default-route gateway address.
            ip6: IPv6 CIDR (e.g. ``"fd00::1/64"``).
            default_route6: IPv6 default-route gateway address.

        Returns:
            The newly created :class:`Host`.
        """
        self._reject_existing_name(name)
        host = Host(
            name=name,
            ip=ip,
            mac=mac,
            default_route=default_route,
            ip6=ip6,
            default_route6=default_route6,
        )
        self._hosts[name] = host
        return host

    def add_switch(
        self,
        name: str,
        p4_src: Path,
        *,
        arch: str = "v1model",
        device_id: int | None = None,
        grpc_port: int | None = None,
        thrift_port: int | None = None,
        cpu_port: int | None = None,
        log_level: str = "info",
        pcap_enabled: bool = True,
    ) -> P4Switch:
        """Append a :class:`P4Switch` to the topology.

        Args:
            name: Switch name; must be unique across hosts and switches.
            p4_src: Path to the P4 source file the switch should run.
            arch: P4 architecture name; only ``"v1model"`` is supported.
            device_id: P4Runtime device ID. Auto-assigned starting at 0.
            grpc_port: gRPC bind port. Auto-assigned starting at 50051.
            thrift_port: Thrift bind port. Auto-assigned starting at 9090.
            cpu_port: CPU port number for controller punt (optional).
            log_level: BMv2 log level passed via ``--log-level``.
            pcap_enabled: Capture this switch's ports (default True). Requires a
                directory via ``Network(pcap_dir=...)`` or CLI ``--pcap-dir``.

        Returns:
            The newly created :class:`P4Switch`.
        """
        self._reject_existing_name(name)
        idx = len(self._switches)
        if device_id is None:
            device_id = idx
        if grpc_port is None:
            grpc_port = _BASE_GRPC_PORT + idx
        if thrift_port is None:
            thrift_port = _BASE_THRIFT_PORT + idx
        switch = P4Switch(
            name=name,
            p4_src=p4_src,
            arch=arch,
            device_id=device_id,
            grpc_port=grpc_port,
            thrift_port=thrift_port,
            cpu_port=cpu_port,
            log_level=log_level,
            pcap_enabled=pcap_enabled,
        )
        self._switches[name] = switch
        return switch

    def add_link(
        self,
        a: NodeRef,
        b: NodeRef,
        *,
        port_a: int | None = None,
        port_b: int | None = None,
        ip_a: str | None = None,
        ip_b: str | None = None,
        mac_a: str | None = None,
        mac_b: str | None = None,
        ip6_a: str | None = None,
        ip6_b: str | None = None,
        bandwidth: str | None = None,
        delay: str | None = None,
        jitter: str | None = None,
        loss_pct: float | None = None,
        mtu: int | None = None,
        bandwidth_a_to_b: str | None = None,
        bandwidth_b_to_a: str | None = None,
        delay_a_to_b: str | None = None,
        delay_b_to_a: str | None = None,
        jitter_a_to_b: str | None = None,
        jitter_b_to_a: str | None = None,
        loss_pct_a_to_b: float | None = None,
        loss_pct_b_to_a: float | None = None,
        delay_a_to_b_extra: str | None = None,
        delay_b_to_a_extra: str | None = None,
        jitter_a_to_b_extra: str | None = None,
        jitter_b_to_a_extra: str | None = None,
        loss_pct_a_to_b_extra: float | None = None,
        loss_pct_b_to_a_extra: float | None = None,
    ) -> Link:
        """Append a :class:`Link` between two nodes.

        Endpoint sides ``a`` and ``b`` are anchored by argument order;
        per-direction impairment fields like ``delay_a_to_b`` shape only
        the direction from the ``a``-side veth toward the ``b`` side.

        Ports must be non-negative ints, excluding bool. Switch data ports
        range from 0 to 510. Automatic allocation starts at 1 for switches,
        skips the CPU port and raises :class:`TopologyError` when exhausted.

        Args:
            a: First endpoint (host or switch name, or a node object).
            b: Second endpoint.
            port_a: Port number on the ``a`` side. Auto-assigned if omitted.
            port_b: Port number on the ``b`` side. Auto-assigned if omitted.
            ip_a: IPv4 CIDR override for the ``a``-side interface.
            ip_b: IPv4 CIDR override for the ``b``-side interface.
            mac_a: MAC override for the ``a``-side interface.
            mac_b: MAC override for the ``b``-side interface.
            ip6_a: IPv6 CIDR override for the ``a``-side interface.
            ip6_b: IPv6 CIDR override for the ``b``-side interface.
            bandwidth: Symmetric link-rate cap (e.g. ``"10mbit"``).
            delay: Symmetric one-way delay (e.g. ``"50ms"``).
            jitter: Symmetric jitter; requires ``delay`` to be set.
            loss_pct: Symmetric per-packet loss in [0.0, 100.0].
            mtu: Link MTU, an int in [68, 65535], excluding bool, or None.
            bandwidth_a_to_b: Per-direction bandwidth, ``a`` → ``b``.
            bandwidth_b_to_a: Per-direction bandwidth, ``b`` → ``a``.
            delay_a_to_b: Per-direction delay, ``a`` → ``b``.
            delay_b_to_a: Per-direction delay, ``b`` → ``a``.
            jitter_a_to_b: Per-direction jitter, ``a`` → ``b``.
            jitter_b_to_a: Per-direction jitter, ``b`` → ``a``.
            loss_pct_a_to_b: Per-direction loss, ``a`` → ``b``.
            loss_pct_b_to_a: Per-direction loss, ``b`` → ``a``.
            delay_a_to_b_extra: Additional delay added on top of the symmetric
                ``delay`` for the ``a`` → ``b`` direction. Requires symmetric
                ``delay``; mutually exclusive with ``delay_a_to_b``.
            delay_b_to_a_extra: Same as above, ``b`` → ``a``.
            jitter_a_to_b_extra: Additional jitter on top of symmetric
                ``jitter`` for ``a`` → ``b``. Requires symmetric ``jitter``.
            jitter_b_to_a_extra: Same as above, ``b`` → ``a``.
            loss_pct_a_to_b_extra: Additional loss percent on top of symmetric
                ``loss_pct`` for ``a`` → ``b``. Requires symmetric
                ``loss_pct``; combined value must stay ≤ 100.0.
            loss_pct_b_to_a_extra: Same as above, ``b`` → ``a``.

        Returns:
            The newly created :class:`Link`.

        Raises:
            TopologyError: on invalid parameter combinations or
                unresolved endpoints.
        """
        node_a = self._resolve(a)
        node_b = self._resolve(b)
        port_a_val = self._auto_port(node_a, port_a)
        port_b_val = self._auto_port(node_b, port_b)
        iface_a = _iface_name(node_a.name, port_a_val)
        iface_b = _iface_name(node_b.name, port_b_val)
        for name in (iface_a, iface_b):
            try:
                _validate_ifname(name)
            except ValueError as exc:
                raise TopologyError(f"{exc}; shorten the node name") from exc
        if ip_a is not None and isinstance(node_a, P4Switch):
            raise TopologyError(
                "P4 switch data ports do not carry IP addresses; remove ip_a from the add_link call"
            )
        if ip_b is not None and isinstance(node_b, P4Switch):
            raise TopologyError(
                "P4 switch data ports do not carry IP addresses; remove ip_b from the add_link call"
            )
        if ip6_a is not None and isinstance(node_a, P4Switch):
            raise TopologyError(
                "P4 switch data ports do not carry IP addresses; "
                "remove ip6_a from the add_link call"
            )
        if ip6_b is not None and isinstance(node_b, P4Switch):
            raise TopologyError(
                "P4 switch data ports do not carry IP addresses; "
                "remove ip6_b from the add_link call"
            )
        ep_a = LinkEndpoint(
            node=node_a.name,
            port=port_a_val,
            iface_name=iface_a,
            ip=ip_a,
            mac=mac_a,
            ip6=ip6_a,
        )
        ep_b = LinkEndpoint(
            node=node_b.name,
            port=port_b_val,
            iface_name=iface_b,
            ip=ip_b,
            mac=mac_b,
            ip6=ip6_b,
        )
        link = Link(
            a=ep_a,
            b=ep_b,
            bandwidth=bandwidth,
            delay=delay,
            jitter=jitter,
            loss_pct=loss_pct,
            mtu=mtu,
            bandwidth_a_to_b=bandwidth_a_to_b,
            bandwidth_b_to_a=bandwidth_b_to_a,
            delay_a_to_b=delay_a_to_b,
            delay_b_to_a=delay_b_to_a,
            jitter_a_to_b=jitter_a_to_b,
            jitter_b_to_a=jitter_b_to_a,
            loss_pct_a_to_b=loss_pct_a_to_b,
            loss_pct_b_to_a=loss_pct_b_to_a,
            delay_a_to_b_extra=delay_a_to_b_extra,
            delay_b_to_a_extra=delay_b_to_a_extra,
            jitter_a_to_b_extra=jitter_a_to_b_extra,
            jitter_b_to_a_extra=jitter_b_to_a_extra,
            loss_pct_a_to_b_extra=loss_pct_a_to_b_extra,
            loss_pct_b_to_a_extra=loss_pct_b_to_a_extra,
        )
        self._links.append(link)
        return link

    def neighbors_of(self, name: str) -> list[tuple[Link, LinkEndpoint, LinkEndpoint]]:
        """Return every link involving `name`, as (link, near, far)."""
        result: list[tuple[Link, LinkEndpoint, LinkEndpoint]] = []
        for link in self._links:
            if link.a.node == name:
                result.append((link, link.a, link.b))
            elif link.b.node == name:
                result.append((link, link.b, link.a))
        return result

    def port_assignments(self, switch_name: str) -> dict[int, LinkEndpoint]:
        """Map of switch port number to the FAR endpoint connected on that port."""
        if switch_name not in self._switches:
            raise TopologyError(f"no switch named {switch_name!r}")
        result: dict[int, LinkEndpoint] = {}
        for link in self._links:
            if link.a.node == switch_name and link.a.port is not None:
                result[link.a.port] = link.b
            elif link.b.node == switch_name and link.b.port is not None:
                result[link.b.port] = link.a
        return result

    def validate(self) -> None:
        """Raise `TopologyError` listing every internal-consistency problem found.

        Node names must be unique across hosts and switches, and mapping keys
        must match their descriptions' names.
        All configured gRPC and Thrift ports must be distinct across the topology,
        including the two ports of each switch. Every link endpoint must have
        an assigned port and a non-empty interface name. Different hosts cannot
        use the same effective IP address, regardless of prefix length.
        Switch data ports cannot carry IPv4 or IPv6 addresses.
        """
        errors: list[str] = []

        for kind, mapping in (("host", self._hosts), ("switch", self._switches)):
            for node_key, node in mapping.items():
                if node_key != node.name:
                    errors.append(f"{kind} key {node_key!r} does not match node name {node.name!r}")
        for node_key in self._hosts:
            if node_key in self._switches:
                errors.append(f"node name {node_key!r} is used by both a host and a switch")

        # 1. Endpoint references resolve to a known node.
        # 2. No self-loops.
        for i, link in enumerate(self._links):
            for which, ep in (("a", link.a), ("b", link.b)):
                if ep.node not in self._hosts and ep.node not in self._switches:
                    errors.append(f"link[{i}] endpoint {which} references unknown node {ep.node!r}")
                if ep.port is None:
                    errors.append(f"link[{i}] endpoint {which} ({ep.node!r}) has no assigned port")
                switch = self._switches.get(ep.node)
                if switch is not None:
                    for field, address in (("ip", ep.ip), ("ip6", ep.ip6)):
                        if address is not None:
                            errors.append(
                                f"link[{i}] endpoint {which} switch {switch.name!r}: "
                                "P4 switch data ports do not carry IP addresses "
                                f"({field}={address!r})"
                            )
                if switch is not None and ep.port is not None and ep.port > _SWITCH_PORT_MAX:
                    errors.append(
                        f"link[{i}] endpoint {which} switch port {ep.port} "
                        f"out of range [0, {_SWITCH_PORT_MAX}]"
                    )
                if (
                    switch is not None
                    and switch.cpu_port is not None
                    and ep.port == switch.cpu_port
                ):
                    errors.append(
                        f"link[{i}] endpoint {which} uses CPU port {switch.cpu_port} "
                        f"on switch {switch.name!r}"
                    )
            if link.a.node == link.b.node:
                errors.append(f"link[{i}] connects node {link.a.node!r} to itself")

        # 3. No (node, port) pair repeats across links.
        seen: dict[tuple[str, int], int] = {}
        for i, link in enumerate(self._links):
            for _which, ep in (("a", link.a), ("b", link.b)):
                if ep.port is None:
                    continue
                key = (ep.node, ep.port)
                if key in seen:
                    errors.append(
                        f"port collision: ({ep.node!r}, port {ep.port}) used by "
                        f"link[{seen[key]}] and link[{i}]"
                    )
                else:
                    seen[key] = i

        # 4. Device IDs are unique; gRPC and Thrift share the TCP port space.
        device_owners: dict[int, str] = {}
        port_owners: dict[int, tuple[str, str]] = {}
        for sw in self._switches.values():
            if sw.device_id is not None:
                if sw.device_id in device_owners:
                    errors.append(
                        f"switch device_id collision: {sw.name!r} and "
                        f"{device_owners[sw.device_id]!r} both use {sw.device_id}"
                    )
                else:
                    device_owners[sw.device_id] = sw.name
            for field_name, port in (("grpc_port", sw.grpc_port), ("thrift_port", sw.thrift_port)):
                if port is None:
                    continue
                if port in port_owners:
                    owner, owner_field = port_owners[port]
                    label = field_name if field_name == owner_field else "service port"
                    errors.append(
                        f"switch {label} collision: {sw.name!r} ({field_name}) and "
                        f"{owner!r} ({owner_field}) both use {port}"
                    )
                else:
                    port_owners[port] = (sw.name, field_name)

        # 5. Interface names are valid and unique within their final namespaces.
        interface_owners: dict[tuple[str | None, str], str] = {}
        for i, link in enumerate(self._links):
            for which, ep in (("a", link.a), ("b", link.b)):
                if ep.iface_name is None or ep.iface_name == "":
                    errors.append(
                        f"link[{i}] endpoint {which} ({ep.node!r}) has no assigned interface name"
                    )
                    continue
                try:
                    _validate_ifname(ep.iface_name)
                except ValueError as exc:
                    errors.append(f"link[{i}] endpoint {which} ({ep.node!r}): {exc}")
                    continue
                owner = f"link[{i}] endpoint {which} ({ep.node!r})"
                if ep.iface_name == "lo":
                    errors.append(f"{owner}: interface name 'lo' is reserved for loopback")
                    continue
                if ep.node in self._hosts:
                    scope: str | None = ep.node
                    label = f"host {ep.node!r}"
                elif ep.node in self._switches:
                    scope = None
                    label = "root namespace"
                else:
                    continue
                interface_key = (scope, ep.iface_name)
                if interface_key in interface_owners:
                    errors.append(
                        f"interface name collision in {label}: {ep.iface_name!r} "
                        f"used by {interface_owners[interface_key]} and {owner}"
                    )
                else:
                    interface_owners[interface_key] = owner

        # 6. Different hosts cannot share an effective address, regardless of prefix.
        ipv4_owners: dict[ipaddress.IPv4Address, set[str]] = {}
        ipv6_owners: dict[ipaddress.IPv6Address, set[str]] = {}
        seen_hosts: set[str] = set()
        for link in self._links:
            for ep in (link.a, link.b):
                host = self._hosts.get(ep.node)
                if host is None:
                    continue
                ipv4, ipv6 = _effective_host_addresses(
                    host, ep, first_link=host.name not in seen_hosts
                )
                seen_hosts.add(host.name)
                if ipv4 is not None:
                    try:
                        iface = ipaddress.IPv4Interface(ipv4)
                    except (ValueError, ipaddress.AddressValueError, ipaddress.NetmaskValueError):
                        errors.append(f"link endpoint {ep.node!r}: invalid IP {ipv4!r}")
                    else:
                        ipv4_owners.setdefault(iface.ip, set()).add(host.name)
                if ipv6 is not None:
                    try:
                        iface6 = ipaddress.IPv6Interface(ipv6)
                    except (ValueError, ipaddress.AddressValueError, ipaddress.NetmaskValueError):
                        errors.append(f"link endpoint {ep.node!r}: invalid ip6 {ipv6!r}")
                    else:
                        ipv6_owners.setdefault(iface6.ip, set()).add(host.name)
        for addr, nodes in ipv4_owners.items():
            if len(nodes) > 1:
                errors.append(f"IP collision: address {addr} used by hosts {sorted(nodes)}")

        for addr6, nodes6 in ipv6_owners.items():
            if len(nodes6) > 1:
                errors.append(f"IPv6 collision: address {addr6} used by hosts {sorted(nodes6)}")

        if errors:
            raise TopologyError(
                "topology validation failed with the following problems:\n  - "
                + "\n  - ".join(errors)
            )

    def to_graphviz(self, *, layout: str = "LR") -> str:
        """Render the topology as a Graphviz DOT graph string.

        Hosts are ellipses labelled with name and primary IP(s); switches are
        boxes labelled with name and gRPC port. Edges are drawn with
        ``arrowhead=none`` to keep the rendering version-stable across
        graphviz releases. ``layout`` controls ``rankdir`` and must be one
        of ``"LR"``, ``"RL"``, ``"TB"``, ``"BT"``.
        """
        if layout not in {"LR", "RL", "TB", "BT"}:
            raise TopologyError(f"layout {layout!r} must be one of 'LR', 'RL', 'TB', 'BT'")
        lines: list[str] = ["digraph p4net {"]
        lines.append(f"  rankdir={layout};")
        lines.append('  node [fontname="monospace"];')
        for host in self._hosts.values():
            label_parts = [host.name]
            if host.ip is not None:
                label_parts.append(host.ip)
            if host.ip6 is not None:
                label_parts.append(host.ip6)
            label = "\\n".join(label_parts)
            lines.append(f'  "{host.name}" [shape=ellipse, label="{label}"];')
        for sw in self._switches.values():
            label_parts = [sw.name]
            if sw.grpc_port is not None:
                label_parts.append(f"grpc :{sw.grpc_port}")
            label = "\\n".join(label_parts)
            lines.append(f'  "{sw.name}" [shape=box, label="{label}"];')
        for link in self._links:
            lines.append(f'  "{link.a.node}" -> "{link.b.node}" [arrowhead=none];')
        lines.append("}")
        return "\n".join(lines) + "\n"

    def render_graphviz(
        self,
        output_path: Path,
        *,
        layout: str = "LR",
        format: str = "png",
    ) -> None:
        """Render via the system ``dot`` binary to ``output_path``.

        ``format`` is forwarded to ``dot -T<format>`` (png, svg, pdf, dot).
        For ``format="dot"`` the source is written verbatim and ``dot`` is
        not invoked, so this path works without graphviz installed.
        Raises :class:`TopologyError` if ``dot`` is missing or the render
        fails.
        """
        import shutil
        import subprocess

        source = self.to_graphviz(layout=layout)
        output_path = Path(output_path)
        output_path.parent.mkdir(parents=True, exist_ok=True)
        if format == "dot":
            output_path.write_text(source)
            return
        dot_bin = shutil.which("dot")
        if dot_bin is None:
            raise TopologyError(
                "graphviz `dot` binary not found on PATH; "
                "install graphviz or use format='dot' to write the source file directly"
            )
        try:
            subprocess.run(
                [dot_bin, f"-T{format}", "-o", str(output_path)],
                input=source.encode("utf-8"),
                check=True,
                capture_output=True,
            )
        except subprocess.CalledProcessError as exc:
            stderr = (exc.stderr or b"").decode("utf-8", errors="replace")
            raise TopologyError(
                f"`dot -T{format}` failed (rc={exc.returncode}): {stderr.strip()}"
            ) from exc

    def to_dict(self) -> dict[str, Any]:
        """Return a JSON-serialisable snapshot of the topology."""
        return {
            "hosts": {name: _host_to_dict(host) for name, host in self._hosts.items()},
            "switches": {name: _switch_to_dict(sw) for name, sw in self._switches.items()},
            "links": [_link_to_dict(link) for link in self._links],
        }

    @classmethod
    def from_dict(cls, data: dict[str, Any]) -> Topology:
        """Reconstruct a :class:`Topology` from a :meth:`to_dict` payload.

        Node keys, names, endpoint ports, interface names, and address overrides
        are restored as stored. Call :meth:`validate` to check node identity and
        endpoint assignments, and reject IP addresses on switch data ports.
        """
        topo = cls()
        for name, host_data in data.get("hosts", {}).items():
            topo._hosts[name] = Host(
                name=host_data["name"],
                ip=host_data.get("ip"),
                mac=host_data.get("mac"),
                default_route=host_data.get("default_route"),
                ip6=host_data.get("ip6"),
                default_route6=host_data.get("default_route6"),
            )
        for name, sw_data in data.get("switches", {}).items():
            topo._switches[name] = P4Switch(
                name=sw_data["name"],
                p4_src=Path(sw_data["p4_src"]),
                arch=sw_data.get("arch", "v1model"),
                device_id=sw_data.get("device_id"),
                grpc_port=sw_data.get("grpc_port"),
                thrift_port=sw_data.get("thrift_port"),
                cpu_port=sw_data.get("cpu_port"),
                log_level=sw_data.get("log_level", "info"),
                pcap_enabled=sw_data.get("pcap_enabled", True),
            )
        for link_data in data.get("links", []):
            ep_a = LinkEndpoint(
                node=link_data["a"]["node"],
                port=link_data["a"].get("port"),
                iface_name=link_data["a"].get("iface_name"),
                ip=link_data["a"].get("ip"),
                mac=link_data["a"].get("mac"),
                ip6=link_data["a"].get("ip6"),
            )
            ep_b = LinkEndpoint(
                node=link_data["b"]["node"],
                port=link_data["b"].get("port"),
                iface_name=link_data["b"].get("iface_name"),
                ip=link_data["b"].get("ip"),
                mac=link_data["b"].get("mac"),
                ip6=link_data["b"].get("ip6"),
            )
            topo._links.append(
                Link(
                    a=ep_a,
                    b=ep_b,
                    bandwidth=link_data.get("bandwidth"),
                    delay=link_data.get("delay"),
                    jitter=link_data.get("jitter"),
                    loss_pct=link_data.get("loss_pct"),
                    mtu=link_data.get("mtu"),
                    bandwidth_a_to_b=link_data.get("bandwidth_a_to_b"),
                    bandwidth_b_to_a=link_data.get("bandwidth_b_to_a"),
                    delay_a_to_b=link_data.get("delay_a_to_b"),
                    delay_b_to_a=link_data.get("delay_b_to_a"),
                    jitter_a_to_b=link_data.get("jitter_a_to_b"),
                    jitter_b_to_a=link_data.get("jitter_b_to_a"),
                    loss_pct_a_to_b=link_data.get("loss_pct_a_to_b"),
                    loss_pct_b_to_a=link_data.get("loss_pct_b_to_a"),
                    delay_a_to_b_extra=link_data.get("delay_a_to_b_extra"),
                    delay_b_to_a_extra=link_data.get("delay_b_to_a_extra"),
                    jitter_a_to_b_extra=link_data.get("jitter_a_to_b_extra"),
                    jitter_b_to_a_extra=link_data.get("jitter_b_to_a_extra"),
                    loss_pct_a_to_b_extra=link_data.get("loss_pct_a_to_b_extra"),
                    loss_pct_b_to_a_extra=link_data.get("loss_pct_b_to_a_extra"),
                )
            )
        return topo

    # Internal helpers ---------------------------------------------------

    def _reject_existing_name(self, name: str) -> None:
        if name in self._hosts or name in self._switches:
            raise TopologyError(f"node name {name!r} already exists in this topology")

    def _resolve(self, ref: NodeRef) -> NodeKind:
        if isinstance(ref, Host):
            if self._hosts.get(ref.name) is not ref:
                raise TopologyError(f"host {ref.name!r} is not part of this topology")
            return ref
        if isinstance(ref, P4Switch):
            if self._switches.get(ref.name) is not ref:
                raise TopologyError(f"switch {ref.name!r} is not part of this topology")
            return ref
        if isinstance(ref, str):
            return self.node(ref)
        raise TopologyError(
            f"node reference must be a name string, Host, or P4Switch; got {type(ref).__name__}"
        )

    def _auto_port(self, node: NodeKind, requested: int | None) -> int:
        if requested is not None:
            _validate_port_number(requested)
            if isinstance(node, P4Switch) and requested > _SWITCH_PORT_MAX:
                raise TopologyError(
                    f"switch {node.name!r} port {requested} out of range [0, {_SWITCH_PORT_MAX}]"
                )
            if (
                isinstance(node, P4Switch)
                and node.cpu_port is not None
                and requested == node.cpu_port
            ):
                raise TopologyError(
                    f"switch {node.name!r} CPU port {requested} cannot be used by a link"
                )
            return requested
        used = {
            ep.port
            for link in self._links
            for ep in (link.a, link.b)
            if ep.node == node.name and ep.port is not None
        }
        if isinstance(node, P4Switch) and node.cpu_port is not None:
            used.add(node.cpu_port)
        candidate = 1 if isinstance(node, P4Switch) else 0
        while candidate in used:
            candidate += 1
        if isinstance(node, P4Switch) and candidate > _SWITCH_PORT_MAX:
            raise TopologyError(
                f"switch {node.name!r} has no free data ports in [1, {_SWITCH_PORT_MAX}]"
            )
        return candidate

hosts property 

hosts: Mapping[str, Host]

Map of host name → :class:Host.

switches property 

switches: Mapping[str, P4Switch]

Map of switch name → :class:P4Switch.

links: Sequence[Link]

Tuple of every :class:Link in declaration order.

node 

node(name: str) -> NodeKind

Look up a node by name.

Returns:

Name Type Description
The NodeKind

class:Host or :class:P4Switch named name.

Raises:

Type Description
TopologyError

if no node with that name exists.
Source code in src/p4net/topo/topology.py 
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def node(self, name: str) -> NodeKind:
    """Look up a node by name.

    Returns:
        The :class:`Host` or :class:`P4Switch` named ``name``.

    Raises:
        TopologyError: if no node with that name exists.
    """
    if name in self._hosts:
        return self._hosts[name]
    if name in self._switches:
        return self._switches[name]
    raise TopologyError(f"no node named {name!r}")

add_host 

add_host(name: str, *, ip: str | None = None, mac: str | None = None, default_route: str | None = None, ip6: str | None = None, default_route6: str | None = None) -> Host

Append a :class:Host to the topology.

Parameters:

Name Type Description Default
name str

Host name; must be unique across hosts and switches.

 required
ip str | None

IPv4 CIDR (e.g. "10.0.0.1/24").

 None
mac str | None

MAC address (e.g. "00:00:00:00:00:01").

 None
default_route str | None

IPv4 default-route gateway address.

 None
ip6 str | None

IPv6 CIDR (e.g. "fd00::1/64").

 None
default_route6 str | None

IPv6 default-route gateway address.

 None

Returns:

Type Description
Host

The newly created :class:Host.
Source code in src/p4net/topo/topology.py 
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def add_host(
    self,
    name: str,
    *,
    ip: str | None = None,
    mac: str | None = None,
    default_route: str | None = None,
    ip6: str | None = None,
    default_route6: str | None = None,
) -> Host:
    """Append a :class:`Host` to the topology.

    Args:
        name: Host name; must be unique across hosts and switches.
        ip: IPv4 CIDR (e.g. ``"10.0.0.1/24"``).
        mac: MAC address (e.g. ``"00:00:00:00:00:01"``).
        default_route: IPv4 default-route gateway address.
        ip6: IPv6 CIDR (e.g. ``"fd00::1/64"``).
        default_route6: IPv6 default-route gateway address.

    Returns:
        The newly created :class:`Host`.
    """
    self._reject_existing_name(name)
    host = Host(
        name=name,
        ip=ip,
        mac=mac,
        default_route=default_route,
        ip6=ip6,
        default_route6=default_route6,
    )
    self._hosts[name] = host
    return host

add_switch 

add_switch(name: str, p4_src: Path, *, arch: str = 'v1model', device_id: int | None = None, grpc_port: int | None = None, thrift_port: int | None = None, cpu_port: int | None = None, log_level: str = 'info', pcap_enabled: bool = True) -> P4Switch

Append a :class:P4Switch to the topology.

Parameters:

Name Type Description Default
name str

Switch name; must be unique across hosts and switches.

 required
p4_src Path

Path to the P4 source file the switch should run.

 required
arch str

P4 architecture name; only "v1model" is supported.

 'v1model'
device_id int | None

P4Runtime device ID. Auto-assigned starting at 0.

 None
grpc_port int | None

gRPC bind port. Auto-assigned starting at 50051.

 None
thrift_port int | None

Thrift bind port. Auto-assigned starting at 9090.

 None
cpu_port int | None

CPU port number for controller punt (optional).

 None
log_level str

BMv2 log level passed via --log-level.

 'info'
pcap_enabled bool

Capture this switch's ports (default True). Requires a directory via Network(pcap_dir=...) or CLI --pcap-dir.

 True

Returns:

Type Description
P4Switch

The newly created :class:P4Switch.
Source code in src/p4net/topo/topology.py 
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
def add_switch(
    self,
    name: str,
    p4_src: Path,
    *,
    arch: str = "v1model",
    device_id: int | None = None,
    grpc_port: int | None = None,
    thrift_port: int | None = None,
    cpu_port: int | None = None,
    log_level: str = "info",
    pcap_enabled: bool = True,
) -> P4Switch:
    """Append a :class:`P4Switch` to the topology.

    Args:
        name: Switch name; must be unique across hosts and switches.
        p4_src: Path to the P4 source file the switch should run.
        arch: P4 architecture name; only ``"v1model"`` is supported.
        device_id: P4Runtime device ID. Auto-assigned starting at 0.
        grpc_port: gRPC bind port. Auto-assigned starting at 50051.
        thrift_port: Thrift bind port. Auto-assigned starting at 9090.
        cpu_port: CPU port number for controller punt (optional).
        log_level: BMv2 log level passed via ``--log-level``.
        pcap_enabled: Capture this switch's ports (default True). Requires a
            directory via ``Network(pcap_dir=...)`` or CLI ``--pcap-dir``.

    Returns:
        The newly created :class:`P4Switch`.
    """
    self._reject_existing_name(name)
    idx = len(self._switches)
    if device_id is None:
        device_id = idx
    if grpc_port is None:
        grpc_port = _BASE_GRPC_PORT + idx
    if thrift_port is None:
        thrift_port = _BASE_THRIFT_PORT + idx
    switch = P4Switch(
        name=name,
        p4_src=p4_src,
        arch=arch,
        device_id=device_id,
        grpc_port=grpc_port,
        thrift_port=thrift_port,
        cpu_port=cpu_port,
        log_level=log_level,
        pcap_enabled=pcap_enabled,
    )
    self._switches[name] = switch
    return switch

add_link(a: NodeRef, b: NodeRef, *, port_a: int | None = None, port_b: int | None = None, ip_a: str | None = None, ip_b: str | None = None, mac_a: str | None = None, mac_b: str | None = None, ip6_a: str | None = None, ip6_b: str | None = None, bandwidth: str | None = None, delay: str | None = None, jitter: str | None = None, loss_pct: float | None = None, mtu: int | None = None, bandwidth_a_to_b: str | None = None, bandwidth_b_to_a: str | None = None, delay_a_to_b: str | None = None, delay_b_to_a: str | None = None, jitter_a_to_b: str | None = None, jitter_b_to_a: str | None = None, loss_pct_a_to_b: float | None = None, loss_pct_b_to_a: float | None = None, delay_a_to_b_extra: str | None = None, delay_b_to_a_extra: str | None = None, jitter_a_to_b_extra: str | None = None, jitter_b_to_a_extra: str | None = None, loss_pct_a_to_b_extra: float | None = None, loss_pct_b_to_a_extra: float | None = None) -> Link

Append a :class:Link between two nodes.

Endpoint sides a and b are anchored by argument order; per-direction impairment fields like delay_a_to_b shape only the direction from the a-side veth toward the b side.

Ports must be non-negative ints, excluding bool. Switch data ports range from 0 to 510. Automatic allocation starts at 1 for switches, skips the CPU port and raises :class:TopologyError when exhausted.

Parameters:

Name Type Description Default
a NodeRef

First endpoint (host or switch name, or a node object).

 required
b NodeRef

Second endpoint.

 required
port_a int | None

Port number on the a side. Auto-assigned if omitted.

 None
port_b int | None

Port number on the b side. Auto-assigned if omitted.

 None
ip_a str | None

IPv4 CIDR override for the a-side interface.

 None
ip_b str | None

IPv4 CIDR override for the b-side interface.

 None
mac_a str | None

MAC override for the a-side interface.

 None
mac_b str | None

MAC override for the b-side interface.

 None
ip6_a str | None

IPv6 CIDR override for the a-side interface.

 None
ip6_b str | None

IPv6 CIDR override for the b-side interface.

 None
bandwidth str | None

Symmetric link-rate cap (e.g. "10mbit").

 None
delay str | None

Symmetric one-way delay (e.g. "50ms").

 None
jitter str | None

Symmetric jitter; requires delay to be set.

 None
loss_pct float | None

Symmetric per-packet loss in [0.0, 100.0].

 None
mtu int | None

Link MTU, an int in [68, 65535], excluding bool, or None.

 None
bandwidth_a_to_b str | None

Per-direction bandwidth, a → b.

 None
bandwidth_b_to_a str | None

Per-direction bandwidth, b → a.

 None
delay_a_to_b str | None

Per-direction delay, a → b.

 None
delay_b_to_a str | None

Per-direction delay, b → a.

 None
jitter_a_to_b str | None

Per-direction jitter, a → b.

 None
jitter_b_to_a str | None

Per-direction jitter, b → a.

 None
loss_pct_a_to_b float | None

Per-direction loss, a → b.

 None
loss_pct_b_to_a float | None

Per-direction loss, b → a.

 None
delay_a_to_b_extra str | None

Additional delay added on top of the symmetric delay for the a → b direction. Requires symmetric delay; mutually exclusive with delay_a_to_b.

 None
delay_b_to_a_extra str | None

Same as above, b → a.

 None
jitter_a_to_b_extra str | None

Additional jitter on top of symmetric jitter for a → b. Requires symmetric jitter.

 None
jitter_b_to_a_extra str | None

Same as above, b → a.

 None
loss_pct_a_to_b_extra float | None

Additional loss percent on top of symmetric loss_pct for a → b. Requires symmetric loss_pct; combined value must stay ≤ 100.0.

 None
loss_pct_b_to_a_extra float | None

Same as above, b → a.

 None

Returns:

Type Description
Link

The newly created :class:Link.

Raises:

Type Description
TopologyError

on invalid parameter combinations or unresolved endpoints.
Source code in src/p4net/topo/topology.py 
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
def add_link(
    self,
    a: NodeRef,
    b: NodeRef,
    *,
    port_a: int | None = None,
    port_b: int | None = None,
    ip_a: str | None = None,
    ip_b: str | None = None,
    mac_a: str | None = None,
    mac_b: str | None = None,
    ip6_a: str | None = None,
    ip6_b: str | None = None,
    bandwidth: str | None = None,
    delay: str | None = None,
    jitter: str | None = None,
    loss_pct: float | None = None,
    mtu: int | None = None,
    bandwidth_a_to_b: str | None = None,
    bandwidth_b_to_a: str | None = None,
    delay_a_to_b: str | None = None,
    delay_b_to_a: str | None = None,
    jitter_a_to_b: str | None = None,
    jitter_b_to_a: str | None = None,
    loss_pct_a_to_b: float | None = None,
    loss_pct_b_to_a: float | None = None,
    delay_a_to_b_extra: str | None = None,
    delay_b_to_a_extra: str | None = None,
    jitter_a_to_b_extra: str | None = None,
    jitter_b_to_a_extra: str | None = None,
    loss_pct_a_to_b_extra: float | None = None,
    loss_pct_b_to_a_extra: float | None = None,
) -> Link:
    """Append a :class:`Link` between two nodes.

    Endpoint sides ``a`` and ``b`` are anchored by argument order;
    per-direction impairment fields like ``delay_a_to_b`` shape only
    the direction from the ``a``-side veth toward the ``b`` side.

    Ports must be non-negative ints, excluding bool. Switch data ports
    range from 0 to 510. Automatic allocation starts at 1 for switches,
    skips the CPU port and raises :class:`TopologyError` when exhausted.

    Args:
        a: First endpoint (host or switch name, or a node object).
        b: Second endpoint.
        port_a: Port number on the ``a`` side. Auto-assigned if omitted.
        port_b: Port number on the ``b`` side. Auto-assigned if omitted.
        ip_a: IPv4 CIDR override for the ``a``-side interface.
        ip_b: IPv4 CIDR override for the ``b``-side interface.
        mac_a: MAC override for the ``a``-side interface.
        mac_b: MAC override for the ``b``-side interface.
        ip6_a: IPv6 CIDR override for the ``a``-side interface.
        ip6_b: IPv6 CIDR override for the ``b``-side interface.
        bandwidth: Symmetric link-rate cap (e.g. ``"10mbit"``).
        delay: Symmetric one-way delay (e.g. ``"50ms"``).
        jitter: Symmetric jitter; requires ``delay`` to be set.
        loss_pct: Symmetric per-packet loss in [0.0, 100.0].
        mtu: Link MTU, an int in [68, 65535], excluding bool, or None.
        bandwidth_a_to_b: Per-direction bandwidth, ``a`` → ``b``.
        bandwidth_b_to_a: Per-direction bandwidth, ``b`` → ``a``.
        delay_a_to_b: Per-direction delay, ``a`` → ``b``.
        delay_b_to_a: Per-direction delay, ``b`` → ``a``.
        jitter_a_to_b: Per-direction jitter, ``a`` → ``b``.
        jitter_b_to_a: Per-direction jitter, ``b`` → ``a``.
        loss_pct_a_to_b: Per-direction loss, ``a`` → ``b``.
        loss_pct_b_to_a: Per-direction loss, ``b`` → ``a``.
        delay_a_to_b_extra: Additional delay added on top of the symmetric
            ``delay`` for the ``a`` → ``b`` direction. Requires symmetric
            ``delay``; mutually exclusive with ``delay_a_to_b``.
        delay_b_to_a_extra: Same as above, ``b`` → ``a``.
        jitter_a_to_b_extra: Additional jitter on top of symmetric
            ``jitter`` for ``a`` → ``b``. Requires symmetric ``jitter``.
        jitter_b_to_a_extra: Same as above, ``b`` → ``a``.
        loss_pct_a_to_b_extra: Additional loss percent on top of symmetric
            ``loss_pct`` for ``a`` → ``b``. Requires symmetric
            ``loss_pct``; combined value must stay ≤ 100.0.
        loss_pct_b_to_a_extra: Same as above, ``b`` → ``a``.

    Returns:
        The newly created :class:`Link`.

    Raises:
        TopologyError: on invalid parameter combinations or
            unresolved endpoints.
    """
    node_a = self._resolve(a)
    node_b = self._resolve(b)
    port_a_val = self._auto_port(node_a, port_a)
    port_b_val = self._auto_port(node_b, port_b)
    iface_a = _iface_name(node_a.name, port_a_val)
    iface_b = _iface_name(node_b.name, port_b_val)
    for name in (iface_a, iface_b):
        try:
            _validate_ifname(name)
        except ValueError as exc:
            raise TopologyError(f"{exc}; shorten the node name") from exc
    if ip_a is not None and isinstance(node_a, P4Switch):
        raise TopologyError(
            "P4 switch data ports do not carry IP addresses; remove ip_a from the add_link call"
        )
    if ip_b is not None and isinstance(node_b, P4Switch):
        raise TopologyError(
            "P4 switch data ports do not carry IP addresses; remove ip_b from the add_link call"
        )
    if ip6_a is not None and isinstance(node_a, P4Switch):
        raise TopologyError(
            "P4 switch data ports do not carry IP addresses; "
            "remove ip6_a from the add_link call"
        )
    if ip6_b is not None and isinstance(node_b, P4Switch):
        raise TopologyError(
            "P4 switch data ports do not carry IP addresses; "
            "remove ip6_b from the add_link call"
        )
    ep_a = LinkEndpoint(
        node=node_a.name,
        port=port_a_val,
        iface_name=iface_a,
        ip=ip_a,
        mac=mac_a,
        ip6=ip6_a,
    )
    ep_b = LinkEndpoint(
        node=node_b.name,
        port=port_b_val,
        iface_name=iface_b,
        ip=ip_b,
        mac=mac_b,
        ip6=ip6_b,
    )
    link = Link(
        a=ep_a,
        b=ep_b,
        bandwidth=bandwidth,
        delay=delay,
        jitter=jitter,
        loss_pct=loss_pct,
        mtu=mtu,
        bandwidth_a_to_b=bandwidth_a_to_b,
        bandwidth_b_to_a=bandwidth_b_to_a,
        delay_a_to_b=delay_a_to_b,
        delay_b_to_a=delay_b_to_a,
        jitter_a_to_b=jitter_a_to_b,
        jitter_b_to_a=jitter_b_to_a,
        loss_pct_a_to_b=loss_pct_a_to_b,
        loss_pct_b_to_a=loss_pct_b_to_a,
        delay_a_to_b_extra=delay_a_to_b_extra,
        delay_b_to_a_extra=delay_b_to_a_extra,
        jitter_a_to_b_extra=jitter_a_to_b_extra,
        jitter_b_to_a_extra=jitter_b_to_a_extra,
        loss_pct_a_to_b_extra=loss_pct_a_to_b_extra,
        loss_pct_b_to_a_extra=loss_pct_b_to_a_extra,
    )
    self._links.append(link)
    return link

neighbors_of 

neighbors_of(name: str) -> list[tuple[Link, LinkEndpoint, LinkEndpoint]]

Return every link involving name, as (link, near, far).

Source code in src/p4net/topo/topology.py 
378
379
380
381
382
383
384
385
386
def neighbors_of(self, name: str) -> list[tuple[Link, LinkEndpoint, LinkEndpoint]]:
    """Return every link involving `name`, as (link, near, far)."""
    result: list[tuple[Link, LinkEndpoint, LinkEndpoint]] = []
    for link in self._links:
        if link.a.node == name:
            result.append((link, link.a, link.b))
        elif link.b.node == name:
            result.append((link, link.b, link.a))
    return result

port_assignments 

port_assignments(switch_name: str) -> dict[int, LinkEndpoint]

Map of switch port number to the FAR endpoint connected on that port.

Source code in src/p4net/topo/topology.py 
388
389
390
391
392
393
394
395
396
397
398
def port_assignments(self, switch_name: str) -> dict[int, LinkEndpoint]:
    """Map of switch port number to the FAR endpoint connected on that port."""
    if switch_name not in self._switches:
        raise TopologyError(f"no switch named {switch_name!r}")
    result: dict[int, LinkEndpoint] = {}
    for link in self._links:
        if link.a.node == switch_name and link.a.port is not None:
            result[link.a.port] = link.b
        elif link.b.node == switch_name and link.b.port is not None:
            result[link.b.port] = link.a
    return result

validate 

validate() -> None

Raise TopologyError listing every internal-consistency problem found.

Node names must be unique across hosts and switches, and mapping keys must match their descriptions' names. All configured gRPC and Thrift ports must be distinct across the topology, including the two ports of each switch. Every link endpoint must have an assigned port and a non-empty interface name. Different hosts cannot use the same effective IP address, regardless of prefix length. Switch data ports cannot carry IPv4 or IPv6 addresses.

Source code in src/p4net/topo/topology.py 
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
def validate(self) -> None:
    """Raise `TopologyError` listing every internal-consistency problem found.

    Node names must be unique across hosts and switches, and mapping keys
    must match their descriptions' names.
    All configured gRPC and Thrift ports must be distinct across the topology,
    including the two ports of each switch. Every link endpoint must have
    an assigned port and a non-empty interface name. Different hosts cannot
    use the same effective IP address, regardless of prefix length.
    Switch data ports cannot carry IPv4 or IPv6 addresses.
    """
    errors: list[str] = []

    for kind, mapping in (("host", self._hosts), ("switch", self._switches)):
        for node_key, node in mapping.items():
            if node_key != node.name:
                errors.append(f"{kind} key {node_key!r} does not match node name {node.name!r}")
    for node_key in self._hosts:
        if node_key in self._switches:
            errors.append(f"node name {node_key!r} is used by both a host and a switch")

    # 1. Endpoint references resolve to a known node.
    # 2. No self-loops.
    for i, link in enumerate(self._links):
        for which, ep in (("a", link.a), ("b", link.b)):
            if ep.node not in self._hosts and ep.node not in self._switches:
                errors.append(f"link[{i}] endpoint {which} references unknown node {ep.node!r}")
            if ep.port is None:
                errors.append(f"link[{i}] endpoint {which} ({ep.node!r}) has no assigned port")
            switch = self._switches.get(ep.node)
            if switch is not None:
                for field, address in (("ip", ep.ip), ("ip6", ep.ip6)):
                    if address is not None:
                        errors.append(
                            f"link[{i}] endpoint {which} switch {switch.name!r}: "
                            "P4 switch data ports do not carry IP addresses "
                            f"({field}={address!r})"
                        )
            if switch is not None and ep.port is not None and ep.port > _SWITCH_PORT_MAX:
                errors.append(
                    f"link[{i}] endpoint {which} switch port {ep.port} "
                    f"out of range [0, {_SWITCH_PORT_MAX}]"
                )
            if (
                switch is not None
                and switch.cpu_port is not None
                and ep.port == switch.cpu_port
            ):
                errors.append(
                    f"link[{i}] endpoint {which} uses CPU port {switch.cpu_port} "
                    f"on switch {switch.name!r}"
                )
        if link.a.node == link.b.node:
            errors.append(f"link[{i}] connects node {link.a.node!r} to itself")

    # 3. No (node, port) pair repeats across links.
    seen: dict[tuple[str, int], int] = {}
    for i, link in enumerate(self._links):
        for _which, ep in (("a", link.a), ("b", link.b)):
            if ep.port is None:
                continue
            key = (ep.node, ep.port)
            if key in seen:
                errors.append(
                    f"port collision: ({ep.node!r}, port {ep.port}) used by "
                    f"link[{seen[key]}] and link[{i}]"
                )
            else:
                seen[key] = i

    # 4. Device IDs are unique; gRPC and Thrift share the TCP port space.
    device_owners: dict[int, str] = {}
    port_owners: dict[int, tuple[str, str]] = {}
    for sw in self._switches.values():
        if sw.device_id is not None:
            if sw.device_id in device_owners:
                errors.append(
                    f"switch device_id collision: {sw.name!r} and "
                    f"{device_owners[sw.device_id]!r} both use {sw.device_id}"
                )
            else:
                device_owners[sw.device_id] = sw.name
        for field_name, port in (("grpc_port", sw.grpc_port), ("thrift_port", sw.thrift_port)):
            if port is None:
                continue
            if port in port_owners:
                owner, owner_field = port_owners[port]
                label = field_name if field_name == owner_field else "service port"
                errors.append(
                    f"switch {label} collision: {sw.name!r} ({field_name}) and "
                    f"{owner!r} ({owner_field}) both use {port}"
                )
            else:
                port_owners[port] = (sw.name, field_name)

    # 5. Interface names are valid and unique within their final namespaces.
    interface_owners: dict[tuple[str | None, str], str] = {}
    for i, link in enumerate(self._links):
        for which, ep in (("a", link.a), ("b", link.b)):
            if ep.iface_name is None or ep.iface_name == "":
                errors.append(
                    f"link[{i}] endpoint {which} ({ep.node!r}) has no assigned interface name"
                )
                continue
            try:
                _validate_ifname(ep.iface_name)
            except ValueError as exc:
                errors.append(f"link[{i}] endpoint {which} ({ep.node!r}): {exc}")
                continue
            owner = f"link[{i}] endpoint {which} ({ep.node!r})"
            if ep.iface_name == "lo":
                errors.append(f"{owner}: interface name 'lo' is reserved for loopback")
                continue
            if ep.node in self._hosts:
                scope: str | None = ep.node
                label = f"host {ep.node!r}"
            elif ep.node in self._switches:
                scope = None
                label = "root namespace"
            else:
                continue
            interface_key = (scope, ep.iface_name)
            if interface_key in interface_owners:
                errors.append(
                    f"interface name collision in {label}: {ep.iface_name!r} "
                    f"used by {interface_owners[interface_key]} and {owner}"
                )
            else:
                interface_owners[interface_key] = owner

    # 6. Different hosts cannot share an effective address, regardless of prefix.
    ipv4_owners: dict[ipaddress.IPv4Address, set[str]] = {}
    ipv6_owners: dict[ipaddress.IPv6Address, set[str]] = {}
    seen_hosts: set[str] = set()
    for link in self._links:
        for ep in (link.a, link.b):
            host = self._hosts.get(ep.node)
            if host is None:
                continue
            ipv4, ipv6 = _effective_host_addresses(
                host, ep, first_link=host.name not in seen_hosts
            )
            seen_hosts.add(host.name)
            if ipv4 is not None:
                try:
                    iface = ipaddress.IPv4Interface(ipv4)
                except (ValueError, ipaddress.AddressValueError, ipaddress.NetmaskValueError):
                    errors.append(f"link endpoint {ep.node!r}: invalid IP {ipv4!r}")
                else:
                    ipv4_owners.setdefault(iface.ip, set()).add(host.name)
            if ipv6 is not None:
                try:
                    iface6 = ipaddress.IPv6Interface(ipv6)
                except (ValueError, ipaddress.AddressValueError, ipaddress.NetmaskValueError):
                    errors.append(f"link endpoint {ep.node!r}: invalid ip6 {ipv6!r}")
                else:
                    ipv6_owners.setdefault(iface6.ip, set()).add(host.name)
    for addr, nodes in ipv4_owners.items():
        if len(nodes) > 1:
            errors.append(f"IP collision: address {addr} used by hosts {sorted(nodes)}")

    for addr6, nodes6 in ipv6_owners.items():
        if len(nodes6) > 1:
            errors.append(f"IPv6 collision: address {addr6} used by hosts {sorted(nodes6)}")

    if errors:
        raise TopologyError(
            "topology validation failed with the following problems:\n  - "
            + "\n  - ".join(errors)
        )

to_graphviz 

to_graphviz(*, layout: str = 'LR') -> str

Render the topology as a Graphviz DOT graph string.

Hosts are ellipses labelled with name and primary IP(s); switches are boxes labelled with name and gRPC port. Edges are drawn with arrowhead=none to keep the rendering version-stable across graphviz releases. layout controls rankdir and must be one of "LR", "RL", "TB", "BT".

Source code in src/p4net/topo/topology.py 
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
def to_graphviz(self, *, layout: str = "LR") -> str:
    """Render the topology as a Graphviz DOT graph string.

    Hosts are ellipses labelled with name and primary IP(s); switches are
    boxes labelled with name and gRPC port. Edges are drawn with
    ``arrowhead=none`` to keep the rendering version-stable across
    graphviz releases. ``layout`` controls ``rankdir`` and must be one
    of ``"LR"``, ``"RL"``, ``"TB"``, ``"BT"``.
    """
    if layout not in {"LR", "RL", "TB", "BT"}:
        raise TopologyError(f"layout {layout!r} must be one of 'LR', 'RL', 'TB', 'BT'")
    lines: list[str] = ["digraph p4net {"]
    lines.append(f"  rankdir={layout};")
    lines.append('  node [fontname="monospace"];')
    for host in self._hosts.values():
        label_parts = [host.name]
        if host.ip is not None:
            label_parts.append(host.ip)
        if host.ip6 is not None:
            label_parts.append(host.ip6)
        label = "\\n".join(label_parts)
        lines.append(f'  "{host.name}" [shape=ellipse, label="{label}"];')
    for sw in self._switches.values():
        label_parts = [sw.name]
        if sw.grpc_port is not None:
            label_parts.append(f"grpc :{sw.grpc_port}")
        label = "\\n".join(label_parts)
        lines.append(f'  "{sw.name}" [shape=box, label="{label}"];')
    for link in self._links:
        lines.append(f'  "{link.a.node}" -> "{link.b.node}" [arrowhead=none];')
    lines.append("}")
    return "\n".join(lines) + "\n"

render_graphviz 

render_graphviz(output_path: Path, *, layout: str = 'LR', format: str = 'png') -> None

Render via the system dot binary to output_path.

format is forwarded to dot -T<format> (png, svg, pdf, dot). For format="dot" the source is written verbatim and dot is not invoked, so this path works without graphviz installed. Raises :class:TopologyError if dot is missing or the render fails.

Source code in src/p4net/topo/topology.py 
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
def render_graphviz(
    self,
    output_path: Path,
    *,
    layout: str = "LR",
    format: str = "png",
) -> None:
    """Render via the system ``dot`` binary to ``output_path``.

    ``format`` is forwarded to ``dot -T<format>`` (png, svg, pdf, dot).
    For ``format="dot"`` the source is written verbatim and ``dot`` is
    not invoked, so this path works without graphviz installed.
    Raises :class:`TopologyError` if ``dot`` is missing or the render
    fails.
    """
    import shutil
    import subprocess

    source = self.to_graphviz(layout=layout)
    output_path = Path(output_path)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    if format == "dot":
        output_path.write_text(source)
        return
    dot_bin = shutil.which("dot")
    if dot_bin is None:
        raise TopologyError(
            "graphviz `dot` binary not found on PATH; "
            "install graphviz or use format='dot' to write the source file directly"
        )
    try:
        subprocess.run(
            [dot_bin, f"-T{format}", "-o", str(output_path)],
            input=source.encode("utf-8"),
            check=True,
            capture_output=True,
        )
    except subprocess.CalledProcessError as exc:
        stderr = (exc.stderr or b"").decode("utf-8", errors="replace")
        raise TopologyError(
            f"`dot -T{format}` failed (rc={exc.returncode}): {stderr.strip()}"
        ) from exc

to_dict 

to_dict() -> dict[str, Any]

Return a JSON-serialisable snapshot of the topology.

Source code in src/p4net/topo/topology.py 
647
648
649
650
651
652
653
def to_dict(self) -> dict[str, Any]:
    """Return a JSON-serialisable snapshot of the topology."""
    return {
        "hosts": {name: _host_to_dict(host) for name, host in self._hosts.items()},
        "switches": {name: _switch_to_dict(sw) for name, sw in self._switches.items()},
        "links": [_link_to_dict(link) for link in self._links],
    }

from_dict classmethod 

from_dict(data: dict[str, Any]) -> Topology

Reconstruct a :class:Topology from a :meth:to_dict payload.

Node keys, names, endpoint ports, interface names, and address overrides are restored as stored. Call :meth:validate to check node identity and endpoint assignments, and reject IP addresses on switch data ports.

Source code in src/p4net/topo/topology.py 
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
@classmethod
def from_dict(cls, data: dict[str, Any]) -> Topology:
    """Reconstruct a :class:`Topology` from a :meth:`to_dict` payload.

    Node keys, names, endpoint ports, interface names, and address overrides
    are restored as stored. Call :meth:`validate` to check node identity and
    endpoint assignments, and reject IP addresses on switch data ports.
    """
    topo = cls()
    for name, host_data in data.get("hosts", {}).items():
        topo._hosts[name] = Host(
            name=host_data["name"],
            ip=host_data.get("ip"),
            mac=host_data.get("mac"),
            default_route=host_data.get("default_route"),
            ip6=host_data.get("ip6"),
            default_route6=host_data.get("default_route6"),
        )
    for name, sw_data in data.get("switches", {}).items():
        topo._switches[name] = P4Switch(
            name=sw_data["name"],
            p4_src=Path(sw_data["p4_src"]),
            arch=sw_data.get("arch", "v1model"),
            device_id=sw_data.get("device_id"),
            grpc_port=sw_data.get("grpc_port"),
            thrift_port=sw_data.get("thrift_port"),
            cpu_port=sw_data.get("cpu_port"),
            log_level=sw_data.get("log_level", "info"),
            pcap_enabled=sw_data.get("pcap_enabled", True),
        )
    for link_data in data.get("links", []):
        ep_a = LinkEndpoint(
            node=link_data["a"]["node"],
            port=link_data["a"].get("port"),
            iface_name=link_data["a"].get("iface_name"),
            ip=link_data["a"].get("ip"),
            mac=link_data["a"].get("mac"),
            ip6=link_data["a"].get("ip6"),
        )
        ep_b = LinkEndpoint(
            node=link_data["b"]["node"],
            port=link_data["b"].get("port"),
            iface_name=link_data["b"].get("iface_name"),
            ip=link_data["b"].get("ip"),
            mac=link_data["b"].get("mac"),
            ip6=link_data["b"].get("ip6"),
        )
        topo._links.append(
            Link(
                a=ep_a,
                b=ep_b,
                bandwidth=link_data.get("bandwidth"),
                delay=link_data.get("delay"),
                jitter=link_data.get("jitter"),
                loss_pct=link_data.get("loss_pct"),
                mtu=link_data.get("mtu"),
                bandwidth_a_to_b=link_data.get("bandwidth_a_to_b"),
                bandwidth_b_to_a=link_data.get("bandwidth_b_to_a"),
                delay_a_to_b=link_data.get("delay_a_to_b"),
                delay_b_to_a=link_data.get("delay_b_to_a"),
                jitter_a_to_b=link_data.get("jitter_a_to_b"),
                jitter_b_to_a=link_data.get("jitter_b_to_a"),
                loss_pct_a_to_b=link_data.get("loss_pct_a_to_b"),
                loss_pct_b_to_a=link_data.get("loss_pct_b_to_a"),
                delay_a_to_b_extra=link_data.get("delay_a_to_b_extra"),
                delay_b_to_a_extra=link_data.get("delay_b_to_a_extra"),
                jitter_a_to_b_extra=link_data.get("jitter_a_to_b_extra"),
                jitter_b_to_a_extra=link_data.get("jitter_b_to_a_extra"),
                loss_pct_a_to_b_extra=link_data.get("loss_pct_a_to_b_extra"),
                loss_pct_b_to_a_extra=link_data.get("loss_pct_b_to_a_extra"),
            )
        )
    return topo