ports module
Traffic Generation APIs
Port APIs Summary Table
Id |
Action |
API |
Description |
|---|---|---|---|
1 |
Get Port Object |
module.ports.obtain() |
Obtain a port object representing a specific port on a module |
2 |
Release Port |
port.reservation.set_release() |
Reserve or release the port. |
3 |
Relinquish Port |
port.reservation.set_relinquish() |
Reserve or release the port. |
4 |
Reserve Port |
port.reservation.set_reserve() |
Reserve or release the port. |
5 |
Get Port Reservation Status |
port.reservation.get() |
Reserve or release the port. |
6 |
Get Port Reserved By |
port.reserved_by.get() |
Get the current reservation status of the port. |
7 |
Set Port Description |
port.comment.set() |
A text description of the port. |
8 |
Get Port Description |
port.comment.get() |
A text description of the port. |
9 |
Get Interface Name |
port.interface.get() |
The interface name of the port. |
10 |
Reset Port |
port.reset.set() |
Reset the port. |
11 |
Flash Port LEDs On |
port.flash.set_on() |
Make the port LED flash on and off with a 1-second interval. |
12 |
Flash Port LEDs Off |
port.flash.set_off() |
Make the port LED flash on and off with a 1-second interval. |
13 |
Get Flash Status |
port.flash.get() |
Make the port LED flash on and off with a 1-second interval. |
14 |
Get Sync Status |
port.sync_status.get() |
Get the port synchronization status. |
15 |
Enable TX Output |
port.tx_config.enable.set_on() |
Enable or disable the port transmitter. |
16 |
Disable TX Output |
port.tx_config.enable.set_off() |
Enable or disable the port transmitter. |
17 |
Get TX Output Status |
port.tx_config.enable.get() |
Enable or disable the port transmitter. |
18 |
Set TX Time Limit |
port.tx_config.time_limit.set() |
Set or get the transmit time limit. |
19 |
Get TX Time Limit |
port.tx_config.time_limit.get() |
Set or get the transmit time limit. |
20 |
Set TX Packet Limit |
port.tx_config.packet_limit.set() |
Set or get the transmit packet limit. |
21 |
Get TX Packet Limit |
port.tx_config.packet_limit.get() |
Set or get the transmit packet limit. |
22 |
Get TX Time Elapsed |
port.tx_config.time.get() |
Get the elapsed transmit time. |
23 |
Get TX Delay |
port.tx_config.delay.get() |
Set or get the transmit delay. |
24 |
Set TX Prepare |
port.tx_config.prepare.set() |
Prepare the port for transmission. |
25 |
Start Traffic |
port.traffic.state.set_start() |
Start or stop traffic on the port. |
26 |
Stop Traffic |
port.traffic.state.set_stop() |
Start or stop traffic on the port. |
27 |
Get Traffic State |
port.traffic.state.get() |
Start or stop traffic on the port. |
28 |
Get Traffic Error |
port.traffic.error.get() |
Get the traffic error status. |
29 |
Send Manual Packet |
port.tx_single_pkt.send.set() |
Transmit a single manually specified packet. |
30 |
Get Manual Packet Time |
port.tx_single_pkt.time.get() |
Get the time taken to transmit a single manually specified packet. |
31 |
Disable Dynamic Traffic |
port.dynamic.set_off() |
Enable or disable dynamic traffic on the port. |
32 |
Enable Dynamic Traffic |
port.dynamic.set_on() |
Enable or disable dynamic traffic on the port. |
33 |
Get Dynamic Traffic Status |
port.dynamic.get() |
Enable or disable dynamic traffic on the port. |
34 |
Set Rate Fraction |
port.rate.fraction.set() |
Set or get the transmit rate as a fraction of line rate in parts per million (ppm). |
35 |
Get Rate Fraction |
port.rate.fraction.get() |
Set or get the transmit rate as a fraction of line rate in parts per million (ppm). |
36 |
Set Rate L2 Bits Per Second |
port.rate.l2_bps.set() |
Set or get the transmit rate in bits per second at Layer 2. |
37 |
Get Rate L2 Bits Per Second |
port.rate.l2_bps.get() |
Set or get the transmit rate in bits per second at Layer 2. |
38 |
Set Rate Frames Per Second |
port.rate.pps.set() |
Set or get the transmit rate in frames per second. |
39 |
Get Rate Frames Per Second |
port.rate.pps.get() |
Set or get the transmit rate in frames per second. |
40 |
Set TX Mode Normal |
port.tx_config.mode.set_normal() |
Set or get the transmit mode. |
41 |
Set TX Mode Burst |
port.tx_config.mode.set_burst() |
Set or get the transmit mode. |
42 |
Set TX Mode Sequential |
port.tx_config.mode.set_sequential() |
Set or get the transmit mode. |
43 |
Set TX Mode Strict Uniform |
port.tx_config.mode.set_strictuniform() |
Set or get the transmit mode. |
44 |
Get TX Mode |
port.tx_config.mode.get() |
Set or get the transmit mode. |
45 |
Set Burst Period |
port.tx_config.burst_period.set() |
Set or get the burst period in milliseconds. |
46 |
Get Burst Period |
port.tx_config.burst_period.get() |
Set or get the burst period in milliseconds. |
47 |
Set Inter-frame Gap |
port.interframe_gap.set() |
Set or get the inter-frame gap minimum byte count. |
48 |
Get Inter-frame Gap |
port.interframe_gap.get() |
Set or get the inter-frame gap minimum byte count. |
49 |
Set Autoneg On for RJ45 Port |
port.autoneg_selection.set_on() |
Set Autoneg On for RJ45 Odin port. |
50 |
Set Autoneg Off for RJ45 Port |
port.autoneg_selection.set_off() |
Set Autoneg Off for RJ45 Odin port. |
51 |
Get Autoneg Status on RJ45 Port |
port.autoneg_selection.get() |
Get Autoneg status on RJ45 Odin port. |
52 |
Set Speed Mode Auto |
port.speed.selection.set_auto() |
Set or get the speed mode selection. |
53 |
Set Speed Mode 10M |
port.speed.selection.set_f10m() |
Set or get the speed mode selection. |
54 |
Set Speed Mode 10M/100M |
port.speed.selection.set_f10m100m() |
Set or get the speed mode selection. |
55 |
Set Speed Mode 10M HDX |
port.speed.selection.set_f10mhdx() |
Set or get the speed mode selection. |
56 |
Set Speed Mode 100M |
port.speed.selection.set_f100m() |
Set or get the speed mode selection. |
57 |
Set Speed Mode 100M/1G |
port.speed.selection.set_f100m1g() |
Set or get the speed mode selection. |
58 |
Set Speed Mode 100M/1G/10G |
port.speed.selection.set_f100m1g10g() |
Set or get the speed mode selection. |
59 |
Set Speed Mode 100M/1G/2.5G |
port.speed.selection.set_f100m1g2500m() |
Set or get the speed mode selection. |
60 |
Set Speed Mode 100M HDX |
port.speed.selection.set_f100mhdx() |
Set or get the speed mode selection. |
61 |
Set Speed Mode 1G |
port.speed.selection.set_f1g() |
Set or get the speed mode selection. |
62 |
Set Speed Mode 2.5G |
port.speed.selection.set_f2500m() |
Set or get the speed mode selection. |
63 |
Set Speed Mode 5G |
port.speed.selection.set_f5g() |
Set or get the speed mode selection. |
64 |
Set Speed Mode 10G |
port.speed.selection.set_f10g() |
Set or get the speed mode selection. |
65 |
Set Speed Mode 40G |
port.speed.selection.set_f40g() |
Set or get the speed mode selection. |
66 |
Set Speed Mode 100G |
port.speed.selection.set_f100g() |
Set or get the speed mode selection. |
67 |
Get Speed Mode Selection |
port.speed.selection.get() |
Set or get the speed mode selection. |
68 |
Get Supported Speed Modes |
port.speed. supported.get() |
|
69 |
Get Current Speed |
port.speed.current.get() |
|
70 |
Set Speed Reduction |
port.speed.reduction.set() |
|
71 |
Get Speed Reduction |
port.speed.reduction.get() |
|
72 |
Get Optical RX Power |
port.status.get() |
|
73 |
Set MAC Address |
port.net_config.mac.address.set() |
|
74 |
Get MAC Address |
port.net_config.mac.address.get() |
|
75 |
Set Auto MAC Training |
port.net_config.mac.autotrain.set() |
|
76 |
Get Auto MAC Training |
port.net_config.mac.autotrain.get() |
|
77 |
Enable PAUSE |
port.pause.set_on() |
|
78 |
Disable PAUSE |
port.pause.set_off() |
|
79 |
Get PAUSE Status |
port.pause.get() |
|
80 |
Set PFC Enable |
port.pfc_enable.set() |
|
81 |
Get PFC Enable |
port.pfc_enable.get() |
|
82 |
Set Gap Monitor |
port.gap_monitor.set() |
|
83 |
Get Gap Monitor |
port.gap_monitor.get() |
|
84 |
Set Checksum Offset |
port.checksum.set() |
|
85 |
Get Checksum Offset |
port.checksum.get() |
|
86 |
Set Random Seed |
port.random_seed.set() |
|
87 |
Get Random Seed |
port.random_seed.get() |
|
88 |
Set Maximum Header Length |
port.max_header_length.set() |
|
89 |
Get Maximum Header Length |
port.max_header_length.get() |
|
90 |
Set IMIX Weights |
port.mix.weights.set() |
|
91 |
Get IMIX Weights |
port.mix.weights.get() |
|
92 |
Set IMIX Lengths |
port.mix.lengths[0].set() |
|
93 |
Set IMIX Lengths |
port.mix.lengths[1].set() |
|
94 |
Set IMIX Lengths |
port.mix.lengths[14].set() |
|
95 |
Set IMIX Lengths |
port.mix.lengths[15].set() |
|
96 |
Get IMIX Lengths |
port.mix.lengths[0].get() |
|
97 |
Get IMIX Lengths |
port.mix.lengths[1].get() |
|
98 |
Get IMIX Lengths |
port.mix.lengths[14].get() |
|
99 |
Get IMIX Lengths |
port.mix.lengths[15].get() |
|
100 |
Set Payload Mode Normal |
port.payload_mode.set_normal() |
|
101 |
Set Payload Mode EXTPL |
port.payload_mode.set_extpl() |
|
102 |
Set Payload Mode CDF |
port.payload_mode.set_cdf() |
|
103 |
Get Payload Mode |
port.payload_mode.get() |
|
104 |
Set TPLD Mode Normal |
port.tpld_mode.set_normal() |
|
105 |
Set TPLD Mode Micro |
port.tpld_mode.set_micro() |
|
106 |
Get TPLD Mode |
port.tpld_mode.get() |
|
107 |
Set Loopback None |
port.loopback.set_none() |
|
108 |
Set Loopback L1RX to TX |
port.loopback.set_l1rx2tx() |
|
109 |
Set Loopback L2RX to TX |
port.loopback.set_l2rx2tx() |
|
110 |
Set Loopback L3RX to TX |
port.loopback.set_l3rx2tx() |
|
111 |
Set Loopback Port to Port |
port.loopback.set_port2port() |
|
112 |
Set Loopback TXOFF to RX |
port.loopback.set_txoff2rx() |
|
113 |
Set Loopback TXON to RX |
port.loopback.set_txon2rx() |
|
114 |
Get Loopback Mode |
port.loopback.get() |
|
115 |
Set Latency Mode First to First |
port.latency_config.mode.set_first2first() |
|
116 |
Set Latency Mode First to Last |
port.latency_config.mode.set_first2last() |
|
117 |
Set Latency Mode Last to First |
port.latency_config.mode.set_last2first() |
|
118 |
Set Latency Mode Last to Last |
port.latency_config.mode.set_last2last() |
|
119 |
Get Latency Mode |
port.latency_config.mode.get() |
|
120 |
Set Latency Offset |
port.latency_config.offset.set() |
|
121 |
Get Latency Offset |
port.latency_config.offset.get() |
|
122 |
Set IPv4 Address |
port.net_config.ipv4.address.set() |
|
123 |
Get IPv4 Address |
port.net_config.ipv4.address.get() |
|
124 |
Enable IPv4 ARP Reply |
port.net_config.ipv4.arp_reply.set_on() |
|
125 |
Disable IPv4 ARP Reply |
port.net_config.ipv4.arp_reply.set_off() |
|
126 |
Get IPv4 ARP Reply |
port.net_config.ipv4.arp_reply.get() |
|
127 |
Enable IPv4 Ping Reply |
port.net_config.ipv4.ping_reply.set_on() |
|
128 |
Disable IPv4 Ping Reply |
port.net_config.ipv4.ping_reply.set_off() |
|
129 |
Get IPv4 Ping Reply |
port.net_config.ipv4.ping_reply.get() |
|
130 |
Set IPv6 Address |
port.net_config.ipv6.address.set() |
|
131 |
Get IPv6 Address |
port.net_config.ipv6.address.get() |
|
132 |
Enable IPv6 NDP Reply |
port.net_config.ipv6.arp_reply.set_on() |
|
133 |
Disable IPv6 NDP Reply |
port.net_config.ipv6.arp_reply.set_off() |
|
134 |
Get IPv6 NDP Reply |
port.net_config.ipv6.arp_reply.get() |
|
135 |
Enable IPv6 Ping Reply |
port.net_config.ipv6.ping_reply.set_on() |
|
136 |
Disable IPv6 Ping Reply |
port.net_config.ipv6.ping_reply.set_off() |
|
137 |
Get IPv6 Ping Reply |
port.net_config.ipv6.ping_reply.get() |
|
138 |
Set IPv4 ARP Table |
port.arp_rx_table.set() |
|
139 |
Get IPv4 ARP Table |
port.arp_rx_table.get() |
|
140 |
Set IPv6 NDP Table |
port.ndp_rx_table.set() |
|
141 |
Get IPv6 NDP Table |
port.ndp_rx_table.get() |
|
142 |
Set IPv4 Multicast |
port.multicast.mode.set() |
|
143 |
Get IPv4 Multicast Mode |
port.multicast.mode.get() |
|
144 |
Set IPv4 Multicast Advanced IGMPv3 |
port.multicast.mode_extended.set() |
|
145 |
Set IPv4 Multicast Advanced IGMPv3 |
port.multicast.mode_extended.set() |
|
146 |
Set IPv4 Multicast Advanced IGMPv2 |
port.multicast.mode_extended.set() |
|
147 |
Set IPv4 Multicast Advanced IGMPv2 |
port.multicast.mode_extended.set() |
|
148 |
Set IPv4 Multicast Advanced IGMPv2 |
port.multicast.mode_extended.set() |
|
149 |
Set IPv4 Multicast Advanced IGMPv2 |
port.multicast.mode_extended.set() |
|
150 |
Set IPv4 Multicast Advanced IGMPv2 |
port.multicast.mode_extended.set() |
|
151 |
Set IPv4 Multicast Advanced IGMPv2 |
port.multicast.mode_extended.set() |
|
152 |
Set IPv4 Multicast Advanced IGMPv2 |
port.multicast.mode_extended.set() |
|
153 |
Set IPv4 Multicast Advanced IGMPv2 |
port.multicast.mode_extended.set() |
|
154 |
Get IPv4 Multicast Advanced |
port.multicast.mode_extended.get() |
|
155 |
Set IPv4 Multicast Source List |
port.multicast.source_list.set() |
|
156 |
Get IPv4 Multicast Source List |
port.multicast.source_list.get() |
|
157 |
Set IPv4 Multicast Header |
port.multicast.header.set() |
|
158 |
Set IPv4 Multicast Header |
port.multicast.header.set() |
|
159 |
Get IPv4 Multicast Header |
port.multicast.header.get() |
|
160 |
Enable UAT Mode |
port.uat.mode.set() |
Only applicable in Xena1564 test suite. |
161 |
Disable UAT Mode |
port.uat.mode.set() |
Only applicable in Xena1564 test suite. |
162 |
Get UAT Mode |
port.uat.mode.get() |
Only applicable in Xena1564 test suite. |
163 |
Get UAT Frame Loss Ratio |
port.uat.flr.get() |
Only applicable in Xena1564 test suite. |
164 |
Get UAT Status |
port.statistics.rx.uat.status.get() |
Only applicable in Xena1564 test suite. |
165 |
Get UAT Time |
port.statistics.rx.uat.time.get() |
Only applicable in Xena1564 test suite. |
166 |
Set Capture Trigger Criteria |
port.capturer.trigger.set() |
|
167 |
Set Capture Trigger Criteria |
port.capturer.trigger.set() |
|
168 |
Set Capture Trigger Criteria |
port.capturer.trigger.set() |
|
169 |
Set Capture Trigger Criteria |
port.capturer.trigger.set() |
|
170 |
Set Capture Trigger Criteria |
port.capturer.trigger.set() |
|
171 |
Get Capture Trigger Criteria |
port.capturer.trigger.get() |
|
172 |
Set Frame to Keep ALL |
port.capturer.keep.set() |
|
173 |
Set Frame to Keep FCSERR |
port.capturer.keep.set() |
|
174 |
Set Frame to Keep NOTPLD |
port.capturer.keep.set() |
|
175 |
Set Frame to Keep PLDERR |
port.capturer.keep.set() |
|
176 |
Set Frame to Keep TPLD |
port.capturer.keep.set() |
TPLD == 5 |
177 |
Set Frame to Keep FILTER |
port.capturer.keep.set() |
Filter index == 2 |
178 |
Get Frame to Keep |
port.capturer.keep.get() |
|
179 |
Start Packet Capture |
port.capturer.state.set_start() |
|
180 |
Stop Packet Capture |
port.capturer.state.set_stop() |
|
181 |
Get Packet Capture State |
port.capturer.state.get() |
|
182 |
Get Capture Stats |
port.capturer.stats.get() |
|
183 |
Read Captured Packets |
port.capturer.obtain_captured() |
|
184 |
Create Filter |
port.filters.create() |
|
185 |
Obtain Filter |
port.filters.obtain(key=0) |
|
186 |
Obtain Multiple Filters |
port.filters.obtain_multiple(*[0,1,2]) |
|
187 |
Sync Filters |
port.filters.server_sync() |
|
188 |
Remove Filter |
port.filters.remove(position_idx=0) |
|
189 |
Delete Filter |
filter.delete() |
|
190 |
Get Filter Index |
filter.idx |
|
191 |
Get Filter Module Index |
filter.kind.module_id |
|
192 |
Get Filter Port Index |
filter.kind.port_id |
|
193 |
Enable Filter |
filter.enable.set_on() |
|
194 |
Disable Filter |
filter.enable.set_off() |
|
195 |
Get Filter Enable |
filter.enable.get() |
|
196 |
Set Filter Comment |
filter.comment.set() |
|
197 |
Get Filter Comment |
filter.comment.get() |
|
198 |
Set Filter Condition |
filter.condition.set() |
|
199 |
Get Filter Condition |
filter.condition.get() |
|
200 |
Set Filter String |
filter.string.set() |
|
201 |
Get Filter String |
filter.string.get() |
|
202 |
Create Length Term |
port.length_terms.create() |
|
203 |
Obtain Length Term |
port.length_terms.obtain(key=0) |
|
204 |
Obtain Multiple Length Terms |
port.length_terms.obtain_multiple(*[0,1,2]) |
|
205 |
Remove Length Term |
port.length_terms.remove(position_idx=0) |
|
206 |
Delete Length Term |
length_term.delete() |
|
207 |
Get Length Term Index |
length_term.idx |
|
208 |
Get Length Term Module Index |
length_term.kind.module_id |
|
209 |
Get Length Term Port Index |
length_term.kind.port_id |
|
210 |
Sync Length Terms |
port.length_terms.server_sync() |
|
211 |
Configure Length Term |
length_term.length.set() |
|
212 |
Configure Length Term |
length_term.length.set() |
|
213 |
Get Length Term Config |
length_term.length.get() |
|
214 |
Create Match Term |
port.match_terms.create() |
|
215 |
Obtain Match Term |
port.match_terms.obtain(key=0) |
|
216 |
Obtain Multiple Match Terms |
port.match_terms.obtain_multiple(*[0,1,2]) |
|
217 |
Remove Match Term |
port.match_terms.remove(position_idx=0) |
|
218 |
Delete Match Term |
match_term.delete() |
|
219 |
Get Match Term Index |
match_term.idx |
|
220 |
Get Match Term Module Index |
match_term.kind.module_id |
|
221 |
Get Match Term Port Index |
match_term.kind.port_id |
|
222 |
Sync Match Terms |
port.match_terms.server_sync() |
|
223 |
Set Match Term Value |
match_term.match.set() |
|
224 |
Get Match Term Value |
match_term.match.get() |
|
225 |
Set Match Term Position |
match_term.position.set() |
|
226 |
Get Match Term Position |
match_term.position.get() |
|
227 |
Set Match Term Protocol |
match_term.protocol.set() |
|
228 |
Get Match Term Protocol |
match_term.protocol.get() |
|
229 |
Create Histogram |
port.datasets.create() |
|
230 |
Obtain Histogram |
port.datasets.obtain(key=0) |
|
231 |
Obtain Multiple Histograms |
port.datasets.obtain_multiple(*[0,1,2]) |
|
232 |
Get Histogram Index |
histogram.idx |
|
233 |
Get Histogram Module Index |
histogram.kind.module_id |
|
234 |
Get Histogram Port Index |
histogram.kind.port_id |
|
235 |
Sync Histograms |
port.datasets.server_sync() |
|
236 |
Remove Histogram |
port.datasets.remove(position_idx=0) |
|
237 |
Delete Histogram |
histogram.delete() |
|
238 |
Enable Histogram |
histogram.enable.set_on() |
|
239 |
Disable Histogram |
histogram.enable.set_off() |
|
240 |
Get Histogram Enable |
histogram.enable.get() |
|
241 |
Configure Histogram Source TX_IFG |
histogram.source.set() |
|
242 |
Configure Histogram Source TX_LEN |
histogram.source.set() |
|
243 |
Configure Histogram Source RX_IFG |
histogram.source.set() |
|
244 |
Configure Histogram Source RX_LEN |
histogram.source.set() |
|
245 |
Configure Histogram Source RX_LATENCY |
histogram.source.set() |
|
246 |
Configure Histogram Source RX_JITTER |
histogram.source.set() |
|
247 |
Get Histogram Source |
histogram.source.get() |
first value going into the second bucket |
248 |
Configure Histogram Range |
histogram.range.set() |
|
249 |
Get Histogram Range |
histogram.range.get() |
|
250 |
Get Histogram Samples |
histogram.samples.get() |
|
251 |
Get Port Error Counter |
port.errors_count.get() |
|
252 |
Clear TX Statistics |
port.statistics.tx.clear.set() |
Clear all the transmit statistics for a port. |
253 |
Get Total TX Traffic |
port.statistics.tx.total.get() |
Obtains statistics concerning all the packets transmitted on the port. |
254 |
Get TX Non-TPLD Traffic |
port.statistics.tx.no_tpld.get() |
Obtains statistics concerning the packets without a test payload transmitted on the port. |
255 |
Get Extra TX Counters |
port.statistics.tx.extra.get() |
Obtains additional statistics for packets transmitted on the port. |
256 |
Get TX Stream Counters |
port.statistics.tx.obtain_from_stream(stream=0).get() |
Obtains statistics concerning the packets of a specific stream transmitted on the port. |
257 |
Clear RX Statistics |
port.statistics.rx.clear.set() |
Clear all the receive statistics for a port. |
258 |
Calibrate RX Latency |
port.statistics.rx.calibrate.set() |
Calibrate the latency calculation for packets received on the port. |
259 |
Get Total RX Traffic |
port.statistics.rx.total.get() |
Obtains statistics concerning all the packets received on the port. |
260 |
Get RX Non-TPLD Traffic |
port.statistics.rx.no_tpld.get() |
Obtains statistics concerning the packets without a test payload received on the port. |
261 |
Get RX PFC Counters |
port.statistics.rx.pfc_stats.get() |
Obtains statistics of received Priority Flow Control (PFC) packets on the port. |
262 |
Get Extra RX Counters |
port.statistics.rx.extra.get() |
Obtains statistics concerning special errors received on the port since received statistics were cleared. |
263 |
Get Received TPLDs |
port.statistics.rx.obtain_available_tplds() |
Obtain the set of test payload IDs observed among the received packets since receive statistics were cleared. |
264 |
Get RX Error Counters Per TPLD |
port.statistics.rx.access_tpld(…).errors.get() |
Obtains statistics concerning errors in the packets with a particular test payload id received on the port. |
265 |
Get RX Latency Per TPLD |
port.statistics.rx.access_tpld(…).latency.get() |
Obtains statistics concerning the latency experienced by the packets with a particular test payload id received on the port. |
266 |
Get RX Jitter Per TPLD |
port.statistics.rx.access_tpld(…).jitter.get() |
Obtains statistics concerning the jitter experienced by the packets with a particular test payload id received on the port. |
267 |
Get RX Traffic Per TPLD |
port.statistics.rx.access_tpld(…).traffic.get() |
Obtains traffic statistics concerning the packets with a particular test payload identifier received on the port. |
268 |
Get RX Filtered Traffic |
port.statistics.rx.obtain_filter_statistics(filter=0).get() |
Obtains statistics concerning the packets satisfying the condition of a particular filter for the port. |
269 |
Set MDI/MDIX Mode Auto |
port.mdix_mode.set_auto() |
|
270 |
Set MDI/MDIX Mode MDI |
port.mdix_mode.set_mdi() |
|
271 |
Set MDI/MDIX Mode MDIX |
port.mdix_mode.set_mdix() |
|
272 |
Get MDI/MDIX Mode |
port.mdix_mode.get() |
|
273 |
Set BroadR Reach Master |
port.brr.mode.set_master() |
|
274 |
Set BroadR Reach Slave |
port.brr.mode.set_slave() |
|
275 |
Get BroadR Reach Mode |
port.brr.mode.get() |
|
276 |
Get BroadR Reach Status |
port.brr.status.get() |
|
277 |
Get EEE Capabilities |
port.eee.capabilities.get() |
|
278 |
Get EEE Partner Capabilities |
port.eee.partner_capabilities.get() |
|
279 |
Disable EEE |
port.eee.enable.set_off() |
|
280 |
Enable EEE |
port.eee.enable.set_on() |
|
281 |
Get EEE Enable |
port.eee.enable.get() |
|
282 |
Set EEE Low Power TX Off |
port.eee.mode.set_off() |
|
283 |
Set EEE Low Power TX On |
port.eee.mode.set_on() |
|
284 |
Get EEE Mode |
port.eee.mode.get() |
|
285 |
Get EEE RX Power |
port.eee.rx_power.get() |
|
286 |
Get EEE SNR Margin |
port.eee.snr_margin.get() |
|
287 |
Get EEE Status |
port.eee.status.get() |
|
288 |
Set Runt RX Length |
port.runt.rx_length.set() |
Runt - RX Length |
289 |
Get Runt RX Length |
port.runt.rx_length.get() |
Runt - RX Length |
290 |
Set Runt TX Length |
port.runt.tx_length.set() |
Runt - TX Length |
291 |
Get Runt TX Length |
port.runt.tx_length.get() |
Runt - TX Length |
292 |
Get Runt Length Errors |
port.runt.has_length_errors.get() |
Runt - Length Error |
293 |
Enable RX Preamble Insert |
port.preamble.rx_insert.set() |
RX Preamble Insert |
294 |
Disable RX Preamble Insert |
port.preamble.rx_insert.set() |
RX Preamble Insert |
295 |
Get RX Preamble Insert |
port.preamble.rx_insert.get() |
RX Preamble Insert |
296 |
Enable TX Preamble Remove |
port.preamble.tx_remove.set() |
TX Preamble Removal |
297 |
Disable TX Preamble Remove |
port.preamble.tx_remove.set() |
TX Preamble Removal |
298 |
Get TX Preamble Remove |
port.preamble.tx_remove.get() |
TX Preamble Removal |
Port API Examples
# [ports]
"""Get Port Object"""
# Obtain a port object representing port 0 on module.
port = module.ports.obtain(0)
# [Check port type]
if isinstance(port, ports.PortChimera):
return
"""RESERVATION"""
"""Port Reservation"""
# Reserve or release the port.
await port.reservation.set_release()
await port.reservation.set_relinquish()
await port.reservation.set_reserve()
resp_obj = await port.reservation.get()
"""Port Reserved By"""
# Get the current reservation status of the port.
resp_obj = await port.reserved_by.get()
"""IDENTIFICATION"""
"""Port Description"""
# A text description of the port.
await port.comment.set(comment="description")
resp_obj = await port.comment.get()
# [Interface Name]
# The interface name of the port.
resp_obj = await port.interface.get()
"""CONTROL"""
"""Reset Port"""
# Reset the port. This will clear the port configuration, statistics, and any ongoing traffic.
await port.reset.set()
"""Flash Port LEDs"""
# Make the port LED flash on and off with a 1-second interval.
await port.flash.set_on()
await port.flash.set_off()
resp_obj = await port.flash.get()
"""TX CONTROL"""
"""Sync Status"""
# Get the port synchronization status.
resp_obj = await port.sync_status.get()
"""TX Output"""
# Enable or disable the port transmitter.
await port.tx_config.enable.set_on()
await port.tx_config.enable.set_off()
resp_obj = await port.tx_config.enable.get()
"""Tx Time Limit"""
# Set or get the transmit time limit.
await port.tx_config.time_limit.set(microseconds=1_000_000)
resp_obj = await port.tx_config.time_limit.get()
"""Tx Packet Limit"""
# Set or get the transmit packet limit.
await port.tx_config.packet_limit.set(packet_count_limit=1_000_000)
resp_obj = await port.tx_config.packet_limit.get()
"""Tx Time Elapsed"""
# Get the elapsed transmit time.
await port.tx_config.time.get()
"""Tx Delay"""
# Set or get the transmit delay.
await port.tx_config.delay.get()
"""Tx Prepare"""
# Prepare the port for transmission.
await port.tx_config.prepare.set()
"""Tx Start/Stop"""
# Start or stop traffic on the port.
await port.traffic.state.set_start()
await port.traffic.state.set_stop()
resp_obj = await port.traffic.state.get()
"""Traffic Error"""
# Get the traffic error status.
resp_obj = await port.traffic.error.get()
"""Transmit Manual Packet"""
# Transmit a single manually specified packet.
await port.tx_single_pkt.send.set(hex_data=Hex("00112233445566778899AABB080045000028000100004006B1E6C0A80001C0A80002"))
"""Transmit Manual Packet Time"""
# Get the time taken to transmit a single manually specified packet.
resp_obj = await port.tx_single_pkt.time.get()
"""Dynamic Traffic"""
# Enable or disable dynamic traffic on the port.
# When dynamic traffic is enabled, you can adjust traffic rate while the port is transmitting.
await port.dynamic.set_off()
await port.dynamic.set_on()
resp_obj = await port.dynamic.get()
"""TX PROFILE"""
"""Rate Fraction"""
# Set or get the transmit rate as a fraction of line rate in parts per million (ppm).
await port.rate.fraction.set(port_rate_ppm=1_000_000)
resp_obj = await port.rate.fraction.get()
"""Rate L2 Bits Per Second"""
# Set or get the transmit rate in bits per second at Layer 2.
await port.rate.l2_bps.set(port_rate_bps=1_000_000)
resp_obj = await port.rate.l2_bps.get()
"""Rate Frames Per Second"""
# Set or get the transmit rate in frames per second.
await port.rate.pps.set(port_rate_pps=10_000)
resp_obj = await port.rate.pps.get()
"""TX Mode"""
# Set or get the transmit mode.
await port.tx_config.mode.set_normal()
await port.tx_config.mode.set_burst()
await port.tx_config.mode.set_sequential()
await port.tx_config.mode.set_strictuniform()
resp_obj = await port.tx_config.mode.get()
"""Burst Period"""
# Set or get the burst period in milliseconds.
await port.tx_config.burst_period.set(burst_period=100)
resp_obj = await port.tx_config.burst_period.get()
"""BASIC LAYER-1 CONTROL"""
"""Inter-frame Gap"""
# Set or get the inter-frame gap minimum byte count.
await port.interframe_gap.set(min_byte_count=20)
resp_obj = await port.interframe_gap.get()
"""Speed Mode Selection"""
# Set or get the speed mode selection.
await port.speed.selection.set_auto()
await port.speed.selection.set_f10m()
await port.speed.selection.set_f10m100m()
await port.speed.selection.set_f10mhdx()
await port.speed.selection.set_f100m()
await port.speed.selection.set_f100m1g()
await port.speed.selection.set_f100m1g10g()
await port.speed.selection.set_f100m1g2500m()
await port.speed.selection.set_f100mhdx()
await port.speed.selection.set_f1g()
await port.speed.selection.set_f2500m()
await port.speed.selection.set_f5g()
await port.speed.selection.set_f10g()
await port.speed.selection.set_f40g()
await port.speed.selection.set_f100g()
resp_obj = await port.speed.selection.get()
"""Supported Speed Modes"""
resp_obj = await port.speed.supported.get()
"""Current Speed"""
resp_obj = await port.speed.current.get()
"""Speed Reduction"""
await port.speed.reduction.set(ppm=100)
resp_obj = await port.speed.reduction.get()
"""Optical Rx Power"""
resp_obj = await port.status.get()
"""LAYER-2 CONTROL"""
"""MAC Address"""
await port.net_config.mac.address.set(mac_address=Hex("000000000000"))
resp_obj = await port.net_config.mac.address.get()
"""Auto MAC Training"""
await port.net_config.mac.autotrain.set(interval=1)
resp_obj = await port.net_config.mac.autotrain.get()
"""PAUSE and PFC"""
await port.pause.set_on()
await port.pause.set_off()
resp_obj = await port.pause.get()
await port.pfc_enable.set(
cos_0=enums.OnOff.ON,
cos_1=enums.OnOff.OFF,
cos_2=enums.OnOff.ON,
cos_3=enums.OnOff.OFF,
cos_4=enums.OnOff.ON,
cos_5=enums.OnOff.OFF,
cos_6=enums.OnOff.ON,
cos_7=enums.OnOff.OFF,
)
resp_obj = await port.pfc_enable.get()
"""Gap Monitor"""
await port.gap_monitor.set(start=100, stop=10)
resp_obj = await port.gap_monitor.get()
"""PAYLOAD CONFIGURATION"""
"""Checksum Offset"""
await port.checksum.set(offset=14)
resp_obj = await port.checksum.get()
"""Random Seed"""
await port.random_seed.set(seed=1)
resp_obj = await port.random_seed.get()
"""Maximum Header Length"""
await port.max_header_length.set(max_header_length=56)
resp_obj = await port.max_header_length.get()
"""IMIX Weights"""
await port.mix.weights.set(
weight_56_bytes=0,
weight_60_bytes=0,
weight_64_bytes=70,
weight_70_bytes=15,
weight_78_bytes=15,
weight_92_bytes=0,
weight_256_bytes=0,
weight_496_bytes=0,
weight_512_bytes=0,
weight_570_bytes=0,
weight_576_bytes=0,
weight_594_bytes=0,
weight_1438_bytes=0,
weight_1518_bytes=0,
weight_9216_bytes=0,
weight_16360_bytes=0)
resp_obj = await port.mix.weights.get()
"""IMIX Lengths"""
await port.mix.lengths[0].set(frame_size=56)
await port.mix.lengths[1].set(frame_size=60)
await port.mix.lengths[14].set(frame_size=9216)
await port.mix.lengths[15].set(frame_size=16360)
resp_obj = await port.mix.lengths[0].get()
resp_obj = await port.mix.lengths[1].get()
resp_obj = await port.mix.lengths[14].get()
resp_obj = await port.mix.lengths[15].get()
"""Payload Mode"""
await port.payload_mode.set_normal()
await port.payload_mode.set_extpl()
await port.payload_mode.set_cdf()
resp_obj = await port.payload_mode.get()
"""TPLD Mode"""
await port.tpld_mode.set_normal()
await port.tpld_mode.set_micro()
resp_obj = await port.tpld_mode.get()
"""LOOPBACK"""
"""Loopback Modes"""
await port.loopback.set_none()
await port.loopback.set_l1rx2tx()
await port.loopback.set_l2rx2tx()
await port.loopback.set_l3rx2tx()
await port.loopback.set_port2port()
await port.loopback.set_txoff2rx()
await port.loopback.set_txon2rx()
resp_obj = await port.loopback.get()
"""LATENCY CONFIG"""
"""Latency Mode"""
await port.latency_config.mode.set_first2first()
await port.latency_config.mode.set_first2last()
await port.latency_config.mode.set_last2first()
await port.latency_config.mode.set_last2last()
resp_obj = await port.latency_config.mode.get()
"""Latency Offset"""
await port.latency_config.offset.set(offset=5)
resp_obj = await port.latency_config.offset.get()
"""IPV4 CONFIGURATION"""
"""IPv4: Address"""
await port.net_config.ipv4.address.set(
ipv4_address=ipaddress.IPv4Address("10.10.10.10"),
subnet_mask=ipaddress.IPv4Address("255.255.255.0"),
gateway=ipaddress.IPv4Address("10.10.1.1"),
wild=ipaddress.IPv4Address("0.0.0.0"))
resp_obj = await port.net_config.ipv4.address.get()
"""IPv4: Reply to ARP"""
await port.net_config.ipv4.arp_reply.set_on()
await port.net_config.ipv4.arp_reply.set_off()
resp_obj = await port.net_config.ipv4.arp_reply.get()
"""IPv4: Reply to Ping"""
await port.net_config.ipv4.ping_reply.set_on()
await port.net_config.ipv4.ping_reply.set_off()
resp_obj = await port.net_config.ipv4.ping_reply.get()
"""IPV6 CONFIGURATION"""
"""IPv6: Address"""
await port.net_config.ipv6.address.set(
ipv6_address=ipaddress.IPv6Address("fc00::0002"),
gateway=ipaddress.IPv6Address("fc00::0001"),
subnet_prefix=7,
wildcard_prefix=0
)
resp_obj = await port.net_config.ipv6.address.get()
"""IPv6: Reply to NDP"""
await port.net_config.ipv6.arp_reply.set_on()
await port.net_config.ipv6.arp_reply.set_off()
resp_obj = await port.net_config.ipv6.arp_reply.get()
"""IPv6: Reply to IPv6 Ping"""
await port.net_config.ipv6.ping_reply.set_on()
await port.net_config.ipv6.ping_reply.set_off()
resp_obj = await port.net_config.ipv6.ping_reply.get()
"""ARP/NDP TABLES"""
"""IPv4: ARP Table"""
arp_entry = ArpEntry(
ipv4_address=ipaddress.IPv4Address("12.12.12.12"),
prefix=24,
patched_mac=enums.OnOff.OFF,
mac_address=Hex("FFEEDDCCBBAA"))
await port.arp_rx_table.set(entries=[arp_entry])
resp_obj = await port.arp_rx_table.get()
"""IPv6: NDP Table"""
ndp_entry = NdpEntry(
ipv6_address=ipaddress.IPv6Address("fc00::1234"),
prefix=64,
patched_mac=enums.OnOff.OFF,
mac_address=Hex("AABBCCDDEEFF"))
await port.ndp_rx_table.set(entries=[ndp_entry])
resp_obj = await port.ndp_rx_table.get()
"""IPV4 MULTICAST CONFIGURATION"""
"""IPv4: Multicast"""
await port.multicast.mode.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastOperation.JOIN,
second_count=10)
await port.multicast.mode.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastOperation.LEAVE,
second_count=10)
await port.multicast.mode.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastOperation.OFF,
second_count=10)
await port.multicast.mode.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastOperation.ON,
second_count=10)
resp_obj = await port.multicast.mode.get()
"""IPv4: Multicast Advanced, IGMPv3"""
await port.multicast.mode_extended.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastExtOperation.EXCLUDE,
second_count=10,
igmp_version=enums.IGMPVersion.IGMPV3
)
await port.multicast.mode_extended.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastExtOperation.INCLUDE,
second_count=10,
igmp_version=enums.IGMPVersion.IGMPV3
)
"""IPv4: Multicast Advanced, IGMPv2"""
await port.multicast.mode_extended.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastExtOperation.JOIN,
second_count=10,
igmp_version=enums.IGMPVersion.IGMPV2
)
await port.multicast.mode_extended.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastExtOperation.LEAVE,
second_count=10,
igmp_version=enums.IGMPVersion.IGMPV2
)
await port.multicast.mode_extended.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastExtOperation.LEAVE_TO_ALL,
second_count=10,
igmp_version=enums.IGMPVersion.IGMPV2
)
await port.multicast.mode_extended.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastExtOperation.GENERAL_QUERY,
second_count=10,
igmp_version=enums.IGMPVersion.IGMPV2
)
await port.multicast.mode_extended.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastExtOperation.GROUP_QUERY,
second_count=10,
igmp_version=enums.IGMPVersion.IGMPV2
)
await port.multicast.mode_extended.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastExtOperation.ON,
second_count=10,
igmp_version=enums.IGMPVersion.IGMPV2
)
await port.multicast.mode_extended.set(
ipv4_multicast_addresses=[],
operation=enums.MulticastExtOperation.OFF,
second_count=10,
igmp_version=enums.IGMPVersion.IGMPV2
)
resp_obj = await port.multicast.mode_extended.get()
"""IPv4: Multicast Source List"""
await port.multicast.source_list.set(ipv4_addresses=[])
resp_obj = await port.multicast.source_list.get()
"""IPv4: Multicast Header"""
await port.multicast.header.set(header_count=1, header_format=enums.MulticastHeaderFormat.VLAN, tag=10, pcp=0, dei=enums.OnOff.OFF)
await port.multicast.header.set(header_count=0, header_format=enums.MulticastHeaderFormat.NOHDR, tag=10, pcp=0, dei=enums.OnOff.OFF)
resp_obj = await port.multicast.header.get()
"""UnAvailable Time (UAT) Mode"""
# Only applicable in Xena1564 test suite.
# This API enables the UAT detection on the port in Xena1564 test suite.
await port.uat.mode.set(mode=enums.OnOff.ON, delay=500)
await port.uat.mode.set(mode=enums.OnOff.OFF, delay=500)
resp = await port.uat.mode.get()
"""UAT Frame Loss Ratio"""
# Only applicable in Xena1564 test suite.
# This API specifies the Frame Loss Ratio threshold used to classify a
# one-second interval as a Severely Errored Second (SES).
# In accordance with Xena1564, Unavailable Time (UAT) is declared after
# the occurrence of 10 consecutive SES intervals.
resp = await port.uat.flr.get()
"""UAT Status"""
# Only applicable in Xena1564 test suite.
# This API retrieves the current UAT status of the port.
resp = await port.statistics.rx.uat.status.get()
"""UAT Time"""
# Only applicable in Xena1564 test suite.
# This API retrieves the total Unavailable Time (UAT) in seconds since the last reset.
resp = await port.statistics.rx.uat.time.get()
"""PACKET CAPTURE"""
"""Capture Trigger Criteria"""
await port.capturer.trigger.set(start_criteria=enums.StartTrigger.ON, start_criteria_filter=0, stop_criteria=enums.StopTrigger.FULL, stop_criteria_filter=0)
await port.capturer.trigger.set(start_criteria=enums.StartTrigger.ON, start_criteria_filter=0, stop_criteria=enums.StopTrigger.USERSTOP, stop_criteria_filter=0)
await port.capturer.trigger.set(start_criteria=enums.StartTrigger.FCSERR, start_criteria_filter=0, stop_criteria=enums.StopTrigger.FCSERR, stop_criteria_filter=0)
await port.capturer.trigger.set(start_criteria=enums.StartTrigger.PLDERR, start_criteria_filter=0, stop_criteria=enums.StopTrigger.PLDERR, stop_criteria_filter=0)
await port.capturer.trigger.set(start_criteria=enums.StartTrigger.FILTER, start_criteria_filter=0, stop_criteria=enums.StopTrigger.FILTER, stop_criteria_filter=0)
resp_obj = await port.capturer.trigger.get()
"""Frame to Keep"""
await port.capturer.keep.set(kind=enums.PacketType.ALL, index=0, byte_count=-1)
await port.capturer.keep.set(kind=enums.PacketType.FCSERR, index=0, byte_count=-1)
await port.capturer.keep.set(kind=enums.PacketType.NOTPLD, index=0, byte_count=-1)
await port.capturer.keep.set(kind=enums.PacketType.PLDERR, index=0, byte_count=-1)
await port.capturer.keep.set(kind=enums.PacketType.TPLD, index=5, byte_count=-1) # TPLD == 5
await port.capturer.keep.set(kind=enums.PacketType.FILTER, index=2, byte_count=-1) # Filter index == 2
resp_obj = await port.capturer.keep.get()
"""Capture State"""
await port.capturer.state.set_start()
await port.capturer.state.set_stop()
resp_obj = await port.capturer.state.get()
"""Capture Start Time, Status"""
resp_obj = await port.capturer.stats.get()
"""Read Captured Packets"""
pkts = await port.capturer.obtain_captured()
for i in range(len(pkts)):
resp_obj = await pkts[i].packet.get()
print(f"Packet content # {i}: {resp_obj.hex_data}")
"""PORT FILTER"""
"""Create Filter"""
filter = await port.filters.create()
"""Obtain One or Multiple Already Existing Filters"""
filter = port.filters.obtain(key=0)
filters = port.filters.obtain_multiple(*[0,1,2])
"""Sync Existing Filters from Port"""
await port.filters.server_sync()
"""Remove Filter using Index"""
await port.filters.remove(position_idx=0)
"""Remove Filter using Object"""
await filter.delete()
"""Get filter index"""
filter.idx
"""Get filter's module and port index"""
filter.kind.module_id
filter.kind.port_id
"""Filter - Enable"""
await filter.enable.set_on()
await filter.enable.set_off()
resp = await filter.enable.get()
"""Filter - Description"""
await filter.comment.set(comment="this is a comment")
resp = await filter.comment.get()
"""Filter - Condition"""
await filter.condition.set(and_expression_0=0, and_not_expression_0=0, and_expression_1=1, and_not_expression_1=0, and_expression_2=0, and_expression_3=0)
resp = await filter.condition.get()
"""Filter - String Representation"""
await filter.string.set(string_name="this is name")
resp = await filter.string.get()
"""PORT LENGTH TERMS"""
"""Create Length Term"""
length_term = await port.length_terms.create()
"""Obtain One or Multiple Already Existing Length Terms"""
length_term = port.length_terms.obtain(key=0)
length_terms = port.length_terms.obtain_multiple(*[0,1,2])
"""Remove Length Term using Index"""
await port.length_terms.remove(position_idx=0)
"""Remove Length Term using Object"""
await length_term.delete()
"""Get length term index"""
length_term.idx
"""Get length term's module and port index"""
length_term.kind.module_id
length_term.kind.port_id
"""Sync Existing Length Terms from Port"""
await port.length_terms.server_sync()
"""Configure Length Term - Length Value"""
await length_term.length.set(
length_check_type=enums.LengthCheckType.AT_MOST,
size=100)
await length_term.length.set(
length_check_type=enums.LengthCheckType.AT_LEAST,
size=100)
resp = await length_term.length.get()
"""PORT MATCH TERMS"""
"""Create Match Term"""
match_term = await port.match_terms.create()
"""Obtain One or Multiple Already Existing Match Terms"""
match_term = port.match_terms.obtain(key=0)
match_terms = port.match_terms.obtain_multiple(*[0,1,2])
"""Remove Match Term using Index"""
await port.match_terms.remove(position_idx=0)
"""Remove Match Term using Object"""
await match_term.delete()
"""Get match term index"""
match_term.idx
"""Get match term's module and port index"""
match_term.kind.module_id
match_term.kind.port_id
"""Sync Existing Match Terms from Port"""
await port.match_terms.server_sync()
"""Configure Match Term - Match Value"""
await match_term.match.set(mask=Hex("FF"), value=Hex("00"))
resp = await match_term.match.get()
"""Configure Match Term - Position"""
await match_term.position.set(byte_offset=0)
resp = await match_term.position.get()
"""Configure Match Term - Protocol Segments"""
await match_term.protocol.set(segments=[enums.ProtocolOption.VLAN])
resp = await match_term.protocol.get()
"""PORT HISTOGRAM"""
"""Create Histogram"""
histogram = await port.datasets.create()
"""Obtain One or Multiple Already Existing Histograms"""
histogram = port.datasets.obtain(key=0)
histograms = port.datasets.obtain_multiple(*[0,1,2])
"""Get histogram index"""
histogram.idx
"""Get histogram's module and port index"""
histogram.kind.module_id
histogram.kind.port_id
"""Sync Existing Histograms from Port"""
await port.datasets.server_sync()
"""Remove Histogram using index"""
await port.datasets.remove(position_idx=0)
"""Remove Histogram using object"""
await histogram.delete()
"""Enable Histogram"""
await histogram.enable.set_on()
await histogram.enable.set_off()
resp = await histogram.enable.get()
"""Configure Histogram - Data Source"""
await histogram.source.set(
source_type=enums.SourceType.TX_IFG,
which_packets=enums.PacketDetailSelection.ALL,
identity=0
)
await histogram.source.set(
source_type=enums.SourceType.TX_LEN,
which_packets=enums.PacketDetailSelection.ALL,
identity=0
)
await histogram.source.set(
source_type=enums.SourceType.RX_IFG,
which_packets=enums.PacketDetailSelection.ALL,
identity=0
)
await histogram.source.set(
source_type=enums.SourceType.RX_LEN,
which_packets=enums.PacketDetailSelection.ALL,
identity=0
)
await histogram.source.set(
source_type=enums.SourceType.RX_LATENCY,
which_packets=enums.PacketDetailSelection.ALL,
identity=0
)
await histogram.source.set(
source_type=enums.SourceType.RX_JITTER,
which_packets=enums.PacketDetailSelection.ALL,
identity=0
)
resp = await histogram.source.get()
"""Configure Histogram - Bucket Range"""
await histogram.range.set(
start=1, #first value going into the second bucket
step=1, # the span of each middle bucket: (1) 1,2,4,8,16,32,64,128,256,512 (bytes, non-latency histograms).(2) 16,32,64,128,...,1048576,2097152 (nanoseconds, latency histograms).
bucket_count=10 # the total number of buckets
)
resp = await histogram.range.get()
"""Get Histogram Samples"""
resp = await histogram.samples.get()
"""PORT STATISTICS"""
"""Port Error Counter"""
resp = await port.errors_count.get()
"""TX STATISTICS"""
"""Clear Tx Statistics"""
# Clear all the transmit statistics for a port.
await port.statistics.tx.clear.set()
"""Total Tx Traffic Counters"""
# Obtains statistics concerning all the packets
# transmitted on the port.
# Includes: bytes transmitted since cleared,
# packets transmitted since cleared,
# bits per second,
# packets per second.
resp = await port.statistics.tx.total.get()
resp.byte_count_since_cleared
resp.packet_count_since_cleared
resp.bit_count_last_sec
resp.packet_count_last_sec
"""Tx Non-TPLD Traffic Counter"""
# Obtains statistics concerning the packets without
# a test payload transmitted on the port.
# Includes: bytes transmitted since cleared,
# packets transmitted since cleared,
# bits per second,
# packets per second.
resp = await port.statistics.tx.no_tpld.get()
resp.byte_count_since_cleared
resp.packet_count_since_cleared
resp.bit_count_last_sec
resp.packet_count_last_sec
"""Extra Tx Counters"""
# Obtains additional statistics for packets
# transmitted on the port.
# Includes: transmitted ARP Requests,
# transmitted ARP Replies,
# transmitted IPv4 ICMP Echo Requests,
# transmitted IPv4 ICMP Echo Replies,
# transmitted FCS error packets,
# transmitted sequence error packets,
# transmitted misordered error packets,
# transmitted payload error packets,
# transmitted TPLD error packets,
# transmitted MAC training frames,
# transmitted IGMP Join packets.
resp = await port.statistics.tx.extra.get()
resp.tx_arp_request_count
resp.tx_arp_reply_count
resp.tx_ping_request_count
resp.tx_ping_reply_count
resp.tx_fcs_inj_count
resp.tx_seq_inj_count
resp.tx_mis_inj_count
resp.tx_pld_inj_count
resp.tx_tpld_inj_count
resp.tx_mac_train_count
resp.tx_igmp_join_count
"""Tx Traffic Counters Per Stream Index"""
# Obtains statistics concerning the packets of
# a specific stream transmitted on the port.
resp = await port.statistics.tx.obtain_from_stream(stream=0).get()
resp.byte_count_since_cleared
resp.packet_count_since_cleared
resp.bit_count_last_sec
resp.packet_count_last_sec
"""RX STATISTICS"""
"""Clear Rx Statistics"""
# Clear all the receive statistics for a port.
await port.statistics.rx.clear.set()
"""Calibrate Rx Latency"""
# Calibrate the latency calculation for packets received on the port.
# The lowest detected latency value (across all Test Payload IDs)
# will be set as the new base.
await port.statistics.rx.calibrate.set()
"""Total Rx Traffic Counters"""
# Obtains statistics concerning all the packets received on the port.
# Includes: bytes received since cleared,
# packets received since cleared,
# bits per second,
# packets per second.
resp = await port.statistics.rx.total.get()
resp.byte_count_since_cleared
resp.packet_count_since_cleared
resp.bit_count_last_sec
resp.packet_count_last_sec
"""Rx Non-TPLD Traffic Counter"""
# Obtains statistics concerning the packets without
# a test payload received on the port.
# Includes: bytes received since cleared,
# packets received since cleared,
# bits per second,
# packets per second.
resp = await port.statistics.rx.no_tpld.get()
resp.byte_count_since_cleared
resp.packet_count_since_cleared
resp.bit_count_last_sec
resp.packet_count_last_sec
"""Rx PFC Counters"""
# Obtains statistics of received
# Priority Flow Control (PFC) packets on the port.
resp = await port.statistics.rx.pfc_stats.get()
resp.packet_count
resp.quanta_pri_0
resp.quanta_pri_1
resp.quanta_pri_2
resp.quanta_pri_3
resp.quanta_pri_4
resp.quanta_pri_5
resp.quanta_pri_6
resp.quanta_pri_7
"""Extra Rx Counters"""
# Obtains statistics concerning special errors
# received on the port since received statistics were cleared.
# Includes: received ARP Requests,
# received ARP Replies,
# received IPv4 ICMP Echo Requests,
# received IPv4 ICMP Echo Replies,
# received FCS error packets,
# received PAUSE frames,
# received gap count,
# received gap duration.
resp = await port.statistics.rx.extra.get()
resp.fcs_error_count
resp.pause_frame_count
resp.gap_count
resp.gap_duration
resp.pause_frame_count
resp.rx_arp_reply_count
resp.rx_arp_request_count
resp.rx_ping_reply_count
resp.rx_ping_request_count
"""Received TPLDs"""
# Obtain the set of test payload IDs observed
# among the received packets since receive statistics were cleared.
# Traffic statistics for these test payload streams
# will have non-zero byte and packet count.
rx_tplds = await port.statistics.rx.obtain_available_tplds()
"""Rx Error Counters Per TPLD"""
# Obtains statistics concerning errors in the packets
# with a particular test payload id received on the port.
resp = await port.statistics.rx.access_tpld(tpld_id=0).errors.get()
resp.packet_loss_by_seq
resp.payload_err_packets
resp.misorder_by_seq
"""Rx Latency Counter Per TPLD"""
# Obtains statistics concerning the latency
# experienced by the packets with a particular
# test payload id received on the port.
resp = await port.statistics.rx.access_tpld(tpld_id=0).latency.get()
resp.avg_last_sec
resp.max_last_sec
resp.min_last_sec
resp.avg_val
resp.max_val
resp.min_val
"""Rx Jitter Counter Per TPLD"""
# Obtains statistics concerning the jitter
# experienced by the packets with a particular
# test payload id received on the port.
resp = await port.statistics.rx.access_tpld(tpld_id=0).jitter.get()
resp.avg_last_sec
resp.max_last_sec
resp.min_last_sec
resp.avg_val
resp.max_val
resp.min_val
"""Rx Traffic Counters Per TPLD"""
# Obtains traffic statistics concerning the packets
# with a particular test payload identifier received
# on the port.
resp = await port.statistics.rx.access_tpld(tpld_id=0).traffic.get()
resp.byte_count_since_cleared
resp.packet_count_since_cleared
resp.bit_count_last_sec
resp.packet_count_last_sec
"""Rx Filtered Traffic Counters"""
# Obtains statistics concerning the packets
# satisfying the condition of a particular
# filter for the port.
resp = await port.statistics.rx.obtain_filter_statistics(filter=0).get()
resp.byte_count_since_cleared
resp.packet_count_since_cleared
resp.bit_count_last_sec
resp.packet_count_last_sec
# [Z01s/Z01sx Odin Specific APIs]
# [MDI/MDIX]
if isinstance(port, ports.POdin1G3S6P) or isinstance(port, ports.POdin1G3S6P_b) or isinstance(port, ports.POdin1G3S6PE):
"""MDI/MDIX Mode"""
await port.mdix_mode.set_auto()
await port.mdix_mode.set_mdi()
await port.mdix_mode.set_mdix()
resp_obj = await port.mdix_mode.get()
# [Z01t Odin Specific APIs]
# [BroadR Reach]
if isinstance(port, ports.POdin1G3S6PT1RJ45):
"""BroadR Reach Configuration"""
await port.brr.mode.set_master()
await port.brr.mode.set_slave()
resp_obj = await port.brr.mode.get()
"""BroadR Reach Status"""
resp_obj = await port.brr.status.get()
# [Z10r Odin Specific APIs]
if isinstance(port, ports.POdin10G5S6PCU) or isinstance(port, ports.POdin10G5S6PCU_b):
# [Auto Negotiation]
"""Auto Negotiation Configuration"""
await port.autoneg_selection.set_on()
await port.autoneg_selection.set_off()
resp_obj = await port.autoneg_selection.get()
# [Energy Efficient Ethernet (EEE)]
"""ENERGY EFFICIENT ETHERNET (EEE) CONFIGURATION"""
"""EEE- Capabilities"""
resp_obj = await port.eee.capabilities.get()
"""EEE - Partner Capabilities"""
resp_obj = await port.eee.partner_capabilities.get()
"""EEE - Control"""
await port.eee.enable.set_off()
await port.eee.enable.set_on()
resp_obj = await port.eee.enable.get()
"""EEE - Low Power TX Mode"""
await port.eee.mode.set_off()
await port.eee.mode.set_on()
resp_obj = await port.eee.mode.get()
"""EEE - RX Power"""
resp_obj = await port.eee.rx_power.get()
"""EEE - SNR Margin"""
resp_obj = await port.eee.snr_margin.get()
"""EEE - Status"""
resp_obj = await port.eee.status.get()
if isinstance(port, ports.POdin10G6S6P_a):
# [Runt Frame Handling]
"""Runt Frame Handling"""
# Runt - RX Length
await port.runt.rx_length.set(runt_length=40)
resp_obj = await port.runt.rx_length.get()
# Runt - TX Length
await port.runt.tx_length.set(runt_length=40)
resp_obj = await port.runt.tx_length.get()
# Runt - Length Error
resp_obj = await port.runt.has_length_errors.get()
# [Preamble Handling]
"""Preamble Handling"""
# RX Preamble Insert
await port.preamble.rx_insert.set(on_off=enums.OnOff.ON)
await port.preamble.rx_insert.set(on_off=enums.OnOff.OFF)
resp_obj = await port.preamble.rx_insert.get()
# TX Preamble Removal
await port.preamble.tx_remove.set(on_off=enums.OnOff.ON)
await port.preamble.tx_remove.set(on_off=enums.OnOff.OFF)
resp_obj = await port.preamble.tx_remove.get()
Stream APIs Summary Table
Id |
Action |
API |
Description |
|---|---|---|---|
1 |
Create Stream |
port.streams.create() |
Create a new stream object on the port |
2 |
Obtain Stream |
port.streams.obtain() |
Obtain an existing stream object using stream index |
3 |
Obtain Multiple Streams |
port.streams.obtain_multiple() |
Obtain multiple existing stream objects |
4 |
Sync Streams |
port.streams.server_sync() |
Sync existing streams from port |
5 |
Get Stream Index |
stream.idx |
Get the stream index |
6 |
Get Stream Module/Port Index |
stream.kind.module_id/port_id |
Get stream’s module and port index |
7 |
Remove Stream by Index |
port.streams.remove() |
Remove a stream from the port by index |
8 |
Delete Stream |
stream.delete() |
Remove a stream from the port using object |
9 |
Set Stream Description |
stream.comment.set() |
Set a text description for the stream |
10 |
Get Stream Description |
stream.comment.get() |
Get the stream description |
11 |
Set Test Payload ID |
stream.tpld_id.set() |
Set the test payload identifier for the stream |
12 |
Get Test Payload ID |
stream.tpld_id.get() |
Get the test payload identifier for the stream |
13 |
Disable Stream |
stream.enable.set_off() |
Disable the stream |
14 |
Enable Stream |
stream.enable.set_on() |
Enable the stream |
15 |
Suppress Stream |
stream.enable.set_suppress() |
Suppress the stream |
16 |
Get Stream Enable Status |
stream.enable.get() |
Get the stream enable status |
17 |
Set Stream Rate Fraction |
stream.rate.fraction.set() |
Set stream rate as fraction of port rate (ppm) |
18 |
Get Stream Rate Fraction |
stream.rate.fraction.get() |
Get stream rate fraction |
19 |
Set Stream Rate PPS |
stream.rate.pps.set() |
Set stream rate in packets per second |
20 |
Get Stream Rate PPS |
stream.rate.pps.get() |
Get stream rate in packets per second |
21 |
Set Stream Rate L2 BPS |
stream.rate.l2bps.set() |
Set stream rate in L2 bits per second |
22 |
Get Stream Rate L2 BPS |
stream.rate.l2bps.get() |
Get stream rate in L2 bits per second |
23 |
Set Stream Packet Limit |
stream.packet.limit.set() |
Set the total number of packets for the stream |
24 |
Get Stream Packet Limit |
stream.packet.limit.get() |
Get the stream packet limit |
25 |
Set Burst Size and Density |
stream.burst.burstiness.set() |
Set burst size and density for the stream |
26 |
Get Burst Size and Density |
stream.burst.burstiness.get() |
Get burst size and density configuration |
27 |
Set Burst Gap |
stream.burst.gap.set() |
Set inter-packet and inter-burst gaps |
28 |
Get Burst Gap |
stream.burst.gap.get() |
Get burst gap configuration |
29 |
Set Priority Flow Control |
stream.priority_flow.set() |
Set PFC CoS value for the stream |
30 |
Get Priority Flow Control |
stream.priority_flow.get() |
Get PFC CoS value for the stream |
31 |
Set Packet Length FIXED |
stream.packet.length.set() |
Set fixed packet length for the stream |
32 |
Set Packet Length INCREMENTING |
stream.packet.length.set() |
Set incrementing packet length for the stream |
33 |
Set Packet Length BUTTERFLY |
stream.packet.length.set() |
Set butterfly packet length for the stream |
34 |
Set Packet Length RANDOM |
stream.packet.length.set() |
Set random packet length for the stream |
35 |
Set Packet Length MIX |
stream.packet.length.set() |
Set mixed packet length for the stream |
36 |
Get Packet Length |
stream.packet.length.get() |
Get packet length configuration |
37 |
Auto Adjust Packet Size |
stream.packet.auto_adjust.set() |
Adjust packet length distribution of the stream |
38 |
Set Protocol Segments |
stream.packet.header.protocol.set() |
Set header protocol segments for the stream |
39 |
Get Protocol Segments |
stream.packet.header.protocol.get() |
Get header protocol segments |
40 |
Set Packet Header Data |
stream.packet.header.data.set() |
Set packet header data using hex or headers object |
41 |
Get Packet Header Data |
stream.packet.header.data.get() |
Get packet header data |
42 |
Set Payload Pattern |
stream.payload.content.set() |
Set payload content pattern for the stream |
43 |
Set Payload INC16 |
stream.payload.content.set_inc_word() |
Set 16-bit incrementing payload |
44 |
Set Payload INC8 |
stream.payload.content.set_inc_byte() |
Set 8-bit incrementing payload |
45 |
Set Payload DEC8 |
stream.payload.content.set_dec_byte() |
Set 8-bit decrementing payload |
46 |
Set Payload DEC16 |
stream.payload.content.set_dec_word() |
Set 16-bit decrementing payload |
47 |
Set Payload PRBS |
stream.payload.content.set_prbs() |
Set PRBS payload |
48 |
Set Payload RANDOM |
stream.payload.content.set_random() |
Set random payload |
49 |
Get Payload Content |
stream.payload.content.get() |
Get payload content configuration |
50 |
Set Extended Payload |
stream.payload.extended.set() |
Set extended payload (up to MTU) for the stream |
51 |
Get Extended Payload |
stream.payload.extended.get() |
Get extended payload data |
52 |
Set CDF Offset |
stream.cdf.offset.set() |
Set Custom Data Field offset for the stream |
53 |
Get CDF Offset |
stream.cdf.offset.get() |
Get CDF offset value |
54 |
Set CDF Count |
stream.cdf.count.set() |
Set number of custom data fields for the stream |
55 |
Get CDF Count |
stream.cdf.count.get() |
Get number of custom data fields |
56 |
Set CDF Data |
stream.cdf.data().set() |
Set custom data field data for the stream |
57 |
Get CDF Data |
stream.cdf.data().get() |
Get custom data field data |
58 |
Set IPv4 Gateway |
stream.gateway.ipv4.set() |
Set IPv4 gateway address for the stream |
59 |
Get IPv4 Gateway |
stream.gateway.ipv4.get() |
Get IPv4 gateway address |
60 |
Set IPv6 Gateway |
stream.gateway.ipv6.set() |
Set IPv6 gateway address for the stream |
61 |
Get IPv6 Gateway |
stream.gateway.ipv6.get() |
Get IPv6 gateway address |
62 |
Request ARP Resolution |
stream.request.arp.get() |
Generate ARP request for peer address resolution |
63 |
Request PING Check |
stream.request.ping.get() |
Generate PING request to check IP peer |
64 |
Inject FCS Error |
stream.inject_err.frame_checksum.set() |
Inject frame checksum error in transmitted packet |
65 |
Inject Misorder Error |
stream.inject_err.misorder.set() |
Inject misorder error by swapping sequence numbers |
66 |
Inject Payload Integrity Error |
stream.inject_err.payload_integrity.set() |
Inject payload integrity error in packet |
67 |
Inject Sequence Error |
stream.inject_err.sequence.set() |
Inject sequence error by skipping sequence number |
68 |
Inject Test Payload Error |
stream.inject_err.test_payload.set() |
Inject test payload error in packet |
69 |
Configure 16/24-bit Modifiers |
stream.packet.header.modifiers.configure() |
Create 16/24-bit modifiers for the stream |
70 |
Clear 16/24-bit Modifiers |
stream.packet.header.modifiers.clear() |
Clear all 16/24-bit modifiers |
71 |
Obtain 16/24-bit Modifier |
stream.packet.header.modifiers.obtain() |
Obtain a 16/24-bit modifier object |
72 |
Set Modifier Range |
modifier.range.set() |
Set min/max/step for modifier range |
73 |
Get Modifier Range |
modifier.range.get() |
Get modifier range configuration |
74 |
Set Modifier Specification |
modifier.specification.set() |
Set modifier position, mask, action, repetition |
75 |
Get Modifier Specification |
modifier.specification.get() |
Get modifier specification configuration |
76 |
Set Modifier Endianness |
modifier.endian.set() |
Set modifier endianness (BIG or LITTLE) |
77 |
Get Modifier Endianness |
modifier.endian.get() |
Get modifier endianness setting |
78 |
Configure 32-bit Modifiers |
stream.packet.header.modifiers_extended.configure() |
Create 32-bit modifiers for the stream |
79 |
Clear 32-bit Modifiers |
stream.packet.header.modifiers_extended.clear() |
Clear all 32-bit modifiers |
80 |
Obtain 32-bit Modifier |
stream.packet.header.modifiers_extended.obtain() |
Obtain a 32-bit modifier object |
81 |
Set 32-bit Modifier Range |
modifier_ext.range.set() |
Set min/max/step for 32-bit modifier range |
82 |
Get 32-bit Modifier Range |
modifier_ext.range.get() |
Get 32-bit modifier range configuration |
83 |
Set 32-bit Modifier Specification |
modifier_ext.specification.set() |
Set 32-bit modifier position, mask, action, repetition |
84 |
Get 32-bit Modifier Specification |
modifier_ext.specification.get() |
Get 32-bit modifier specification configuration |
Stream API Examples
# [streams]
"""Create Stream Object"""
# Create a stream object representing a specific stream on the port.
stream = await port.streams.create()
"""Obtain One or Multiple Stream Objects"""
# Obtain already existing stream object(s) using stream indices.
stream = port.streams.obtain(0)
streams = port.streams.obtain_multiple(*[0,1,2])
"""Sync Existing Streams from Port"""
await port.streams.server_sync()
"""Get stream index"""
stream.idx
"""Get stream's module and port index"""
stream.kind.module_id
stream.kind.port_id
"""Remove Stream by Index"""
# Remove a stream object from the port.
await port.streams.remove(position_idx=0)
"""Remove Stream by Removing Stream Object"""
# Remove a stream object from the port.
await stream.delete()
"""Description"""
await stream.comment.set(comment="description")
resp = await stream.comment.get()
"""Test Payload ID"""
# The identifier of the test payloads inserted
# into packets transmitted for a stream.
# A value of -1 disables test payloads for the stream.
await stream.tpld_id.set(test_payload_identifier=0)
resp = await stream.tpld_id.get()
"""Enable/Disable Stream"""
# This property determines if a stream contributes
# outgoing packets for a port. The value can be
# toggled between ON and SUPPRESS while traffic is enabled
# at the port level.
# Streams in the OFF state cannot be set to any other value
# while traffic is enabled.
# The sum of the rates of all enabled or suppressed streams
# must not exceed the effective port rate.
await stream.enable.set_off()
await stream.enable.set_on()
await stream.enable.set_suppress()
resp = await stream.enable.get()
"""Stream Rate Configuration"""
# The rate of the traffic transmitted for a stream
# expressed in millionths of the effective rate for
# the port.
await stream.rate.fraction.set(stream_rate_ppm=1_000_000)
resp = await stream.rate.fraction.get()
# The rate of the traffic transmitted for a stream
# expressed in packets per second.
await stream.rate.pps.set(stream_rate_pps=1_000)
resp = await stream.rate.pps.get()
# The rate of the traffic transmitted for a stream,
# expressed in units of bits-per-second at layer-2,
# thus including the Ethernet header but excluding the
# inter-frame gap.
await stream.rate.l2bps.set(l2_bps=1_000_000)
resp = await stream.rate.l2bps.get()
"""Stream Packet Limit"""
# The total number of packets to be transmitted for a stream.
# A value of 0 indicates that there is no limit to the
# number of packets transmitted for the stream.
await stream.packet.limit.set(packet_count=1_000)
resp = await stream.packet.limit.get()
"""Burst Size and Density"""
# The burstiness of the traffic transmitted for a
# stream, expressed in terms of the number of packets
# in each burst, and how densely they are packed together.
await stream.burst.burstiness.set(size=20, density=80)
resp = await stream.burst.burstiness.get()
"""Inter Burst/Packet Gap"""
# When the port is in in Burst TX mode, this command
# defines the gap between packets in a burst
# (inter-packet gap) and the gap after a burst defined
# in one stream stops until a burst defined in the next
# stream starts (inter-burst gap).
await stream.burst.gap.set(inter_packet_gap=30, inter_burst_gap=30)
resp = await stream.burst.gap.get()
"""Priority Flow"""
# Set and get the Priority Flow Control (PFC) Cos
# value of a stream.
await stream.priority_flow.set(cos=enums.PFCMode.ZERO)
await stream.priority_flow.set(cos=enums.PFCMode.ONE)
await stream.priority_flow.set(cos=enums.PFCMode.TWO)
await stream.priority_flow.set(cos=enums.PFCMode.THREE)
await stream.priority_flow.set(cos=enums.PFCMode.FOUR)
await stream.priority_flow.set(cos=enums.PFCMode.FIVE)
await stream.priority_flow.set(cos=enums.PFCMode.SIX)
await stream.priority_flow.set(cos=enums.PFCMode.SEVEN)
await stream.priority_flow.set(cos=enums.PFCMode.VLAN_PCP)
resp = await stream.priority_flow.get()
"""Packet Size Configuration"""
# The length distribution of the packets transmitted for
# a stream. The length of the packets transmitted for a
# stream can be varied from packet to packet, according
# to a choice of distributions within a specified min...
# max range.
await stream.packet.length.set(length_type=enums.LengthType.FIXED, min_val=64, max_val=64)
await stream.packet.length.set(length_type=enums.LengthType.INCREMENTING, min_val=64, max_val=1500)
await stream.packet.length.set(length_type=enums.LengthType.BUTTERFLY, min_val=64, max_val=1500)
await stream.packet.length.set(length_type=enums.LengthType.RANDOM, min_val=64, max_val=1500)
await stream.packet.length.set(length_type=enums.LengthType.MIX, min_val=64, max_val=64)
resp = await stream.packet.length.get()
resp.length_type
"""Packet Auto Size"""
# Adjust the packet length distribution
# of the stream.
await stream.packet.auto_adjust.set()
"""Packet Header Configuration"""
# Header Protocol Segments
await stream.packet.header.protocol.set(
segments=[
enums.ProtocolOption.ETHERNET,
enums.ProtocolOption.VLAN,
enums.ProtocolOption.IP,
enums.ProtocolOption.UDP
])
resp = await stream.packet.header.protocol.get()
# Header Values
# Use hlfuncs.headers to build packet headers for the stream.
eth = headers.Ethernet()
eth.src_mac = "aaaa.aaaa.0005"
eth.dst_mac = "bbbb.bbbb.0005"
eth.ethertype = headers.EtherType.VLAN
vlan = headers.VLAN()
vlan.pri = 3
vlan.dei = 0
vlan.id = 100
vlan.type = headers.EtherType.IPv4
ipv4 = headers.IPV4()
ipv4.src = "1.1.1.5"
ipv4.dst = "2.2.2.5"
ipv4.proto = headers.IPProtocol.UDP
udp = headers.UDP()
udp.src_port = 5005
udp.dst_port = 6006
udp.length = 0
await stream.packet.header.data.set(hex_data=Hex(str(eth) + str(vlan) + str(ipv4) + str(udp)))
resp = await stream.packet.header.data.get()
# Alternatively, you can also use raw hex data.
await stream.packet.header.data.set(hex_data=Hex("bbbbbbbbbbbbaaaaaaaabbbb8100006408004500002e000100004011f9b201010105020202050505060"))
resp = await stream.packet.header.data.get()
"""Payload Type"""
# The payload content of the packets transmitted for
# a stream. The payload portion of a packet starts
# after the header and continues up until the test
# payload or the frame checksum.
# Pattern string in hex, min = 1 byte, max = 18 bytes
await stream.payload.content.set(payload_type=enums.PayloadType.PATTERN, hex_data=Hex("000102030405060708090A0B0C0D0E0FDEAD"))
await stream.payload.content.set(payload_type=enums.PayloadType.PATTERN, hex_data=Hex("F5"))
# Patter string ignored for non-pattern types
await stream.payload.content.set(payload_type=enums.PayloadType.INC16, hex_data=Hex("F5"))
await stream.payload.content.set_inc_word("00")
await stream.payload.content.set(payload_type=enums.PayloadType.INC8, hex_data=Hex("F5"))
await stream.payload.content.set_inc_byte("00")
await stream.payload.content.set(payload_type=enums.PayloadType.DEC8, hex_data=Hex("F5"))
await stream.payload.content.set_dec_byte("00")
await stream.payload.content.set(payload_type=enums.PayloadType.DEC16, hex_data=Hex("F5"))
await stream.payload.content.set_dec_word("00")
await stream.payload.content.set(payload_type=enums.PayloadType.PRBS, hex_data=Hex("F5"))
await stream.payload.content.set_prbs("00")
await stream.payload.content.set(payload_type=enums.PayloadType.RANDOM, hex_data=Hex("F5"))
await stream.payload.content.set_random("00")
resp = await stream.payload.content.get()
"""Extended Payload"""
# This API allows the definition of a much larger (up to MTU) payload buffer for each stream. The extended payload will be inserted immediately after the end of the protocol segment area.
# To use this API, the port must be in Extended
# Payload Mode. This mode is set at the port level
# using await port.payload_mode.set_extpl()
await stream.payload.extended.set(hex_data=Hex("00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF00110022FF"))
resp = await stream.payload.extended.get()
"""Custom Data Field"""
# To use CDF API, the port must be in Custom Data Field Mode. This mode is set at the port level
# using await port.payload_mode.set_cdf()
# CDF Offset
# The CDF offset for the stream is the location in
# the stream data packets where the various CDF data
# will be inserted. All fields for a given stream
# uses the same offset value.
await stream.cdf.offset.set(offset=1)
resp = await stream.cdf.offset.get()
# CDF Count
# Controls the number of custom data fields
# available for each stream. You can set a different
# number of fields for each stream.
await stream.cdf.count.set(cdf_count=1)
resp = await stream.cdf.count.get()
# CDF Data
# Controls the actual field data for a single field. It is possible to define fields with different data lengths for each stream. If the length of a data field exceeds (packet length - CDF offset) defined for the stream the field data will be truncated when transmitted.
await stream.cdf.data(0).set(hex_data=Hex("AABBCCDD"))
resp = await stream.cdf.data(0).get()
"""IPv4 Gateway Address"""
# An IPv4 gateway configuration specified for the stream.
await stream.gateway.ipv4.set(gateway=ipaddress.IPv4Address("10.10.10.1"))
resp = await stream.gateway.ipv4.get()
"""IPv6 Gateway Address"""
# An IPv6 gateway configuration specified for the stream.
await stream.gateway.ipv6.set(gateway=ipaddress.IPv6Address("2001:0db8:85a3:0000:0000:8a2e:0370:7334"))
resp = await stream.gateway.ipv6.get()
"""ARP Resolve Peer Address"""
# Generates an outgoing ARP request on the test port.
# The packet header for the stream must contain an IP
# protocol segment, and the destination IP address is
# used in the ARP request.
resp = await stream.request.arp.get()
resp.mac_address
"""PING Check IP Peer"""
# Generates an outgoing ping request using the ICMP
# protocol on the test port. The packet header for
# the stream must contain an IP protocol segment,
# with valid source and destination IP addresses.
resp = await stream.request.ping.get()
resp.delay
resp.time_to_live
"""FCS Error Injection"""
# Force a frame checksum error in one of the packets
# currently being transmitted from a stream.
await stream.inject_err.frame_checksum.set()
"""Misorder Error Injection"""
# Force a misorder error by swapping the test payload
# sequence numbers in two of the packets currently
# being transmitted from a stream.
await stream.inject_err.misorder.set()
"""Payload Integrity Error Injection"""
# Force a payload integrity error in one of the
# packets currently being transmitted from a stream.
# Payload integrity validation is only available for
# incrementing payloads, and the error is created by
# changing a byte from the incrementing sequence.
await stream.inject_err.payload_integrity.set()
"""Sequence Error Injection"""
# Force a sequence error by skipping a test payload
# sequence number in one of the packets currently
# being transmitted from a stream.
await stream.inject_err.sequence.set()
"""Test Payload Error Injection"""
# Force a test payload error in one of the packets
# currently being transmitted from a stream. This
# means that the test payload will not be recognized
# at the receiving port, so it will be counted as a
# no-test-payload packet, and there will be a lost
# packet for the stream.
await stream.inject_err.test_payload.set()
"""MODIFIERS"""
"""16/24-bit modifier"""
"""Create modifier"""
await stream.packet.header.modifiers.configure(number=1)
"""Clear modifiers"""
await stream.packet.header.modifiers.clear()
"""Obtain modifier object"""
# Must create modifiers before obtain.
modifier = stream.packet.header.modifiers.obtain(idx=0)
"""Modifier Range Configuration"""
# Range specification for a packet modifier for a
# stream header, specifying which values the modifier
# should take on. This applies only to incrementing
# and decrementing modifiers; random modifiers always
# produce every possible bit pattern.
await modifier.range.set(min_val=0, step=10, max_val=9)
resp = await modifier.range.get()
resp.min_val
resp.max_val
resp.step
"""Modifier Position, Action, Mask"""
# A packet modifier for a stream header. The headers
# of each packet transmitted for the stream will be
# varied according to the modifier specification.
# This command requires two sub-indices, one for the
# stream and one for the modifier.
# A modifier is positioned at a fixed place in the
# header, selects a number of consecutive bits
# starting from that position, and applies an action
# to those bits in each packet. Packets can be
# repeated so that a certain number of identical
# packets are transmitted before applying the next
# modification.
await modifier.specification.set(position=0, mask=Hex("FFFF0000"), action=enums.ModifierAction.INC, repetition=1)
await modifier.specification.set(position=0, mask=Hex("FFFF0000"), action=enums.ModifierAction.DEC, repetition=1)
await modifier.specification.set(position=0, mask=Hex("FFFF0000"), action=enums.ModifierAction.RANDOM, repetition=1)
resp = await modifier.specification.get()
resp.action
resp.mask
resp.position
resp.repetition
"""Endianness"""
# Network byte order is Big Endian, where the MSB is
# assigned with the smallest address. Xena’s modifier
# (16-bit, 24-bit, or 32-bit) inc/dec mode is default
# to BIG, which inc/dec starts from the LSB (the
# largest address). The user can set the mode to
# LITTLE, which the modifier inc/dec starts from the
# MSB (the smallest address)
await modifier.endian.set(mode=enums.ModifierEndianness.BIG)
await modifier.endian.set(mode=enums.ModifierEndianness.LITTLE)
resp = await modifier.endian.get()
"""32-bit modifier"""
"""Create modifier"""
await stream.packet.header.modifiers_extended.configure(number=1)
"""Clear modifiers"""
await stream.packet.header.modifiers_extended.clear()
"""Obtain modifier object"""
# Must create modifiers before obtain.
modifier_ext = stream.packet.header.modifiers_extended.obtain(idx=0)
"""Modifier Range Configuration"""
await modifier_ext.range.set(min_val=0, step=1, max_val=100)
resp = await modifier_ext.range.get()
resp.max_val
resp.min_val
resp.step
"""Modifier Position, Action, Mask"""
await modifier_ext.specification.set(position=0, mask=Hex("FFFFFFFF"), action=enums.ModifierAction.INC, repetition=1)
await modifier_ext.specification.set(position=0, mask=Hex("FFFFFFFF"), action=enums.ModifierAction.DEC, repetition=1)
await modifier_ext.specification.set(position=0, mask=Hex("FFFFFFFF"), action=enums.ModifierAction.RANDOM, repetition=1)
resp = await modifier_ext.specification.get()
resp.action
resp.mask
resp.position
resp.repetition
Layer-1 APIs
Layer-1 APIs Summary Table
Id |
Action |
API |
Description |
|---|---|---|---|
1 |
Set Link Flap Config |
port.layer1.impairment.link_flap.params.set() |
|
2 |
Get Link Flap Config |
port.layer1.impairment.link_flap.params.get() |
|
3 |
Enable Link Flap |
port.layer1.impairment.link_flap.enable.set_on() |
|
4 |
Disable Link Flap |
port.layer1.impairment.link_flap.enable.set_off() |
|
5 |
Get Link Flap Enable |
port.layer1.impairment.link_flap.enable.get() |
|
6 |
Set PMA Error Inject Config |
port.layer1.impairment.pma_error_inject.params.set() |
|
7 |
Get PMA Error Inject Config |
port.layer1.impairment.pma_error_inject.params.get() |
|
8 |
Enable PMA Error Injection |
port.layer1.impairment.pma_error_inject.enable.set_on() |
|
9 |
Disable PMA Error Injection |
port.layer1.impairment.pma_error_inject.enable.set_off() |
|
10 |
Get PMA Error Injection Enable |
port.layer1.impairment.pma_error_inject.enable.get() |
|
11 |
Set RS Fault Signaling Disabled |
port.layer1.rs_fault.signaling.set_disabled() |
|
12 |
Set RS Fault Signaling Force Local |
port.layer1.rs_fault.signaling.set_force_local() |
|
13 |
Set RS Fault Signaling Force Remote |
port.layer1.rs_fault.signaling.set_force_remote() |
|
14 |
Set RS Fault Signaling Normal |
port.layer1.rs_fault.signaling.set_normal() |
|
15 |
Get RS Fault Signaling |
port.layer1.rs_fault.signaling.get() |
|
16 |
Get RS Fault Status |
port.layer1.rs_fault.status.get() |
|
17 |
Set FEC Mode RS_FEC |
port.layer1.pcs.fec_mode.set() |
|
18 |
Set FEC Mode RS_FEC_KP |
port.layer1.pcs.fec_mode.set() |
|
19 |
Set FEC Mode RS_FEC_KR |
port.layer1.pcs.fec_mode.set() |
|
20 |
Set FEC Mode FC_FEC |
port.layer1.pcs.fec_mode.set() |
|
21 |
Set FEC Mode OFF |
port.layer1.pcs.fec_mode.set() |
|
22 |
Set FEC Mode ON |
port.layer1.pcs.fec_mode.set() |
|
23 |
Get FEC Mode |
port.layer1.pcs.fec_mode.get() |
|
24 |
Set PCS TX Config Skew |
port.layer1.pcs.lane[0].tx_config.set() |
|
25 |
Get PCS RX Status Skew |
port.layer1.pcs.lane[0].rx_status.status.get() |
|
26 |
Get Lane RX Errors |
port.layer1.pcs.lane[0].rx_status.errors.get() |
|
27 |
Get Lane RX Lock |
port.layer1.pcs.lane[0].rx_status.lock.get() |
|
28 |
Clear PCS/FEC Counters |
port.layer1.pcs.clear.set() |
|
29 |
Get FEC Symbol Error Distribution |
port.layer1.pcs.fec_symbol_status.fec_status.get() |
|
30 |
Get FEC Symbol Total Status |
port.layer1.pcs.fec_symbol_status.total_status.get() |
|
31 |
Get PCS/FEC Alarms |
port.layer1.pcs.alarms.errors.get() |
|
32 |
Set PRBS Configuration |
port.layer1.prbs_config.set() |
|
33 |
Get PRBS Configuration |
port.layer1.prbs_config.get() |
|
34 |
Enable PRBS |
port.layer1.serdes[0].prbs.control.set_on() |
|
35 |
Disable PRBS |
port.layer1.serdes[0].prbs.control.set_off() |
|
36 |
Get PRBS Status |
port.layer1.serdes[0].prbs.status.get() |
|
37 |
Set Gray Coding Configuration |
port.layer1.serdes[0].pma.graycoding.set() |
|
38 |
Get Gray Coding Configuration |
port.layer1.serdes[0].pma.graycoding.get() |
|
39 |
Set Precoding Configuration |
port.layer1.serdes[0].pma.precoding.set() |
|
40 |
Get Precoding Configuration |
port.layer1.serdes[0].pma.precoding.get() |
|
41 |
Enable P/N Swap TX |
port.layer1.serdes[0].pma.pn_swap_tx.set_on() |
|
42 |
Disable P/N Swap TX |
port.layer1.serdes[0].pma.pn_swap_tx.set_off() |
|
43 |
Get P/N Swap TX |
port.layer1.serdes[0].pma.pn_swap_tx.get() |
|
44 |
Enable P/N Swap RX |
port.layer1.serdes[0].pma.pn_swap_rx.set_on() |
|
45 |
Disable P/N Swap RX |
port.layer1.serdes[0].pma.pn_swap_rx.set_off() |
|
46 |
Get P/N Swap RX |
port.layer1.serdes[0].pma.pn_swap_rx.get() |
|
47 |
Set PCS Variant ETC |
port.layer1.pcs.variant.set() |
|
48 |
Set PCS Variant IEEE |
port.layer1.pcs.variant.set() |
|
49 |
Get PCS Variant |
port.layer1.pcs.variant.get() |
|
50 |
Start FEC Error Injection |
port.layer1.pcs.fec_error_inject.control.set_start() |
|
51 |
Stop FEC Error Injection |
port.layer1.pcs.fec_error_inject.control.set_stop() |
|
52 |
Get FEC Error Injection Status |
port.layer1.pcs.fec_error_inject.control.get() |
|
53 |
Set FEC Error Codeword Pattern |
port.layer1.pcs.fec_error_inject.cycle.set() |
|
54 |
Get FEC Error Codeword Pattern |
port.layer1.pcs.fec_error_inject.cycle.get() |
|
55 |
Set FEC Error Symbol Pattern |
port.layer1.pcs.fec_error_inject.err_symbols.set() |
|
56 |
Get FEC Error Symbol Pattern |
port.layer1.pcs.fec_error_inject.err_symbols.get() |
|
57 |
Set FEC Error Bit Mask |
port.layer1.pcs.fec_error_inject.bit_err_mask.set() |
|
58 |
Set FEC Error Bit Mask |
port.layer1.pcs.fec_error_inject.bit_err_mask.set() |
|
59 |
Set FEC Error Bit Mask |
port.layer1.pcs.fec_error_inject.bit_err_mask.set() |
|
60 |
Set FEC Error Bit Mask All |
port.layer1.pcs.fec_error_inject.bit_err_mask.set_all_bits() |
|
61 |
Set FEC Error Bit Mask None |
port.layer1.pcs.fec_error_inject.bit_err_mask.set_no_bits() |
|
62 |
Get FEC Error Bit Mask |
port.layer1.pcs.fec_error_inject.bit_err_mask.get() |
|
63 |
Set FEC Error Engine |
port.layer1.pcs.fec_error_inject.engine.set_all_engines() |
|
64 |
Get FEC Error Engine |
port.layer1.pcs.fec_error_inject.engine.get() |
|
65 |
Get FEC Error Statistics |
port.layer1.pcs.fec_error_inject.statistics.get() |
|
66 |
Clear FEC Error Statistics |
port.layer1.pcs.fec_error_inject.clear_stats.set() |
|
67 |
Set Auto-Negotiation Settings |
port.layer1.anlt.an.settings.set() |
|
68 |
Get Auto-Negotiation Settings |
port.layer1.anlt.an.settings.get() |
|
69 |
Get Auto-Negotiation Status |
port.layer1.anlt.an.status.get() |
|
70 |
Set Link Training Settings |
port.layer1.anlt.lt.settings.set() |
|
71 |
Get Link Training Settings |
port.layer1.anlt.lt.settings.get() |
|
72 |
Get Link Training Status |
port.layer1.serdes[0].lt_status.get() |
|
73 |
Set ANLT Auto-Restart |
port.layer1.anlt.set_autorestart(…) |
|
74 |
Get ANLT Auto-Restart |
port.layer1.anlt.get_autorestart() |
|
75 |
Set ANLT Strict Mode |
port.layer1.anlt.set_strict_mode(enable=False) |
|
76 |
Get ANLT Strict Mode |
port.layer1.anlt.get_strict_mode() |
|
77 |
Set Auto-Negotiation Allow Loopback |
port.layer1.anlt.an.set_allow_loopback(allow=True) |
|
78 |
Get Auto-Negotiation Allow Loopback |
port.layer1.anlt.an.get_allow_loopback() |
|
79 |
Set Auto-Negotiation Empty NP |
port.layer1.anlt.an.set_empty_np(enable=True) |
|
80 |
Get Auto-Negotiation Empty NP |
port.layer1.anlt.an.get_empty_np() |
|
81 |
Get Auto-Negotiation Supported Abilities |
port.layer1.anlt.an.abilities.get() |
|
82 |
Set Auto-Negotiation Advertise |
port.layer1.anlt.an.advertise.set() |
|
83 |
Get Auto-Negotiation Advertise |
port.layer1.anlt.an.advertise.get() |
|
84 |
Set Link Training Algorithm |
port.layer1.serdes[0].lt.set_algorithm_default() |
|
85 |
Get Link Training Algorithm |
port.layer1.serdes[0].lt.get_algorithm() |
|
86 |
Set Link Training Modulation NRZ |
port.layer1.serdes[0].lt.set_initial_modulation_nrz() |
|
87 |
Set Link Training Modulation PAM4 |
port.layer1.serdes[0].lt.set_initial_modulation_pam4() |
|
88 |
Set Link Training Modulation PAM4Precoding |
port.layer1.serdes[0].lt.set_initial_modulation_pam4precoding() |
|
89 |
Get Link Training Modulation |
port.layer1.serdes[0].lt.get_initial_modulation() |
|
90 |
Set Preset Native |
port.layer1.serdes[0].lt.preset1.native.set() |
Using the native values for Preset 1 |
91 |
Get Preset Native |
port.layer1.serdes[0].lt.preset1.native.get() |
Using the native values for Preset 1 |
92 |
Set Preset IEEE |
port.layer1.serdes[0].lt.preset1.ieee.set() |
Using the IEEE values for Preset 1 |
93 |
Get Preset IEEE |
port.layer1.serdes[0].lt.preset1.ieee.get() |
Using the IEEE values for Preset 1 |
94 |
Set Preset Level |
port.layer1.serdes[0].lt.preset1.level.set() |
Using the level values for Preset 1 |
95 |
Get Preset Level |
port.layer1.serdes[0].lt.preset1.level.get() |
Using the level values for Preset 1 |
96 |
Reset Preset to Default |
port.layer1.serdes[0].lt.preset1.reset.set() |
|
97 |
Set EQ Range Native |
port.layer1.serdes[0].lt.range.main.native.set() |
Using native values, set the response when reaches the min or max |
98 |
Get EQ Range Native |
port.layer1.serdes[0].lt.range.main.native.get() |
Using native values, set the response when reaches the min or max |
99 |
Set EQ Range IEEE |
port.layer1.serdes[0].lt.range.main.ieee.set() |
Using ieee values, set the response when reaches the min or max |
100 |
Get EQ Range IEEE |
port.layer1.serdes[0].lt.range.main.ieee.get() |
Using ieee values, set the response when reaches the min or max |
101 |
Set EQ Range Auto Native |
port.layer1.serdes[0].lt.range.main.native.set() |
When in Auto mode, the min and max values use the default values ignored |
102 |
Set EQ Range Auto IEEE |
port.layer1.serdes[0].lt.range.main.ieee.set() |
When in Auto mode, the min and max values use the default values ignored |
103 |
Set Link Training Config |
port.layer1.anlt.lt_config.set() |
Using IEEE standard OOS preset |
104 |
Set Link Training Config |
port.layer1.anlt.lt_config.set() |
Using IEEE standard OOS preset |
105 |
Control ANLT AN Only |
port.layer1.anlt.ctrl.enable_an_only() |
Only enable Auto-Negotiation |
106 |
Control ANLT AN & LT Auto |
port.layer1.anlt.ctrl.enable_an_lt_auto() |
Enable Auto-Negotiation and Link Training (auto mode) |
107 |
Control ANLT AN & LT Interactive |
port.layer1.anlt.ctrl.enable_an_lt_interactive() |
Enable Auto-Negotiation and Link Training (manual mode) |
108 |
Control ANLT LT Auto Only |
port.layer1.anlt.ctrl.enable_lt_auto_only() |
Only enable Link Training (auto mode), skipping Auto-Negotiation |
109 |
Control ANLT LT Interactive Only |
port.layer1.anlt.ctrl.enable_lt_interactive_only() |
Only enable Link Training (manual mode), skipping Auto-Negotiation |
110 |
Control ANLT Disable |
port.layer1.anlt.ctrl.disable_anlt() |
Disable both Auto-Negotiation and Link Training |
111 |
Send Link Training Modulation |
port.layer1.serdes[0].lt.send_cmd_modulation() |
Send PAM4 Request |
112 |
Send Link Training Modulation |
port.layer1.serdes[0].lt.send_cmd_modulation() |
Send PAM4Precoding Request |
113 |
Send Link Training Preset |
port.layer1.serdes[0].lt.send_cmd_preset() |
Send Preset Request |
114 |
Send Link Training Inc |
port.layer1.serdes[0].lt.send_cmd_inc() |
Send EQ Coefficient Inc Request |
115 |
Send Link Training Dec |
port.layer1.serdes[0].lt.send_cmd_dec() |
Send EQ Coefficient Dec Request |
116 |
Send Link Training No Equalization |
port.layer1.serdes[0].lt.send_cmd_no_equalization() |
Send No EQ |
117 |
Send Link Training Trained |
port.layer1.serdes[0].lt.send_cmd_trained() |
Send Trained Notification |
118 |
Get Link Training Command Result |
port.layer1.serdes[0].lt.get_cmd_result_flag() |
Check response status after sending any command |
119 |
Get Auto-Negotiation Results |
port.layer1.anlt.an.results.get() |
Get Auto-Negotiation results |
120 |
Get Auto-Negotiation Statistics |
port.layer1.anlt.an.statistics.get() |
Get Auto-Negotiation statistics |
121 |
Get Link Training Results |
port.layer1.serdes[0].lt.results.get() |
Get Link Training results, per-serdes |
122 |
Get Link Training Statistics |
port.layer1.serdes[0].lt.statistics.get() |
Get Link Training statistics, per-serdes |
123 |
Set Signal Integrity Sample Control |
port.layer1.serdes[0].siv.control.set() |
|
124 |
Get Signal Integrity Sample Data |
port.layer1.serdes[0].siv.data.get() |
|
125 |
Set TX Equalizer |
port.layer1.serdes[0].medium.tx_equalizer.set() |
Tx Equalizer Settings |
126 |
Set RX Equalizer |
port.layer1.serdes[0].medium.rx_equalizer.set() |
Rx Equalizer Settings |
127 |
Set TX Equalizer |
port.layer1.serdes[0].medium.tx_equalizer.set() |
Tx Equalizer Settings |
128 |
Set RX Equalizer |
port.layer1.serdes[0].medium.rx_equalizer.set() |
Rx Equalizer Settings |
129 |
Set TX Native |
port.layer1.serdes[0].medium.tx.native.set() |
Tx Equalizer Settings |
130 |
Set RX AGC |
port.layer1.serdes[0].medium.rx.config.agc.set() |
Rx Equalizer Settings, either in Auto, Manual, or Freeze |
131 |
Set RX CDR |
port.layer1.serdes[0].medium.rx.config.cdr.set() |
Rx Equalizer Settings, either in Auto, Manual, or Freeze |
132 |
Set RX CTLE High |
port.layer1.serdes[0].medium.rx.config.ctle_high.set() |
Rx Equalizer Settings, either in Auto, Manual, or Freeze |
133 |
Set RX CTLE Low |
port.layer1.serdes[0].medium.rx.config.ctle_low.set() |
Rx Equalizer Settings, either in Auto, Manual, or Freeze |
134 |
Set RX DFE |
port.layer1.serdes[0].medium.rx.config.dfe.set() |
Rx Equalizer Settings, either in Auto, Manual, or Freeze |
135 |
Set RX OC |
port.layer1.serdes[0].medium.rx.config.oc.set() |
Rx Equalizer Settings, either in Auto, Manual, or Freeze |
136 |
Set RX Post-FFE 1 |
port.layer1.serdes[0].medium.rx.config.post_ffe_1.set() |
Post-FFE 1-23 |
137 |
Set RX Pre-FFE 1 |
port.layer1.serdes[0].medium.rx.config.pre_ffe_1.set() |
Pre-FFE 1-8 |
138 |
Set TX IEEE |
port.layer1.serdes[0].medium.tx.ieee.set() |
Tx Equalizer Settings |
139 |
Transceiver Read/Write |
port.transceiver.access_rw(…).set() |
Provides read and write access to the |
140 |
Transceiver Read/Write |
port.transceiver.access_rw(…).get() |
Provides read and write access to the |
141 |
Transceiver Sequential Read/Write |
port.transceiver.access_rw_seq(…).set() |
I2C sequential access to a transceiver’s register. |
142 |
Transceiver Sequential Read/Write |
port.transceiver.access_rw_seq(…).get() |
I2C sequential access to a transceiver’s register. |
143 |
Transceiver Sequential Read/Write Banked |
port.transceiver.access_rw_seq_bank(…).set() |
I2C sequential access with bank selection. |
144 |
Transceiver Sequential Read/Write Banked |
port.transceiver.access_rw_seq_bank(…).get() |
I2C sequential access with bank selection. |
145 |
Transceiver MII Access |
port.transceiver.access_mii(…).set() |
Provides access to the register interface supported. |
146 |
Transceiver MII Access |
port.transceiver.access_mii(…).get() |
Provides access to the register interface supported. |
147 |
Configure I2C |
port.transceiver.i2c_config.set() |
Configures the I2C interface parameters |
148 |
Get TX Laser Power |
port.transceiver.laser_power.tx_laser_power.get() |
Reading of the optical power level of the Tx/Rx signal |
149 |
Get RX Laser Power |
port.transceiver.laser_power.rx_laser_power.get() |
Reading of the optical power level of the Tx/Rx signal |
150 |
Get CDB Instances Supported |
port.transceiver.cmis.cdb_instances_supported.get() |
Return the supported CDB instances from the CMIS transceiver |
Layer-1 API Examples
# [Layer-1 Advanced Features]
# [Z100 Loki, Z400 Thor, Z800 Freya, Z1600 Edun Specific APIs]
if isinstance(port, ports.Z100LokiPort) or isinstance(port, ports.Z400ThorPort) or isinstance(port, ports.Z800FreyaPort) or isinstance(port, ports.Z1600EdunPort):
"""LINK FLAP"""
"""Link Flap - Configuration"""
await port.layer1.impairment.link_flap.params.set(duration=10, period=20, repetition=0)
resp_obj = await port.layer1.impairment.link_flap.params.get()
"""Link Flap - Control"""
await port.layer1.impairment.link_flap.enable.set_on()
await port.layer1.impairment.link_flap.enable.set_off()
resp_obj = await port.layer1.impairment.link_flap.enable.get()
"""PMA ERROR INJECTION"""
"""PMA Error Insertion - Configuration"""
await port.layer1.impairment.pma_error_inject.params.set(duration=10, period=20, repetition=0, coeff=1, exp=-6)
resp_obj = await port.layer1.impairment.pma_error_inject.params.get()
"""PMA Error Insertion - Control"""
await port.layer1.impairment.pma_error_inject.enable.set_on()
await port.layer1.impairment.pma_error_inject.enable.set_off()
resp_obj = await port.layer1.impairment.pma_error_inject.enable.get()
"""RS FAULT"""
"""RS Fault - Signaling"""
await port.layer1.rs_fault.signaling.set_disabled()
await port.layer1.rs_fault.signaling.set_force_local()
await port.layer1.rs_fault.signaling.set_force_remote()
await port.layer1.rs_fault.signaling.set_normal()
resp_obj = await port.layer1.rs_fault.signaling.get()
"""RS Fault - Status"""
resp_obj = await port.layer1.rs_fault.status.get()
"""PCS/FEC"""
"""FEC Mode"""
await port.layer1.pcs.fec_mode.set(mode=enums.FECMode.RS_FEC)
await port.layer1.pcs.fec_mode.set(mode=enums.FECMode.RS_FEC_KP)
await port.layer1.pcs.fec_mode.set(mode=enums.FECMode.RS_FEC_KR)
await port.layer1.pcs.fec_mode.set(mode=enums.FECMode.FC_FEC)
await port.layer1.pcs.fec_mode.set(mode=enums.FECMode.OFF)
await port.layer1.pcs.fec_mode.set(mode=enums.FECMode.ON)
resp_obj = await port.layer1.pcs.fec_mode.get()
"""PCS TX Configuration - Skew"""
await port.layer1.pcs.lane[0].tx_config.set(virt_lane_index=0, skew=0)
"""PCS RX Status - Skew"""
resp_obj = await port.layer1.pcs.lane[0].rx_status.status.get()
"""PCS Lane BER and Error Counters"""
resp_obj = await port.layer1.pcs.lane[0].rx_status.errors.get()
"""PCS Header Lock & Align Lock"""
resp_obj = await port.layer1.pcs.lane[0].rx_status.lock.get()
"""Clear Counters"""
await port.layer1.pcs.clear.set()
"""FEC Symbol Error Distribution Status"""
resp_obj = await port.layer1.pcs.fec_symbol_status.fec_status.get()
"""FEC Symbol Total Status"""
# Total Codewords
# Corrected Codewords
# Uncorrectable Codewords
# Corrected Symbols
# Pre-FEC BER
# Post-FEC BER
resp_obj = await port.layer1.pcs.fec_symbol_status.total_status.get()
"""Error Counters"""
# Total Alarms, LOS Errors, Alignment Errors, BIP Errors, FEC Errors, Header Errors, Higher Layer Errors, PCS Errors, Valid Mask
resp_obj = await port.layer1.pcs.alarms.errors.get()
"""PRBS PATTERN GENERATION"""
"""PRBS Configuration"""
await port.layer1.prbs_config.set(prbs_inserted_type=enums.PRBSInsertedType.PHY_LINE, polynomial=enums.PRBSPolynomial.PRBS31, invert=enums.PRBSInvertState.NON_INVERTED, statistics_mode=enums.PRBSStatisticsMode.ACCUMULATIVE)
resp_obj = await port.layer1.prbs_config.get()
"""PRBS Control"""
await port.layer1.serdes[0].prbs.control.set_on()
await port.layer1.serdes[0].prbs.control.set_off()
resp_obj = await port.layer1.serdes[0].prbs.control.get()
"""PRBS Bit Error and Lock Status"""
resp_obj = await port.layer1.serdes[0].prbs.status.get()
# [Z800 Freya, Z1600 Edun Specific APIs]
if isinstance(port, ports.Z800FreyaPort) or isinstance(port, ports.Z1600EdunPort):
"""GRAY CODING AND PRECODING"""
# [Z800 Freya, Z1600 Edun Specific APIs]
"""Gray Coding - Configuration"""
await port.layer1.serdes[0].pma.graycoding.set(rx_mode=enums.GrayCodingMode.ON, rx_endianness=enums.Endianness.NORMAL, tx_mode=enums.GrayCodingMode.ON, tx_endianness=enums.Endianness.NORMAL)
resp = await port.layer1.serdes[0].pma.graycoding.get()
"""Precoding - Configuration"""
await port.layer1.serdes[0].pma.precoding.set(rx_mode=enums.PreCodingMode.ON, rx_endianness=enums.Endianness.NORMAL,tx_mode=enums.PreCodingMode.ON, tx_endianness=enums.Endianness.NORMAL)
resp = await port.layer1.serdes[0].pma.precoding.get()
"""P/N POLARITY SWAP"""
# [Z800 Freya, Z1600 Edun Specific APIs]
"""P/N Polarity - Tx/Rx Swap"""
await port.layer1.serdes[0].pma.pn_swap_tx.set_on()
await port.layer1.serdes[0].pma.pn_swap_tx.set_off()
resp = await port.layer1.serdes[0].pma.pn_swap_tx.get()
await port.layer1.serdes[0].pma.pn_swap_rx.set_on()
await port.layer1.serdes[0].pma.pn_swap_rx.set_off()
resp = await port.layer1.serdes[0].pma.pn_swap_rx.get()
"""PCS VARIANT"""
# [Z800 Freya, Z1600 Edun Specific APIs]
await port.layer1.pcs.variant.set(variant=enums.FreyaPCSVariant.ETC)
await port.layer1.pcs.variant.set(variant=enums.FreyaPCSVariant.IEEE)
resp_obj = await port.layer1.pcs.variant.get()
"""FEC ERROR INSERTION"""
# [Z800 Freya, Z1600 Edun Specific APIs]
"""FEC Error Insertion - Control"""
await port.layer1.pcs.fec_error_inject.control.set_start()
await port.layer1.pcs.fec_error_inject.control.set_stop()
resp = await port.layer1.pcs.fec_error_inject.control.get()
"""FEC Error Insertion - Codeword Error Pattern"""
await port.layer1.pcs.fec_error_inject.cycle.set(loop=0, cycle_len=8, error_len=4)
resp = await port.layer1.pcs.fec_error_inject.cycle.get()
"""FEC Error Insertion - Symbol Error Pattern"""
await port.layer1.pcs.fec_error_inject.err_symbols.set(error_sym_indices=[543, 542, 541, 50, 44, 76, 88])
resp = await port.layer1.pcs.fec_error_inject.err_symbols.get()
"""FEC Error Insertion - Bit Error Mask"""
await port.layer1.pcs.fec_error_inject.bit_err_mask.set(mode=enums.FecCodewordBitErrorMaskMode.STATIC, bitmask=Hex("000F"))
await port.layer1.pcs.fec_error_inject.bit_err_mask.set(mode=enums.FecCodewordBitErrorMaskMode.ROTATE_HIGH, bitmask=Hex("000F"))
await port.layer1.pcs.fec_error_inject.bit_err_mask.set(mode=enums.FecCodewordBitErrorMaskMode.INC, bitmask=Hex("000F"))
await port.layer1.pcs.fec_error_inject.bit_err_mask.set_all_bits()
await port.layer1.pcs.fec_error_inject.bit_err_mask.set_no_bits()
resp = await port.layer1.pcs.fec_error_inject.bit_err_mask.get()
"""FEC Error Insertion - FEC Engine Control"""
await port.layer1.pcs.fec_error_inject.engine.set_all_engines()
resp = await port.layer1.pcs.fec_error_inject.engine.get()
"""FEC Error Insertion - Statistics"""
resp = await port.layer1.pcs.fec_error_inject.statistics.get()
"""FEC Error Insertion - Clear Statistics"""
await port.layer1.pcs.fec_error_inject.clear_stats.set()
"""AUTO-NEGOTIATION AND LINK TRAINING"""
"""Basic Auto-Negotiation and Link Training Operations"""
# [For Z400 Thor and Z1600 Edun]
if isinstance(port, ports.Z1600EdunPort) or isinstance(port, ports.Z400ThorPort):
"""Auto-Negotiation Settings"""
ta_conf = enums.AutoNegTecAbility.IEEE_1_6TBASE_CR8_KR8 | enums.AutoNegTecAbility.IEEE_800GBASE_CR4_KR4
ta_conf = Hex(format(ta_conf, 'X').zfill(16))
fec_conf = enums.AutoNegFECOption.RS528 | enums.AutoNegFECOption.RS272
fec_conf = Hex(format(fec_conf, 'X').zfill(2))
pause_conf = enums.PauseMode.SYM_PAUSE | enums.PauseMode.ASYM_PAUSE
pause_conf = Hex(format(pause_conf, 'X').zfill(2))
await port.layer1.anlt.an.settings.set(
mode=enums.AutoNegMode.ANEG_ON,
tec_ability=ta_conf,
fec_capable=fec_conf,
fec_requested=fec_conf,
pause_mode=pause_conf
)
resp = await port.layer1.anlt.an.settings.get()
"""Auto-Negotiation Status"""
await port.layer1.anlt.an.status.get()
"""Link Training Settings"""
await port.layer1.anlt.lt.settings.set(
mode=enums.LinkTrainingMode.DISABLED,
pam4_frame_size=enums.PAM4FrameSize.P4K_FRAME,
nrz_pam4_init_cond=enums.LinkTrainingInitCondition.NO_INIT,
nrz_preset=enums.NRZPreset.NRZ_WITH_PRESET,
timeout_mode=enums.TimeoutMode.DEFAULT)
resp = await port.layer1.anlt.lt.settings.get()
"""Link Training Status"""
resp = await port.layer1.serdes[0].lt_status.get()
"""Advanced Auto-Negotiation and Link Training Operations"""
# [For Z800 Freya]
if isinstance(port, ports.Z800FreyaPort):
"""ANLT Auto-Restart Settings"""
await port.layer1.anlt.set_autorestart(restart_link_down=True, restart_lt_failure=True)
resp = await port.layer1.anlt.get_autorestart()
"""ANLT Strict Mode Settings"""
await port.layer1.anlt.set_strict_mode(enable=False)
resp = await port.layer1.anlt.get_strict_mode()
"""Auto-Negotiation Allow-in-Loopback Settings"""
await port.layer1.anlt.an.set_allow_loopback(allow=True)
resp = await port.layer1.anlt.an.get_allow_loopback()
"""Auto-Negotiation Send Empty Next Page Settings"""
await port.layer1.anlt.an.set_empty_np(enable=True)
resp = await port.layer1.anlt.an.get_empty_np()
"""Read Auto-Negotiation supported abilities per-port"""
resp= await port.layer1.anlt.an.abilities.get()
"""Set Auto-Negotiation advertised abilities per-port"""
ta_conf = enums.AutoNegTecAbility.IEEE_1_6TBASE_CR8_KR8 | enums.AutoNegTecAbility.IEEE_800GBASE_CR4_KR4
ta_conf = Hex(format(ta_conf, 'X').zfill(16))
fec_conf = enums.AutoNegFECOption.RS528 | enums.AutoNegFECOption.RS272
fec_conf = Hex(format(fec_conf, 'X').zfill(2))
pause_conf = enums.PauseMode.SYM_PAUSE | enums.PauseMode.ASYM_PAUSE
pause_conf = Hex(format(pause_conf, 'X').zfill(2))
await port.layer1.anlt.an.advertise.set(
tech_abilities=ta_conf,
fec_abilities=fec_conf,
pause_mode=pause_conf
)
resp = await port.layer1.anlt.an.advertise.get()
"""Link Training Per-Serdes Algorithm Selection"""
await port.layer1.serdes[0].lt.set_algorithm_default()
resp = await port.layer1.serdes[0].lt.get_algorithm()
"""Link Training Per-Serdes Initial Modulation Selection"""
await port.layer1.serdes[0].lt.set_initial_modulation_nrz()
await port.layer1.serdes[0].lt.set_initial_modulation_pam4()
await port.layer1.serdes[0].lt.set_initial_modulation_pam4precoding()
resp = await port.layer1.serdes[0].lt.get_initial_modulation()
"""Link Training Per-Serdes Preset Custom Configuration"""
# Using the native values for Preset 1
await port.layer1.serdes[0].lt.preset1.native.set(
response=enums.FreyaPresetResponse.ACCEPT,
pre3=0,
pre2=0,
pre=0,
main=0,
post=0
)
resp = await port.layer1.serdes[0].lt.preset1.native.get()
# Using the IEEE values for Preset 1
await port.layer1.serdes[0].lt.preset1.ieee.set(
response=enums.FreyaPresetResponse.ACCEPT,
pre3=-200,
pre2=100,
pre=-100,
main=507,
post=100,
)
resp = await port.layer1.serdes[0].lt.preset1.ieee.get()
# Using the level values for Preset 1
await port.layer1.serdes[0].lt.preset1.level.set(
response=enums.FreyaPresetResponse.ACCEPT,
pre3=6,
pre2=3,
pre=2,
main=64,
post=3,
)
resp = await port.layer1.serdes[0].lt.preset1.level.get()
"""Reset Link Training Per-Serdes Preset to Default"""
await port.layer1.serdes[0].lt.preset1.reset.set()
"""Link Training Per-Serdes Eq Range Configuration"""
# Using native values, set the response to COEFF_AT_LIMIT when the coeff reaches the min or max
await port.layer1.serdes[0].lt.range.main.native.set(response=enums.FreyaLinkTrainingRangeResponse.COEFF_AT_LIMIT, min=0, max=50)
resp = await port.layer1.serdes[0].lt.range.main.native.get()
# Using ieee values, set the response to COEFF_AT_LIMIT when the coeff reaches the min or max
await port.layer1.serdes[0].lt.range.main.ieee.set(response=enums.FreyaLinkTrainingRangeResponse.COEFF_AT_LIMIT, min=0, max=50)
resp = await port.layer1.serdes[0].lt.range.main.ieee.get()
"""Reset Link Training Per-Serdes Eq Range to Auto"""
# When in Auto mode, the min and max values use the default values to the full range supported by the serdes, thus min and max are ignored
await port.layer1.serdes[0].lt.range.main.native.set(response=enums.FreyaLinkTrainingRangeResponse.AUTO, min=0, max=0)
await port.layer1.serdes[0].lt.range.main.ieee.set(response=enums.FreyaLinkTrainingRangeResponse.AUTO, min=0, max=0)
"""Link Training Configurationl, per-port"""
# Using IEEE standard OOS preset
# Enable Default Timeout Mode for Link Training Auto mode
await port.layer1.anlt.lt_config.set(
oos_preset=enums.FreyaOutOfSyncPreset.IEEE,
timeout_mode=enums.TimeoutMode.DEFAULT
)
# Using IEEE standard OOS preset
# Disable Timeout for Link Training Manual mode
await port.layer1.anlt.lt_config.set(
oos_preset=enums.FreyaOutOfSyncPreset.IEEE,
timeout_mode=enums.TimeoutMode.DISABLED
)
"""Control ANLT with various modes, per-port"""
# Only enable Auto-Negotiation
await port.layer1.anlt.ctrl.enable_an_only()
# Enable Auto-Negotiation and Link Training (auto mode)
# When Link Training is set to auto mode, the serdes will automatically select the best algorithm and initial modulation based on the negotiated capabilities
await port.layer1.anlt.ctrl.enable_an_lt_auto()
# Enable Auto-Negotiation and Link Training (manual mode)
# When Link Training is set to manual mode, the serdes will use the user configured algorithm and initial modulation
await port.layer1.anlt.ctrl.enable_an_lt_interactive()
# Only enable Link Training (auto mode), skipping Auto-Negotiation
await port.layer1.anlt.ctrl.enable_lt_auto_only()
# Only enable Link Training (manual mode), skipping Auto-Negotiation
await port.layer1.anlt.ctrl.enable_lt_interactive_only()
# Disable both Auto-Negotiation and Link Training
await port.layer1.anlt.ctrl.disable_anlt()
"""For Link Training in manual mode, you can send different LT commands to the remote partner, per-serdes"""
# Send PAM4 Request
await port.layer1.serdes[0].lt.send_cmd_modulation(modulation=enums.LinkTrainEncoding.PAM4)
# Send PAM4Precoding Request
await port.layer1.serdes[0].lt.send_cmd_modulation(modulation=enums.LinkTrainEncoding.PAM4_WITH_PRECODING)
# Send Preset Request
await port.layer1.serdes[0].lt.send_cmd_preset(preset=enums.LinkTrainPresets.PRESET_1)
# Send EQ Coefficient Inc Request
await port.layer1.serdes[0].lt.send_cmd_inc(coeff=enums.LinkTrainCoeffs.MAIN)
# Send EQ Coefficient Dec Request
await port.layer1.serdes[0].lt.send_cmd_dec(coeff=enums.LinkTrainCoeffs.MAIN)
# Send No EQ
await port.layer1.serdes[0].lt.send_cmd_no_equalization()
# Send Trained Notification
await port.layer1.serdes[0].lt.send_cmd_trained()
# Check response status after sending any command
cmd, result, flag = await port.layer1.serdes[0].lt.get_cmd_result_flag()
"""ANLT Statistics, Result and Status"""
# Get Auto-Negotiation results
resp = await port.layer1.anlt.an.results.get()
# Get Auto-Negotiation statistics
resp = await port.layer1.anlt.an.statistics.get()
# Get Link Training results, per-serdes
resp = await port.layer1.serdes[0].lt.results.get()
# Get Link Training statistics, per-serdes
resp = await port.layer1.serdes[0].lt.statistics.get()
"""SIGNAL INTEGRITY SAMPLE READING"""
# Only for Freya and Edun Modules
if isinstance(port, ports.Z800FreyaPort) or isinstance(port, ports.Z1600EdunPort):
"""Signal Integrity Sample Read - Control"""
await port.layer1.serdes[0].siv.control.set(opcode=enums.Layer1Opcode.START_SCAN)
"""Signal Integrity Sample Read - Data"""
resp = await port.layer1.serdes[0].siv.data.get()
"""HOST TX/RX EQUALIZATION CONFIGURATION"""
# For Loki Modules
if isinstance(port, ports.Z100LokiPort):
# Tx Equalizer Settings
await port.layer1.serdes[0].medium.tx_equalizer.set(tap_values=[0, 60, 0])
# Rx Equalizer Settings
await port.layer1.serdes[0].medium.rx_equalizer.set(auto=1, ctle=5, reserved=0)
# For Thor Modules
if isinstance(port, ports.Z400ThorPort):
# Tx Equalizer Settings
await port.layer1.serdes[0].medium.tx_equalizer.set(tap_values=[0, 60, 0, 0, 0])
# Rx Equalizer Settings
await port.layer1.serdes[0].medium.rx_equalizer.set(auto=1, ctle=5, reserved=0)
# For Freya Modules
if isinstance(port, ports.Z800FreyaPort):
# Tx Equalizer Settings
await port.layer1.serdes[0].medium.tx.native.set(tap_values=[0, 60, 0, 0, 0, 0])
# Rx Equalizer Settings, either in Auto, Manual, or Freeze
await port.layer1.serdes[0].medium.rx.config.agc.set(mode=enums.RxEqExtCapMode.MANUAL, value=5)
await port.layer1.serdes[0].medium.rx.config.cdr.set(mode=enums.RxEqExtCapMode.MANUAL, value=5)
await port.layer1.serdes[0].medium.rx.config.ctle_high.set(mode=enums.RxEqExtCapMode.MANUAL, value=5)
await port.layer1.serdes[0].medium.rx.config.ctle_low.set(mode=enums.RxEqExtCapMode.MANUAL, value=5)
await port.layer1.serdes[0].medium.rx.config.dfe.set(mode=enums.RxEqExtCapMode.MANUAL, value=5)
await port.layer1.serdes[0].medium.rx.config.oc.set(mode=enums.RxEqExtCapMode.MANUAL, value=5)
# Post-FFE 1-23
await port.layer1.serdes[0].medium.rx.config.post_ffe_1.set(mode=enums.RxEqExtCapMode.MANUAL, value=5)
# Pre-FFE 1-8
await port.layer1.serdes[0].medium.rx.config.pre_ffe_1.set(mode=enums.RxEqExtCapMode.MANUAL, value=5)
# For Edun Modules
if isinstance(port, ports.Z1600EdunPort):
# Tx Equalizer Settings
await port.layer1.serdes[0].medium.tx.ieee.set(tap_values=[0, 60, 0, 0, 0, 0, 0])
# Rx Equalizer Settings
Transceiver API Examples
# [Transceiver Management]
"""TRANSCEIVER MANAGEMENT"""
"""Read / Write"""
# Provides read and write access to the
# register interface supported by the transceiver.
await port.transceiver.access_rw(page_address=0, register_address=0).set(value=Hex("FF"))
resp = await port.transceiver.access_rw(page_address=0, register_address=0).get()
"""Sequential Read & Write"""
# I2C sequential access to a transceiver's register.
# When invoked, the ``<byte_count>`` number of bytes will be
# read or written in one I2C transaction, in which the
# ``<value>`` is read or written with only a single register
# address setup. A subsequent invocation will perform a second
# I2C transaction in the same manner.
await port.transceiver.access_rw_seq(page_address=0, register_address=0, byte_count=4).set(value=Hex("00FF00FF"))
resp = await port.transceiver.access_rw_seq(page_address=0, register_address=0, byte_count=4).get()
"""Sequential Read & Write (Banked)"""
# I2C sequential access to a transceiver's register
# with bank selection.
await port.transceiver.access_rw_seq_bank(bank_address=1, page_address=0x9F, register_address=200, byte_count=1).set(value=Hex("00"))
resp = await port.transceiver.access_rw_seq_bank(bank_address=1, page_address=0x9F, register_address=200, byte_count=1).get()
"""MII Access"""
# Provides access to the register interface supported
# by the media-independent interface (MII) transceiver.
await port.transceiver.access_mii(register_address=0).set(value=Hex("00"))
resp = await port.transceiver.access_mii(register_address=0).get()
"""I2C Configuration"""
# Configures the I2C interface parameters
await port.transceiver.i2c_config.set(frequency=100)
"""Tx/Rx Laser Power"""
# Reading of the optical power level of the Tx/Rx signal
resp = await port.transceiver.laser_power.tx_laser_power.get()
resp.nanowatts
resp = await port.transceiver.laser_power.rx_laser_power.get()
resp.nanowatts
"""Get CDB Instances Supported"""
# Return the supported CDB instances from the CMIS transceiver
await port.transceiver.cmis.cdb_instances_supported.get()
"""CMIS CDB"""
# See hlfuncs.cmis.cdb module for CMIS CDB APIs
Ultra Ethernet APIs
Ultra Ethernet APIs Summary Table
Id |
Action |
API |
Description |
|---|---|---|---|
1 |
LLR Mode Set On |
await port.uec.llr.mode.set_on() |
Manually set the port’s LLR mode to on |
2 |
LLR Mode Set Off |
await port.uec.llr.mode.set_off() |
Manually set the port’s LLR mode to off |
3 |
LLR Link Negotiation Config Set |
await port.uec.linkneg.link_options.config.set(lldp_agent_index=0, llr_wanted_local=enums.UecLinkOptionLlr.BI_WANTED) |
Set the port’s LLR mode based on the link negotiation result |
4 |
Stream Create |
stream = await port.streams.create() |
Create a stream on the port |
5 |
Stream LLR Desire Set Eligible |
await stream.uec.llr.desire.set_llr_eligible() |
Set the stream’s LLR desire to LLR-eligible |
6 |
CtlOS Tx Stats Get |
resp = await port.uec.ctlos.stats.tx.get() |
Get LLR CtlOS Tx statistics |
7 |
CtlOS Tx LLR Init Count |
resp.llr_init_cnt |
Number of LLR Init frames sent (Tx) |
8 |
CtlOS Tx LLR Init Echo Count |
resp.llr_init_echo_cnt |
Number of LLR Init Echo frames sent (Tx) |
9 |
CtlOS Tx LLR Ack Count |
resp.llr_ack_cnt |
Number of LLR ACK frames sent (Tx) |
10 |
CtlOS Tx LLR Nack Count |
resp.llr_nack_cnt |
Number of LLR NACK frames sent (Tx) |
11 |
CtlOS Tx LLR Init Seq |
resp.llr_init_seq |
LLR Init sequence number (Tx) |
12 |
CtlOS Tx LLR Init Data |
resp.llr_init_data |
LLR Init data payload (Tx) |
13 |
CtlOS Tx LLR Init Echo Seq |
resp.llr_init_echo_seq |
LLR Init Echo sequence number (Tx) |
14 |
CtlOS Tx LLR Init Echo Data |
resp.llr_init_echo_data |
LLR Init Echo data payload (Tx) |
15 |
CtlOS Rx Stats Get |
resp = await port.uec.ctlos.stats.rx.get() |
Get LLR CtlOS Rx statistics |
16 |
CtlOS Rx LLR Init Count |
resp.llr_init_cnt |
Number of LLR Init frames received (Rx) |
17 |
CtlOS Rx LLR Init Echo Count |
resp.llr_init_echo_cnt |
Number of LLR Init Echo frames received (Rx) |
18 |
CtlOS Rx LLR Ack Count |
resp.llr_ack_cnt |
Number of LLR ACK frames received (Rx) |
19 |
CtlOS Rx LLR Nack Count |
resp.llr_nack_cnt |
Number of LLR NACK frames received (Rx) |
20 |
CtlOS Rx LLR Init Seq |
resp.llr_init_seq |
LLR Init sequence number (Rx) |
21 |
CtlOS Rx LLR Init Data |
resp.llr_init_data |
LLR Init data payload (Rx) |
22 |
CtlOS Rx LLR Init Echo Seq |
resp.llr_init_echo_seq |
LLR Init Echo sequence number (Rx) |
23 |
CtlOS Rx LLR Init Echo Data |
resp.llr_init_echo_data |
LLR Init Echo data payload (Rx) |
24 |
CtlOS Rx LLR Init Echo Mismatch |
resp.llr_init_echo_mismatch |
Number of LLR Init Echo mismatches detected (Rx) |
25 |
LLR Tx Layer-2 Stats Get |
resp = await port.uec.llr.stats.tx.get() |
Get LLR Tx layer-2 statistics |
26 |
LLR Tx Eligible Good FCS |
resp.tx_llr_eligible_good_fcs |
Tx LLR-eligible frames with good FCS |
27 |
LLR Tx Eligible Poisoned FCS |
resp.tx_llr_eligible_poisoned_fcs |
Tx LLR-eligible frames with poisoned FCS |
28 |
LLR Tx Eligible Bad FCS |
resp.tx_llr_eligible_bad_fcs |
Tx LLR-eligible frames with bad FCS |
29 |
LLR Tx Eligible Discard |
resp.tx_llr_eligible_discard |
Tx LLR-eligible frames discarded |
30 |
LLR Tx Ineligible Good FCS |
resp.tx_llr_ineligible_good_fcs |
Tx LLR-ineligible frames with good FCS |
31 |
LLR Tx Ineligible Bad FCS |
resp.tx_llr_ineligible_bad_fcs |
Tx LLR-ineligible frames with bad FCS |
32 |
LLR Tx Ineligible Total |
resp.tx_llr_ineligible_total |
Total Tx LLR-ineligible frames |
33 |
LLR Tx Replayed |
resp.tx_replayed |
Number of Tx replayed frames |
34 |
LLR Rx Layer-2 Stats Get |
resp = await port.uec.llr.stats.rx.get() |
Get LLR Rx layer-2 statistics |
35 |
LLR Rx Eligible Good FCS |
resp.rx_llr_eligible_good_fcs |
Rx LLR-eligible frames with good FCS |
36 |
LLR Rx Eligible Poisoned FCS |
resp.rx_llr_eligible_poisoned_fcs |
Rx LLR-eligible frames with poisoned FCS |
37 |
LLR Rx Eligible Bad FCS |
resp.rx_llr_eligible_bad_fcs |
Rx LLR-eligible frames with bad FCS |
38 |
LLR Rx Eligible Total |
resp.rx_llr_eligible_total |
Total Rx LLR-eligible frames |
39 |
LLR Rx Eligible Good FCS Exp Seq |
resp.rx_llr_eligible_good_fcs_exp_seq |
Rx LLR-eligible frames with good FCS and expected sequence number |
40 |
LLR Rx Eligible Poisoned FCS Exp Seq |
resp.rx_llr_eligible_poisoned_fcs_exp_seq |
Rx LLR-eligible frames with poisoned FCS and expected sequence number |
41 |
LLR Rx Eligible Bad FCS Exp Seq |
resp.rx_llr_eligible_bad_fcs_exp_seq |
Rx LLR-eligible frames with bad FCS and expected sequence number |
42 |
LLR Rx Eligible Total Exp Seq |
resp.rx_llr_eligible_total_exp_seq |
Total Rx LLR-eligible frames with expected sequence number |
43 |
LLR Rx Eligible Missing Seq |
resp.rx_llr_eligible_missing_seq |
Rx LLR-eligible frames with missing sequence number |
44 |
LLR Rx Eligible Duplicate Seq |
resp.rx_llr_eligible_duplicate_seq |
Rx LLR-eligible frames with duplicate sequence number |
45 |
LLR Rx Ineligible Good FCS |
resp.rx_llr_ineligible_good_fcs |
Rx LLR-ineligible frames with good FCS |
46 |
LLR Rx Ineligible Bad FCS |
resp.rx_llr_ineligible_bad_fcs |
Rx LLR-ineligible frames with bad FCS |
47 |
LLR Rx Ineligible Total |
resp.rx_llr_ineligible_total |
Total Rx LLR-ineligible frames |
Ultra Ethernet API Examples
# [ports]
"""Get Port Object"""
port = module.ports.obtain(0)
"""LLR mode - manual configuration"""
# Manually set the port's LLR mode
await port.uec.llr.mode.set_on()
await port.uec.llr.mode.set_off()
"""LLR mode - automatic configuration using Link Negotiation"""
# Auto configure the port's LLR mode based on the link partner's capability using Link Negotiation
await port.uec.linkneg.link_options.config.set(lldp_agent_index = 0, llr_wanted_local=enums.UecLinkOptionLlr.BI_WANTED)
"""Create stream on port and set LLR desire"""
# Create a stream on the port and set the LLR desire of the stream to LLR-eligible.
stream = await port.streams.create()
await stream.uec.llr.desire.set_llr_eligible()
"""Get LLR statistics"""
# Get LLR CtlOS Tx statistics
resp = await port.uec.ctlos.stats.tx.get()
resp.llr_init_cnt
resp.llr_init_echo_cnt
resp.llr_ack_cnt
resp.llr_nack_cnt
resp.llr_init_seq
resp.llr_init_data
resp.llr_init_echo_seq
resp.llr_init_echo_data
# Get LLR CtlOS Rx statistics
resp = await port.uec.ctlos.stats.rx.get()
resp.llr_init_cnt
resp.llr_init_echo_cnt
resp.llr_ack_cnt
resp.llr_nack_cnt
resp.llr_init_seq
resp.llr_init_data
resp.llr_init_echo_seq
resp.llr_init_echo_data
resp.llr_init_echo_mismatch
# Get LLR Tx layer-2 statistics
resp = await port.uec.llr.stats.tx.get()
resp.tx_llr_eligible_good_fcs
resp.tx_llr_eligible_poisoned_fcs
resp.tx_llr_eligible_bad_fcs
resp.tx_llr_eligible_discard
resp.tx_llr_ineligible_good_fcs
resp.tx_llr_ineligible_bad_fcs
resp.tx_llr_ineligible_total
resp.tx_replayed
# Get LLR Rx layer-2 statistics
resp = await port.uec.llr.stats.rx.get()
resp.rx_llr_eligible_good_fcs
resp.rx_llr_eligible_poisoned_fcs
resp.rx_llr_eligible_bad_fcs
resp.rx_llr_eligible_total
resp.rx_llr_eligible_good_fcs_exp_seq
resp.rx_llr_eligible_poisoned_fcs_exp_seq
resp.rx_llr_eligible_bad_fcs_exp_seq
resp.rx_llr_eligible_total_exp_seq
resp.rx_llr_eligible_missing_seq
resp.rx_llr_eligible_duplicate_seq
resp.rx_llr_ineligible_good_fcs
resp.rx_llr_ineligible_bad_fcs
resp.rx_llr_ineligible_total
MACsec APIs
MACsec APIs Summary Table
Id |
Action |
API |
Description |
|---|---|---|---|
1 |
Create |
port.macsec.txscs.create() |
Create a TX Secure Channel |
2 |
Set |
txsc_obj.config.description.set() |
Tx SC Description |
3 |
Set |
txsc_obj.config.sci.set() |
Tx SC SCI |
4 |
Set |
txsc_obj.config.sci_mode.set() |
Tx SC Mode |
5 |
Set |
txsc_obj.config.confidentiality_offset.set() |
Tx SC Confidentiality Offset |
6 |
Set |
txsc_obj.config.cipher_suite.set() |
Tx SC Cipher Suite |
7 |
Set |
txsc_obj.config.starting_pn.set() |
Tx SC Starting PN |
8 |
Set |
txsc_obj.config.rekey_mode.set() |
Tx SC ReKey Mode |
9 |
Set |
txsc_obj.config.encryption_mode.set() |
Tx SC Encryption Mode |
10 |
Set |
txsc_obj.config.xpn_ssci.set() |
Tx SC XPN SSCI |
11 |
Set |
txsc_obj.config.xpn_salt.set() |
Tx SC XPN Salt |
12 |
Set |
txsc_obj.config.next_pn.set() |
Tx SC Next PN |
13 |
Get |
txsc_obj.config.next_pn.get() |
Tx SC Next PN |
14 |
Set |
txsc_obj.config.next_an.set() |
Tx SC Next AN |
15 |
Get |
txsc_obj.config.next_an.get() |
Tx SC Next AN |
16 |
Create |
port.macsec.rxscs.create() |
Create a RX Secure Channel |
17 |
Set |
rxsc_obj.config.description.set() |
Rx SC Description |
18 |
Set |
rxsc_obj.config.sci.set() |
Rx SC SCI |
19 |
Set |
rxsc_obj.config.confidentiality_offset.set() |
Rx SC Confidentiality Offset |
20 |
Set |
rxsc_obj.config.cipher_suite.set() |
Rx SC Cipher Suite |
21 |
Set |
rxsc_obj.config.lowest_pn.set() |
Rx SC Lowest PN |
22 |
Set |
rxsc_obj.config.tpld_id.set() |
Rx SC TPLD ID |
23 |
Set |
rxsc_obj.config.xpn_ssci.set() |
Rx SC XPN SSCI |
24 |
Set |
rxsc_obj.config.xpn_salt.set() |
Rx SC XPN Salt |
25 |
Get |
rxsc_obj.config.an.get() |
Rx SC AN |
26 |
Get |
rxsc_obj.config.next_pn.get() |
RX SC Next PN |
27 |
Get |
rxsc_obj.config.pn.get() |
RX SC PN |
28 |
Set |
port.macsec.decode.set_on() |
Enable MACsec decode on the port |
29 |
Set |
port.macsec.decode.set_off() |
Disable MACsec decode on the port |
30 |
Set |
stream.macsec.assign.set() |
Assing Tx SC to the stream by Tx SC index |
31 |
Set |
stream.macsec.enable.set_on() |
Enable MACsec encode on the stream |
32 |
Get |
port.macsec.statistics.tx.total.get() |
Collect MACsec port statistics TX |
33 |
Get |
port.macsec.statistics.rx.total.get() |
Collect MACsec port statistics RX |
34 |
Set |
port.macsec.statistics.tx.clear.set() |
Clear MACsec port statistics TX |
35 |
Set |
port.macsec.statistics.rx.clear.set() |
Clear MACsec port statistics RX |
36 |
Get |
txsc_obj.stats.get() |
Collect MACsec TXSC statistics for the port |
37 |
Get |
rxsc_obj.stats.get() |
Collect MACsec RXSC statistics for the port |
MACsec API Examples
# [ports]
"""Get Port Object"""
port = module.ports.obtain(0)
"""Create a stream"""
stream = await port.streams.create()
# [Loki-4P Specific]
if isinstance(port, ports.PLoki100G5S4P_a):
"""MACSEC CONFIGURATION"""
"""MACsec - Tx SCs"""
# Create a TX Secure Channel
txsc_obj = await port.macsec.txscs.create()
# Tx SC Description
await txsc_obj.config.description.set(description="This is a TX SC comment")
# Tx SC SCI
await txsc_obj.config.sci.set(sci=Hex("AABBCCDDEEFF0011"))
# Tx SC Mode
await txsc_obj.config.sci_mode.set(mode=enums.MACSecSCIMode.WITH_SCI)
await txsc_obj.config.sci_mode.set(mode=enums.MACSecSCIMode.END_STATION)
# Tx SC Confidentiality Offset
await txsc_obj.config.confidentiality_offset.set(offset=0)
# Tx SC Cipher Suite
await txsc_obj.config.cipher_suite.set(cipher_suite=enums.MACSecCipherSuite.GCM_AES_128)
# Tx SC Starting PN
await txsc_obj.config.starting_pn.set(starting_pn=1, mode=enums.MACSecPNMode.RESET)
# Tx SC ReKey Mode
await txsc_obj.config.rekey_mode.set(mode=enums.MACSecRekeyMode.PN_EXHAUSTION, value=0)
# Tx SC Encryption Mode
await txsc_obj.config.encryption_mode.set(mode=enums.MACSecEncryptionMode.ENCRYPT_INTEGRITY)
# Tx SC XPN SSCI
await txsc_obj.config.xpn_ssci.set(xpn_ssci=Hex("000000000001"))
# Tx SC XPN Salt
await txsc_obj.config.xpn_salt.set(xpn_salt=Hex("AABBCCDDEEFFAABBCCDDEEFF"))
# Tx SC Next PN
await txsc_obj.config.next_pn.set(next_pn=1)
resp = await txsc_obj.config.next_pn.get()
# Tx SC Next AN
await txsc_obj.config.next_an.set(next_an=1)
resp = await txsc_obj.config.next_an.get()
"""MACsec - Rx SCs"""
# Create a RX Secure Channel
rxsc_obj = await port.macsec.rxscs.create()
# Rx SC Description
await rxsc_obj.config.description.set(description="This is a RX SC comment")
# Rx SC SCI
await rxsc_obj.config.sci.set(sci=Hex("AABBCCDDEEFF0011"))
# Rx SC Confidentiality Offset
await rxsc_obj.config.confidentiality_offset.set(offset=0)
# Rx SC Cipher Suite
await rxsc_obj.config.cipher_suite.set(cipher_suite=enums.MACSecCipherSuite.GCM_AES_128)
# Rx SC Lowest PN
await rxsc_obj.config.lowest_pn.set(lowest_pn=1)
# Rx SC TPLD ID
await rxsc_obj.config.tpld_id.set(tpld_id=0)
# Rx SC XPN SSCI
await rxsc_obj.config.xpn_ssci.set(xpn_ssci=Hex("000000000001"))
# Rx SC XPN Salt
await rxsc_obj.config.xpn_salt.set(xpn_salt=Hex("AABBCCDDEEFFAABBCCDDEEFF"))
# Rx SC AN
resp = await rxsc_obj.config.an.get()
# RX SC Next PN
resp = await rxsc_obj.config.next_pn.get()
# RX SC PN
resp = await rxsc_obj.config.pn.get()
# Enable MACsec decode on the port
await port.macsec.decode.set_on()
await port.macsec.decode.set_off()
# Assing Tx SC to the stream by Tx SC index
await stream.macsec.assign.set(tx_sc_index=txsc_obj.kind.index_id)
# Enable MACsec encode on the stream
await stream.macsec.enable.set_on()
# Collect MACsec port statistics
await port.macsec.statistics.tx.total.get()
await port.macsec.statistics.rx.total.get()
# Clear MACsec port statistics
await port.macsec.statistics.tx.clear.set()
await port.macsec.statistics.rx.clear.set()
# Collect MACsec TXSC/RXSC statistics for the port
resp = await txsc_obj.stats.get()
resp = await rxsc_obj.stats.get()
LLDP APIs
LLDP APIs Summary Table
Id |
Action |
API |
Description |
|---|---|---|---|
1 |
Create LLDP Agent Object |
port.lldp.agents.create() |
Create an LLDP agent on the port. |
2 |
Obtain LLDP Agent Object |
port.lldp.agents.obtain() |
Obtain an existing LLDP agent object by index. |
3 |
Obtain Multiple LLDP Agent Objects |
port.lldp.agents.obtain_multiple() |
Obtain multiple existing LLDP agent objects by indices. |
4 |
Sync Existing LLDP Agents |
port.lldp.agents.server_sync() |
Sync existing LLDP agents from the port to the script. |
5 |
Get LLDP Agent Index |
lldp_agent.idx |
Get the LLDP agent index. |
6 |
Get LLDP Agent Module Index |
lldp_agent.kind.module_id |
Get the module index of the LLDP agent. |
7 |
Get LLDP Agent Port Index |
lldp_agent.kind.port_id |
Get the port index of the LLDP agent. |
8 |
Remove LLDP Agent by Index |
port.lldp.agents.remove() |
Remove an LLDP agent from the port by index. |
9 |
Remove LLDP Agent by Object |
lldp_agent.delete() |
Remove an LLDP agent by deleting the LLDP agent object. |
10 |
Set LLDP Agent Time Parameters |
lldp_agent.config.set() |
Configure LLDP agent time parameters. |
11 |
Get LLDP Agent Time Parameters |
lldp_agent.config.get() |
Get LLDP agent time parameters. |
12 |
Set LLDP Agent Frame Header |
lldp_agent.header.set() |
Configure the LLDP agent Ethernet frame header. |
13 |
Get LLDP Agent Frame Header |
lldp_agent.header.get() |
Get the LLDP agent frame header configuration. |
14 |
Set LLDP Agent LLDPDU |
lldp_agent.lldpdu.set() |
Configure the LLDPDU payload for the LLDP agent. |
15 |
Get LLDP Agent LLDPDU |
lldp_agent.lldpdu.get() |
Get the LLDPDU payload of the LLDP agent. |
16 |
Set LLDP Agent Opmode Disable |
lldp_agent.opmode.set_disable() |
Set LLDP agent operation mode to disable. |
17 |
Set LLDP Agent Opmode TX Only |
lldp_agent.opmode.set_tx_only() |
Set LLDP agent operation mode to TX only. |
18 |
Set LLDP Agent Opmode RX Only |
lldp_agent.opmode.set_rx_only() |
Set LLDP agent operation mode to RX only. |
19 |
Set LLDP Agent Opmode TX/RX |
lldp_agent.opmode.set_tx_rx() |
Set LLDP agent operation mode to TX/RX. |
20 |
Get LLDP Agent Opmode |
lldp_agent.opmode.get() |
Get the current LLDP agent operation mode. |
21 |
Get LLDP Statistics |
port.lldp.statistics.get() |
Get LLDP statistics from the port. |
22 |
Get LLDP Neighbor Information |
port.lldp.neighbors.get() |
Get LLDP neighbor information from the port. |
23 |
Clear LLDP Counters (All) |
port.lldp.clear.clear_all() |
Clear all LLDP counters. |
24 |
Clear LLDP Counters (Neighbor) |
port.lldp.clear.clear_neighbor() |
Clear LLDP neighbor counters. |
25 |
Clear LLDP Counters (Stats) |
port.lldp.clear.clear_stats() |
Clear LLDP statistics counters. |
26 |
Clear LLDP Counters (None) |
port.lldp.clear.clear_none() |
Set LLDP clear action to none. |
LLDP API Examples
# [LLDP]
if isinstance(port, ports.Z800FreyaPort):
"""Create LLDP Agent Object"""
# Create a LLDP agent representing the LLDP agent on the port.
lldp_agent = await port.lldp.agents.create()
"""Obtain One or Multiple LLDP Agent Objects"""
# Obtain already existing LLDP agent object(s) using agent indices.
lldp_agent = port.lldp.agents.obtain(0)
lldp_agents = port.lldp.agents.obtain_multiple(*[0,1,2])
"""Sync Existing LLDP Agents from Port to Script"""
await port.lldp.agents.server_sync()
"""Get LLDP Agent index"""
lldp_agent.idx
"""Get LLDP Agent's module and port index"""
lldp_agent.kind.module_id
lldp_agent.kind.port_id
"""Remove LLDP Agent by Index"""
# Remove a LLDP agent from the port by its index.
await port.lldp.agents.remove(position_idx=0)
"""Remove LLDP Agent by Removing the Object"""
# Remove a LLDP agent from the port by removing the LLDP agent object.
await lldp_agent.delete()
"""Configure a LLDP Agent time parameters"""
# Configure the LLDP agent time parameters, including reinit delay, tx interval, tx hold multiplier, and tx delay.
await lldp_agent.config.set(
reinit_delay=2,
tx_interval=5,
tx_hold_multiplier=4,
tx_delay=2)
resp = await lldp_agent.config.get()
"""Configure a LLDP Agent frame header"""
# Configure the LLDP agent's frame header
# Here we use the hlfuncs.headers.Ethernet class to create an Ethernet header, and then set the destination MAC address, source MAC address, and ethertype for the LLDP frame header.
eth = hlfuncs.headers.Ethernet()
eth.dst_mac = "0180.c200.000e"
eth.src_mac = "00:11:22:33:44:55"
eth.ethertype = 0x88cc
await lldp_agent.header.set(header=Hex(str(eth)))
resp = await lldp_agent.header.get()
"""Configure a LLDP Agent LLDPDU"""
# Configure a LLDP agent's LLDPDU, which is the payload of the LLDP frame.
# HLAPI doesn't have a specific class for LLDPDU, so we can directly set the LLDPDU using a hexadecimal string.
await lldp_agent.lldpdu.set(Hex("020704000130f9ada0040405312f31060200780000"))
resp = await lldp_agent.lldpdu.get()
"""Control the LLDP Agent's operation mode"""
# Control the LLDP agent's operation mode
await lldp_agent.opmode.set_disable()
await lldp_agent.opmode.set_tx_only()
await lldp_agent.opmode.set_rx_only()
await lldp_agent.opmode.set_tx_rx()
resp = await lldp_agent.opmode.get()
"""Get LLDP statistics from the port"""
resp = await port.lldp.statistics.get()
"""Get LLDP neighbor information"""
resp = await port.lldp.neighbors.get()
"""Clear LLDP counters"""
await port.lldp.clear.clear_all()
await port.lldp.clear.clear_neighbor()
await port.lldp.clear.clear_stats()
await port.lldp.clear.clear_none()
Network Impairment APIs
Network Impairment APIs Summary Table
Id |
Action |
API |
Description |
|---|---|---|---|
1 |
Set Link Flap Parameters |
port.pcs_pma.link_flap.params.set() |
Set link flap duration/period/repetition |
2 |
Enable Link Flap |
port.pcs_pma.link_flap.enable.set_on() |
Enable link flap impairment |
3 |
Disable Link Flap |
port.pcs_pma.link_flap.enable.set_off() |
Disable link flap impairment |
4 |
Set PMA Error Inject Params |
port.pcs_pma.pma_error_inject.params.set() |
Set PMA error injection parameters |
5 |
Enable PMA Error Injection |
port.pcs_pma.pma_error_inject.enable.set_on() |
Enable PMA error injection |
6 |
Disable PMA Error Injection |
port.pcs_pma.pma_error_inject.enable.set_off() |
Disable PMA error injection |
7 |
Disable Port Impairment |
port.emulate.set_off() |
Disable impairment on the port |
8 |
Enable Port Impairment |
port.emulate.set_on() |
Enable impairment on the port |
9 |
Get Port Impairment Status |
port.emulate.get() |
Get port impairment enable/disable status |
10 |
Set TPLD Mode Normal |
port.emulation.tpld_mode.set() |
Set TPLD mode to normal |
11 |
Set TPLD Mode Micro |
port.emulation.tpld_mode.set() |
Set TPLD mode to micro |
12 |
Get TPLD Mode |
port.emulation.tpld_mode.get() |
Get the current TPLD mode |
13 |
Set Flow Description |
flow.comment.set() |
Set a text description for the flow |
14 |
Get Flow Description |
flow.comment.get() |
Get the flow description |
15 |
Initialize Shadow Filter |
flow.shadow_filter.initiating.set() |
Initialize the shadow copy of the filter |
16 |
Configure Shadow Filter Basic Mode |
flow.shadow_filter.use_basic_mode() |
Configure shadow filter to basic mode |
17 |
Get Filter Mode |
flow.shadow_filter.get_mode() |
Query the filter mode (basic or extended) |
18 |
Set Ethernet Filter Settings |
filter.ethernet.settings.set() |
Set ethernet subfilter use and action |
19 |
Set Ethernet Source Address |
filter.ethernet.src_address.set() |
Set ethernet source address filter |
20 |
Set Ethernet Destination Address |
filter.ethernet.dest_address.set() |
Set ethernet destination address filter |
21 |
Disable Layer 2+ Subfilter |
filter.l2plus_use.set() |
Disable Layer 2+ subfilter |
22 |
Enable VLAN1 Filter |
filter.l2plus_use.set() |
Enable Layer 2+ subfilter with VLAN1 |
23 |
Set VLAN Filter Settings |
filter.vlan.settings.set() |
Set VLAN subfilter use and action |
24 |
Set VLAN Inner Tag |
filter.vlan.inner.tag.set() |
Set VLAN inner tag filter |
25 |
Set VLAN Inner PCP |
filter.vlan.inner.pcp.set() |
Set VLAN inner PCP filter |
26 |
Enable VLAN2 Filter |
filter.l2plus_use.set() |
Enable Layer 2+ subfilter with VLAN2 |
27 |
Set VLAN Outer Tag |
filter.vlan.outer.tag.set() |
Set VLAN outer tag filter |
28 |
Set VLAN Outer PCP |
filter.vlan.outer.pcp.set() |
Set VLAN outer PCP filter |
29 |
Enable MPLS Filter |
filter.l2plus_use.set() |
Enable Layer 2+ subfilter with MPLS |
30 |
Set MPLS Filter Settings |
filter.mpls.settings.set() |
Set MPLS subfilter use and action |
31 |
Set MPLS Label |
filter.mpls.label.set() |
Set MPLS label filter |
32 |
Set MPLS ToC |
filter.mpls.toc.set() |
Set MPLS traffic class filter |
33 |
Disable Layer 3 Subfilter |
filter.l3_use.set() |
Disable Layer 3 subfilter |
34 |
Enable IPv4 Filter |
filter.l3_use.set() |
Enable Layer 3 subfilter with IPv4 |
35 |
Set IPv4 Filter Settings |
filter.ip.v4.settings.set() |
Set IPv4 subfilter use and action |
36 |
Set IPv4 Source Address |
filter.ip.v4.src_address.set() |
Set IPv4 source address filter |
37 |
Set IPv4 Destination Address |
filter.ip.v4.dest_address.set() |
Set IPv4 destination address filter |
38 |
Set IPv4 DSCP |
filter.ip.v4.dscp.set() |
Set IPv4 DSCP filter |
39 |
Enable IPv6 Filter |
filter.l3_use.set() |
Enable Layer 3 subfilter with IPv6 |
40 |
Set IPv6 Filter Settings |
filter.ip.v6.settings.set() |
Set IPv6 subfilter use and action |
41 |
Set IPv6 Source Address |
filter.ip.v6.src_address.set() |
Set IPv6 source address filter |
42 |
Set IPv6 Destination Address |
filter.ip.v6.dest_address.set() |
Set IPv6 destination address filter |
43 |
Set IPv6 Traffic Class |
filter.ip.v6.traffic_class.set() |
Set IPv6 traffic class filter |
44 |
Set TCP Filter Settings |
filter.tcp.settings.set() |
Set TCP subfilter use and action |
45 |
Set TCP Source Port |
filter.tcp.src_port.set() |
Set TCP source port filter |
46 |
Set TCP Destination Port |
filter.tcp.dest_port.set() |
Set TCP destination port filter |
47 |
Set UDP Filter Settings |
filter.udp.settings.set() |
Set UDP subfilter use and action |
48 |
Set UDP Source Port |
filter.udp.src_port.set() |
Set UDP source port filter |
49 |
Set UDP Destination Port |
filter.udp.dest_port.set() |
Set UDP destination port filter |
50 |
Set TPLD Filter Settings |
filter.tpld.settings.set() |
Set test payload filter use and action |
51 |
Enable TPLD Filter 0 |
filter.tpld.test_payload_filters_config[0].set() |
Enable TPLD 0 filter |
52 |
Disable TPLD Filter 0 |
filter.tpld.test_payload_filters_config[0].set() |
Disable TPLD 0 filter |
53 |
Enable TPLD Filter 1 |
filter.tpld.test_payload_filters_config[1].set() |
Enable TPLD 1 filter |
54 |
Disable TPLD Filter 1 |
filter.tpld.test_payload_filters_config[1].set() |
Disable TPLD 1 filter |
55 |
Enable TPLD Filter 2 |
filter.tpld.test_payload_filters_config[2].set() |
Enable TPLD 2 filter |
56 |
Disable TPLD Filter 2 |
filter.tpld.test_payload_filters_config[2].set() |
Disable TPLD 2 filter |
57 |
Enable TPLD Filter 3 |
filter.tpld.test_payload_filters_config[3].set() |
Enable TPLD 3 filter |
58 |
Disable TPLD Filter 3 |
filter.tpld.test_payload_filters_config[3].set() |
Disable TPLD 3 filter |
59 |
Enable TPLD Filter 4 |
filter.tpld.test_payload_filters_config[4].set() |
Enable TPLD 4 filter |
60 |
Disable TPLD Filter 4 |
filter.tpld.test_payload_filters_config[4].set() |
Disable TPLD 4 filter |
61 |
Enable TPLD Filter 5 |
filter.tpld.test_payload_filters_config[5].set() |
Enable TPLD 5 filter |
62 |
Disable TPLD Filter 5 |
filter.tpld.test_payload_filters_config[5].set() |
Disable TPLD 5 filter |
63 |
Enable TPLD Filter 6 |
filter.tpld.test_payload_filters_config[6].set() |
Enable TPLD 6 filter |
64 |
Disable TPLD Filter 6 |
filter.tpld.test_payload_filters_config[6].set() |
Disable TPLD 6 filter |
65 |
Enable TPLD Filter 7 |
filter.tpld.test_payload_filters_config[7].set() |
Enable TPLD 7 filter |
66 |
Disable TPLD Filter 7 |
filter.tpld.test_payload_filters_config[7].set() |
Disable TPLD 7 filter |
67 |
Enable TPLD Filter 8 |
filter.tpld.test_payload_filters_config[8].set() |
Enable TPLD 8 filter |
68 |
Disable TPLD Filter 8 |
filter.tpld.test_payload_filters_config[8].set() |
Disable TPLD 8 filter |
69 |
Enable TPLD Filter 9 |
filter.tpld.test_payload_filters_config[9].set() |
Enable TPLD 9 filter |
70 |
Disable TPLD Filter 9 |
filter.tpld.test_payload_filters_config[9].set() |
Disable TPLD 9 filter |
71 |
Enable TPLD Filter 10 |
filter.tpld.test_payload_filters_config[10].set() |
Enable TPLD 10 filter |
72 |
Disable TPLD Filter 10 |
filter.tpld.test_payload_filters_config[10].set() |
Disable TPLD 10 filter |
73 |
Enable TPLD Filter 11 |
filter.tpld.test_payload_filters_config[11].set() |
Enable TPLD 11 filter |
74 |
Disable TPLD Filter 11 |
filter.tpld.test_payload_filters_config[11].set() |
Disable TPLD 11 filter |
75 |
Enable TPLD Filter 12 |
filter.tpld.test_payload_filters_config[12].set() |
Enable TPLD 12 filter |
76 |
Disable TPLD Filter 12 |
filter.tpld.test_payload_filters_config[12].set() |
Disable TPLD 12 filter |
77 |
Enable TPLD Filter 13 |
filter.tpld.test_payload_filters_config[13].set() |
Enable TPLD 13 filter |
78 |
Disable TPLD Filter 13 |
filter.tpld.test_payload_filters_config[13].set() |
Disable TPLD 13 filter |
79 |
Enable TPLD Filter 14 |
filter.tpld.test_payload_filters_config[14].set() |
Enable TPLD 14 filter |
80 |
Disable TPLD Filter 14 |
filter.tpld.test_payload_filters_config[14].set() |
Disable TPLD 14 filter |
81 |
Enable TPLD Filter 15 |
filter.tpld.test_payload_filters_config[15].set() |
Enable TPLD 15 filter |
82 |
Disable TPLD Filter 15 |
filter.tpld.test_payload_filters_config[15].set() |
Disable TPLD 15 filter |
83 |
Set Any Filter Settings |
filter.any.settings.set() |
Set generic any subfilter use and action |
84 |
Set Any Filter Config |
filter.any.config.set() |
Set generic any filter position/value/mask |
85 |
Disable Shadow Filter |
flow.shadow_filter.enable.set_off() |
Disable the shadow filter |
86 |
Enable Shadow Filter |
flow.shadow_filter.enable.set_on() |
Enable the shadow filter |
87 |
Apply Shadow Filter |
flow.shadow_filter.apply.set() |
Apply shadow filter configuration |
88 |
Configure Shadow Filter Extended Mode |
flow.shadow_filter.use_extended_mode() |
Configure shadow filter to extended mode |
89 |
Set Protocol Segments |
filter.use_segments() |
Set protocol segments for extended mode filter |
90 |
Get Protocol Segments |
filter.get_protocol_segments() |
Get protocol segments for extended mode filter |
91 |
Set Protocol Segment Value |
protocol_segments[].value.set() |
Set protocol segment value for extended filter |
92 |
Set Protocol Segment Mask |
protocol_segments[].mask.set() |
Set protocol segment mask for extended filter |
93 |
Set Drop Fixed Burst |
flow.impairment_distribution.drop_type_config.fixed_burst.set() |
Set fixed burst for drop impairment |
94 |
Set Drop Schedule Repeat |
flow.impairment_distribution.drop_type_config.schedule.set() |
Set repeat schedule for drop impairment |
95 |
Set Drop Schedule One Shot |
flow.impairment_distribution.drop_type_config.schedule.set() |
Set one shot schedule for drop impairment |
96 |
Set Drop Random Burst |
flow.impairment_distribution.drop_type_config.random_burst.set() |
Set random burst for drop impairment |
97 |
Set Drop Fixed Rate |
flow.impairment_distribution.drop_type_config.fixed_rate.set() |
Set fixed rate for drop impairment |
98 |
Set Drop Bit Error Rate |
flow.impairment_distribution.drop_type_config.bit_error_rate.set() |
Set bit error rate for drop impairment |
99 |
Set Drop Random Rate |
flow.impairment_distribution.drop_type_config.random_rate.set() |
Set random rate for drop impairment |
100 |
Set Drop Gilbert Elliot |
flow.impairment_distribution.drop_type_config.ge.set() |
Set Gilbert Elliot distribution for drop impairment |
101 |
Set Drop Uniform |
flow.impairment_distribution.drop_type_config.uniform.set() |
Set uniform distribution for drop impairment |
102 |
Set Drop Gaussian |
flow.impairment_distribution.drop_type_config.gaussian.set() |
Set Gaussian distribution for drop impairment |
103 |
Set Drop Poisson |
flow.impairment_distribution.drop_type_config.poisson.set() |
Set Poisson distribution for drop impairment |
104 |
Set Drop Gamma |
flow.impairment_distribution.drop_type_config.gamma.set() |
Set Gamma distribution for drop impairment |
105 |
Assign Custom Distribution |
port.custom_distributions.assign() |
Assign custom distribution to port |
106 |
Set Custom Distribution Comment |
port.custom_distributions[].comment.set() |
Set description for custom distribution |
107 |
Set Custom Distribution Definition |
port.custom_distributions[].definition.set() |
Set custom distribution definition |
108 |
Set Drop Custom Distribution |
flow.impairment_distribution.drop_type_config.custom.set() |
Set custom distribution for drop impairment |
109 |
Enable Drop Impairment |
flow.impairment_distribution.drop_type_config.enable.set_on() |
Enable drop impairment on flow |
110 |
Disable Drop Impairment |
flow.impairment_distribution.drop_type_config.enable.set_off() |
Disable drop impairment on flow |
111 |
Set Misorder Fixed Burst |
flow.impairment_distribution.misorder_type_config.fixed_burst.set() |
Set fixed burst for misorder impairment |
112 |
Set Misorder Schedule Repeat |
flow.impairment_distribution.misorder_type_config.schedule.set() |
Set repeat schedule for misorder impairment |
113 |
Set Misorder Schedule One Shot |
flow.impairment_distribution.misorder_type_config.schedule.set() |
Set one shot schedule for misorder impairment |
114 |
Set Misorder Fixed Rate |
flow.impairment_distribution.misorder_type_config.fixed_rate.set() |
Set fixed rate for misorder impairment |
115 |
Set Misorder Configuration |
flow.misordering.set() |
Set misorder depth configuration |
116 |
Enable Misorder Impairment |
flow.impairment_distribution.misorder_type_config.enable.set_on() |
Enable misorder impairment on flow |
117 |
Disable Misorder Impairment |
flow.impairment_distribution.misorder_type_config.enable.set_off() |
Disable misorder impairment on flow |
118 |
Set Latency Constant Delay |
flow.impairment_distribution.latency_jitter_type_config.constant_delay.set() |
Set constant delay for latency/jitter impairment |
119 |
Set Latency Accumulate and Burst |
flow.impairment_distribution.latency_jitter_type_config.accumulate_and_burst.set() |
Set accumulate and burst for latency/jitter impairment |
120 |
Set Latency Schedule |
flow.impairment_distribution.latency_jitter_type_config.schedule.set() |
Set schedule for latency/jitter impairment |
121 |
Set Latency Step |
flow.impairment_distribution.latency_jitter_type_config.step.set() |
Set step distribution for latency/jitter impairment |
122 |
Disable Corruption |
flow.impairment_distribution.corruption_type_config.off.set() |
Disable corruption impairment |
123 |
Set Latency Uniform |
flow.impairment_distribution.latency_jitter_type_config.uniform.set() |
Set uniform distribution for latency/jitter impairment |
124 |
Set Latency Gaussian |
flow.impairment_distribution.latency_jitter_type_config.gaussian.set() |
Set Gaussian distribution for latency/jitter impairment |
125 |
Get Latency Range |
flow.latency_range.get() |
Get minimum and maximum latency values |
126 |
Set Latency Poisson |
flow.impairment_distribution.latency_jitter_type_config.poisson.set() |
Set Poisson distribution for latency/jitter impairment |
127 |
Set Latency Gamma |
flow.impairment_distribution.latency_jitter_type_config.gamma.set() |
Set Gamma distribution for latency/jitter impairment |
128 |
Set Latency Custom Distribution |
flow.impairment_distribution.latency_jitter_type_config.custom.set() |
Set custom distribution for latency/jitter impairment |
129 |
Enable Latency/Jitter Impairment |
flow.impairment_distribution.latency_jitter_type_config.enable.set_on() |
Enable latency/jitter impairment on flow |
130 |
Disable Latency/Jitter Impairment |
flow.impairment_distribution.latency_jitter_type_config.enable.set_off() |
Disable latency/jitter impairment on flow |
131 |
Set Duplication Fixed Burst |
flow.impairment_distribution.duplication_type_config.fixed_burst.set() |
Set fixed burst for duplication impairment |
132 |
Set Duplication Schedule |
flow.impairment_distribution.duplication_type_config.schedule.set() |
Set schedule for duplication impairment |
133 |
Set Duplication Random Burst |
flow.impairment_distribution.duplication_type_config.random_burst.set() |
Set random burst for duplication impairment |
134 |
Set Duplication Fixed Rate |
flow.impairment_distribution.duplication_type_config.fixed_rate.set() |
Set fixed rate for duplication impairment |
135 |
Set Duplication Bit Error Rate |
flow.impairment_distribution.duplication_type_config.bit_error_rate.set() |
Set bit error rate for duplication impairment |
136 |
Set Duplication Random Rate |
flow.impairment_distribution.duplication_type_config.random_rate.set() |
Set random rate for duplication impairment |
137 |
Set Duplication Gilbert Elliot |
flow.impairment_distribution.duplication_type_config.ge.set() |
Set Gilbert Elliot distribution for duplication impairment |
138 |
Set Duplication Uniform |
flow.impairment_distribution.duplication_type_config.uniform.set() |
Set uniform distribution for duplication impairment |
139 |
Set Duplication Gaussian |
flow.impairment_distribution.duplication_type_config.gaussian.set() |
Set Gaussian distribution for duplication impairment |
140 |
Set Duplication Poisson |
flow.impairment_distribution.duplication_type_config.poisson.set() |
Set Poisson distribution for duplication impairment |
141 |
Set Duplication Gamma |
flow.impairment_distribution.duplication_type_config.gamma.set() |
Set Gamma distribution for duplication impairment |
142 |
Set Duplication Custom Distribution |
flow.impairment_distribution.duplication_type_config.custom.set() |
Set custom distribution for duplication impairment |
143 |
Enable Duplication Impairment |
flow.impairment_distribution.duplication_type_config.enable.set_on() |
Enable duplication impairment on flow |
144 |
Disable Duplication Impairment |
flow.impairment_distribution.duplication_type_config.enable.set_off() |
Disable duplication impairment on flow |
145 |
Set Corruption Fixed Burst |
flow.impairment_distribution.corruption_type_config.fixed_burst.set() |
Set fixed burst for corruption impairment |
146 |
Set Corruption Random Burst |
flow.impairment_distribution.corruption_type_config.random_burst.set() |
Set random burst for corruption impairment |
147 |
Set Corruption Fixed Rate |
flow.impairment_distribution.corruption_type_config.fixed_rate.set() |
Set fixed rate for corruption impairment |
148 |
Set Corruption Bit Error Rate |
flow.impairment_distribution.corruption_type_config.bit_error_rate.set() |
Set bit error rate for corruption impairment |
149 |
Set Corruption Random Rate |
flow.impairment_distribution.corruption_type_config.random_rate.set() |
Set random rate for corruption impairment |
150 |
Set Corruption Gilbert Elliot |
flow.impairment_distribution.corruption_type_config.ge.set() |
Set Gilbert Elliot distribution for corruption impairment |
151 |
Set Corruption Uniform |
flow.impairment_distribution.corruption_type_config.uniform.set() |
Set uniform distribution for corruption impairment |
152 |
Set Corruption Gaussian |
flow.impairment_distribution.corruption_type_config.gaussian.set() |
Set Gaussian distribution for corruption impairment |
153 |
Set Corruption Poisson |
flow.impairment_distribution.corruption_type_config.poisson.set() |
Set Poisson distribution for corruption impairment |
154 |
Set Corruption Gamma |
flow.impairment_distribution.corruption_type_config.gamma.set() |
Set Gamma distribution for corruption impairment |
155 |
Set Corruption Custom Distribution |
flow.impairment_distribution.corruption_type_config.custom.set() |
Set custom distribution for corruption impairment |
156 |
Set Corruption Type OFF |
flow.corruption.set() |
Set corruption type to off |
157 |
Set Corruption Type ETH |
flow.corruption.set() |
Set corruption type to ethernet |
158 |
Set Corruption Type IP |
flow.corruption.set() |
Set corruption type to IP |
159 |
Set Corruption Type TCP |
flow.corruption.set() |
Set corruption type to TCP |
160 |
Set Corruption Type UDP |
flow.corruption.set() |
Set corruption type to UDP |
161 |
Set Corruption Type BER |
flow.corruption.set() |
Set corruption type to bit error rate |
162 |
Enable Corruption Impairment |
flow.impairment_distribution.corruption_type_config.enable.set_on() |
Enable corruption impairment on flow |
163 |
Disable Corruption Impairment |
flow.impairment_distribution.corruption_type_config.enable.set_off() |
Disable corruption impairment on flow |
164 |
Get One Shot Status |
flow.impairment_distribution.corruption_type_config.one_shot_status.get() |
Get one shot execution status |
165 |
Set Policer L1 |
flow.bandwidth_control.policer.set() |
Set policer L1 mode with CIR and CBS |
166 |
Set Policer L2 |
flow.bandwidth_control.policer.set() |
Set policer L2 mode with CIR and CBS |
167 |
Set Shaper L1 |
flow.bandwidth_control.shaper.set() |
Set shaper L1 mode with CIR, CBS, and buffer size |
168 |
Set Shaper L2 |
flow.bandwidth_control.shaper.set() |
Set shaper L2 mode with CIR, CBS, and buffer size |
169 |
Get Flow RX Total |
flow.statistics.rx.total.get() |
Get flow receive total statistics |
170 |
Get Flow TX Total |
flow.statistics.tx.total.get() |
Get flow transmit total statistics |
171 |
Get Flow Drop Packets |
flow.statistics.total.drop_packets.get() |
Get flow drop packet statistics |
172 |
Get Flow Corrupted Packets |
flow.statistics.total.corrupted_packets.get() |
Get flow corrupted packet statistics |
173 |
Get Flow Latency Packets |
flow.statistics.total.latency_packets.get() |
Get flow latency packet statistics |
174 |
Get Flow Jittered Packets |
flow.statistics.total.jittered_packets.get() |
Get flow jittered packet statistics |
175 |
Get Flow Duplicated Packets |
flow.statistics.total.duplicated_packets.get() |
Get flow duplicated packet statistics |
176 |
Get Flow Misordered Packets |
flow.statistics.total.mis_ordered_packets.get() |
Get flow misordered packet statistics |
177 |
Clear Flow TX Statistics |
flow.statistics.tx.clear.set() |
Clear flow transmit statistics |
178 |
Clear Flow RX Statistics |
flow.statistics.rx.clear.set() |
Clear flow receive statistics |
179 |
Clear Flow Statistics |
flow.statistics.clear.set() |
Clear all flow statistics |
180 |
Get Port Drop Statistics |
port.emulation.statistics.drop.get() |
Get port drop impairment statistics |
181 |
Get Port Corrupted Statistics |
port.emulation.statistics.corrupted.get() |
Get port corrupted packet statistics |
182 |
Get Port Latency Statistics |
port.emulation.statistics.latency.get() |
Get port latency packet statistics |
183 |
Get Port Jittered Statistics |
port.emulation.statistics.jittered.get() |
Get port jittered packet statistics |
184 |
Get Port Duplicated Statistics |
port.emulation.statistics.duplicated.get() |
Get port duplicated packet statistics |
185 |
Get Port Misordered Statistics |
port.emulation.statistics.mis_ordered.get() |
Get port misordered packet statistics |
186 |
Clear Port Statistics |
port.emulation.clear.set() |
Clear all port emulation statistics |
Network Impairment API Examples
# [chimera ports]
"""Get Port Object"""
port = module.ports.obtain(0)
"""Link Flap and PMA Error Injection"""
await port.pcs_pma.link_flap.params.set(duration=100, period=1000, repetition=0)
await port.pcs_pma.link_flap.enable.set_on()
await port.pcs_pma.link_flap.enable.set_off()
await port.pcs_pma.pma_error_inject.params.set(duration=100, period=1000, repetition=0, coeff=100, exp=-4)
await port.pcs_pma.pma_error_inject.enable.set_on()
await port.pcs_pma.pma_error_inject.enable.set_off()
"""Enable/Disable Impairment on the Port"""
# Enable impairment on the port.
await port.emulate.set_off()
await port.emulate.set_on()
resp = await port.emulate.get()
"""TPLD Mode Configuration"""
# Set TPLD mode
await port.emulation.tpld_mode.set(mode=enums.TPLDMode.NORMAL)
await port.emulation.tpld_mode.set(mode=enums.TPLDMode.MICRO)
resp = await port.emulation.tpld_mode.get()
# [chimera flows]
"""Basic and Extended Flow Filter Configuration"""
# Configure flow's basic filter on a port
# Configure flow properties
flow = port.emulation.flows[1]
await flow.comment.set(comment="Flow description")
resp = await flow.comment.get()
resp.comment
# Initializing the shadow copy of the filter.
await flow.shadow_filter.initiating.set()
# Configure shadow filter to BASIC mode
await flow.shadow_filter.use_basic_mode()
# Query the mode of the filter (either basic or extended)
filter = await flow.shadow_filter.get_mode()
if isinstance(filter, misc.BasicImpairmentFlowFilter):
#------------------
# Ethernet subfilter
#------------------
# Use and configure basic-mode shadow filter's Ethernet subfilter
await utils.apply(
filter.ethernet.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.EXCLUDE),
filter.ethernet.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.INCLUDE),
filter.ethernet.src_address.set(use=enums.OnOff.ON, value=Hex("AAAAAAAAAAAA"), mask=Hex("FFFFFFFFFFFF")),
filter.ethernet.dest_address.set(use=enums.OnOff.ON, value=Hex("BBBBBBBBBBBB"), mask=Hex("FFFFFFFFFFFF"))
)
#------------------
# Layer 2+ subfilter
#------------------
# Not use basic-mode shadow filter's Layer 2+ subfilter
await filter.l2plus_use.set(use=enums.L2PlusPresent.NA)
# Use and configure basic-mode shadow filter's Layer2+ subfilter (One VLAN tag)
await utils.apply(
filter.l2plus_use.set(use=enums.L2PlusPresent.VLAN1),
filter.vlan.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.EXCLUDE),
filter.vlan.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.INCLUDE),
filter.vlan.inner.tag.set(use=enums.OnOff.ON, value=1234, mask=Hex("0FFF")),
filter.vlan.inner.pcp.set(use=enums.OnOff.OFF, value=3, mask=Hex("07")),
)
# Use and configure basic-mode shadow filter's Layer2+ subfilter (Two VLAN tag)
await utils.apply(
filter.l2plus_use.set(use=enums.L2PlusPresent.VLAN2),
filter.vlan.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.EXCLUDE),
filter.vlan.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.INCLUDE),
filter.vlan.inner.tag.set(use=enums.OnOff.ON, value=1234, mask=Hex("0FFF")),
filter.vlan.inner.pcp.set(use=enums.OnOff.OFF, value=3, mask=Hex("07")),
filter.vlan.outer.tag.set(use=enums.OnOff.ON, value=2345, mask=Hex("0FFF")),
filter.vlan.outer.pcp.set(use=enums.OnOff.OFF, value=0, mask=Hex("07")),
)
# Use and configure basic-mode shadow filter's Layer2+ subfilter (MPLS)
await utils.apply(
filter.l2plus_use.set(use=enums.L2PlusPresent.MPLS),
filter.mpls.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.EXCLUDE),
filter.mpls.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.INCLUDE),
filter.mpls.label.set(use=enums.OnOff.ON, value=1000, mask=Hex("FFFFF")),
filter.mpls.toc.set(use=enums.OnOff.ON, value=0, mask=Hex("07")),
)
#------------------
# Layer 3 subfilter
#------------------
# Not use basic-mode shadow filter's Layer 3 subfilter
await filter.l3_use.set(use=enums.L3Present.NA)
# Use and configure basic-mode shadow filter's Layer 3 subfilter (IPv4)
await utils.apply(
filter.l3_use.set(use=enums.L3Present.IP4),
filter.ip.v4.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.EXCLUDE),
filter.ip.v4.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.INCLUDE),
filter.ip.v4.src_address.set(use=enums.OnOff.ON, value=ipaddress.IPv4Address("10.0.0.2"), mask=Hex("FFFFFFFF")),
filter.ip.v4.dest_address.set(use=enums.OnOff.ON, value=ipaddress.IPv4Address("10.0.0.2"), mask=Hex("FFFFFFFF")),
filter.ip.v4.dscp.set(use=enums.OnOff.ON, value=0, mask=Hex("FC")),
)
# Use and configure basic-mode shadow filter's Layer 3 subfilter (IPv6)
await utils.apply(
filter.l3_use.set(use=enums.L3Present.IP6),
filter.ip.v6.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.EXCLUDE),
filter.ip.v6.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.INCLUDE),
filter.ip.v6.src_address.set(use=enums.OnOff.ON, value=ipaddress.IPv6Address("2001::2"), mask=Hex("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF")),
filter.ip.v6.dest_address.set(use=enums.OnOff.ON, value=ipaddress.IPv6Address("2002::2"), mask=Hex("FFFFFFFFFFFFFFFFFFFFFFFFFFFFFFFF")),
filter.ip.v6.traffic_class.set(use=enums.OnOff.ON, value=0, mask=Hex("FC")),
)
#------------------
# Layer 4 subfilter
#------------------
# Use and configure basic-mode shadow filter's Layer 4 subfilter (TCP)
await utils.apply(
filter.tcp.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.EXCLUDE),
filter.tcp.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.INCLUDE),
filter.tcp.src_port.set(use=enums.OnOff.ON, value=1234, mask=Hex("FFFF")),
filter.tcp.dest_port.set(use=enums.OnOff.ON, value=80, mask=Hex("FFFF")),
)
# Use and configure basic-mode shadow filter's Layer 4 subfilter (UDP)
await utils.apply(
filter.udp.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.EXCLUDE),
filter.udp.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.INCLUDE),
filter.udp.src_port.set(use=enums.OnOff.ON, value=1234, mask=Hex("FFFF")),
filter.udp.dest_port.set(use=enums.OnOff.ON, value=80, mask=Hex("FFFF")),
)
#------------------
# Layer Xena subfilter
#------------------
await utils.apply(
filter.tpld.settings.set(action=enums.InfoAction.EXCLUDE),
filter.tpld.settings.set(action=enums.InfoAction.INCLUDE),
filter.tpld.test_payload_filters_config[0].set(use=enums.OnOff.ON, id = 2),
filter.tpld.test_payload_filters_config[0].set(use=enums.OnOff.OFF, id = 2),
filter.tpld.test_payload_filters_config[1].set(use=enums.OnOff.ON, id = 4),
filter.tpld.test_payload_filters_config[1].set(use=enums.OnOff.OFF, id = 4),
filter.tpld.test_payload_filters_config[2].set(use=enums.OnOff.ON, id = 6),
filter.tpld.test_payload_filters_config[2].set(use=enums.OnOff.OFF, id = 6),
filter.tpld.test_payload_filters_config[3].set(use=enums.OnOff.ON, id = 8),
filter.tpld.test_payload_filters_config[3].set(use=enums.OnOff.OFF, id = 8),
filter.tpld.test_payload_filters_config[4].set(use=enums.OnOff.ON, id = 10),
filter.tpld.test_payload_filters_config[4].set(use=enums.OnOff.OFF, id = 10),
filter.tpld.test_payload_filters_config[5].set(use=enums.OnOff.ON, id = 20),
filter.tpld.test_payload_filters_config[5].set(use=enums.OnOff.OFF, id = 20),
filter.tpld.test_payload_filters_config[6].set(use=enums.OnOff.ON, id = 40),
filter.tpld.test_payload_filters_config[6].set(use=enums.OnOff.OFF, id = 40),
filter.tpld.test_payload_filters_config[7].set(use=enums.OnOff.ON, id = 60),
filter.tpld.test_payload_filters_config[7].set(use=enums.OnOff.OFF, id = 60),
filter.tpld.test_payload_filters_config[8].set(use=enums.OnOff.ON, id = 80),
filter.tpld.test_payload_filters_config[8].set(use=enums.OnOff.OFF, id = 80),
filter.tpld.test_payload_filters_config[9].set(use=enums.OnOff.ON, id = 100),
filter.tpld.test_payload_filters_config[9].set(use=enums.OnOff.OFF, id = 100),
filter.tpld.test_payload_filters_config[10].set(use=enums.OnOff.ON, id = 102),
filter.tpld.test_payload_filters_config[10].set(use=enums.OnOff.OFF, id = 102),
filter.tpld.test_payload_filters_config[11].set(use=enums.OnOff.ON, id = 104),
filter.tpld.test_payload_filters_config[11].set(use=enums.OnOff.OFF, id = 104),
filter.tpld.test_payload_filters_config[12].set(use=enums.OnOff.ON, id = 106),
filter.tpld.test_payload_filters_config[12].set(use=enums.OnOff.OFF, id = 106),
filter.tpld.test_payload_filters_config[13].set(use=enums.OnOff.ON, id = 108),
filter.tpld.test_payload_filters_config[13].set(use=enums.OnOff.OFF, id = 108),
filter.tpld.test_payload_filters_config[14].set(use=enums.OnOff.ON, id = 110),
filter.tpld.test_payload_filters_config[14].set(use=enums.OnOff.OFF, id = 110),
filter.tpld.test_payload_filters_config[15].set(use=enums.OnOff.ON, id = 200),
filter.tpld.test_payload_filters_config[15].set(use=enums.OnOff.OFF, id = 200),
)
#------------------
# Layer Any subfilter
#------------------
await utils.apply(
filter.any.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.EXCLUDE),
filter.any.settings.set(use=enums.FilterUse.AND, action=enums.InfoAction.INCLUDE),
filter.any.config.set(position=0, value=Hex("112233445566"), mask=Hex("112233445566"))
)
# Apply the filter so the configuration data in the shadow copy is committed to the working copy automatically.
await flow.shadow_filter.enable.set_off()
await flow.shadow_filter.enable.set_on()
await flow.shadow_filter.apply.set()
# Configure flow's extended filter on a port
# Configure flow properties
flow = port.emulation.flows[1]
await flow.comment.set("Flow description")
# Initializing the shadow copy of the filter.
await flow.shadow_filter.initiating.set()
# Configure shadow filter to EXTENDED mode
await flow.shadow_filter.use_extended_mode()
# Query the mode of the filter (either basic or extended)
filter = await flow.shadow_filter.get_mode()
if isinstance(filter, misc.ExtendedImpairmentFlowFilter):
await filter.use_segments(
enums.ProtocolOption.VLAN
)
protocol_segments = await filter.get_protocol_segments()
await protocol_segments[0].value.set(value=Hex("AAAAAAAAAAAABBBBBBBBBBBB8100"))
await protocol_segments[0].mask.set(masks=Hex("0000000000000000000000000000"))
await protocol_segments[1].value.set(value=Hex("0064FFFF"))
await protocol_segments[1].mask.set(masks=Hex("00000000"))
"""Impairment Distribution Configuration for Different Impairment Types"""
# Configure impairment - Drop
# Fixed Burst distribution for impairment Drop
await utils.apply(
flow.impairment_distribution.drop_type_config.fixed_burst.set(burst_size=5),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=5), #repeat (duration = 1, period = x)
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=0), #one shot
)
# Random Burst distribution for impairment Drop
await utils.apply(
flow.impairment_distribution.drop_type_config.random_burst.set(minimum=1, maximum=10, probability=10_000),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=1), # repeat pattern
flow.impairment_distribution.drop_type_config.schedule.set(duration=0, period=0), #continuous
)
# Fixed Rate distribution for impairment Drop
await utils.apply(
flow.impairment_distribution.drop_type_config.fixed_rate.set(probability=10_000),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.drop_type_config.schedule.set(duration=0, period=0), #continuous
)
# Bit Error Rate distribution for impairment Drop
await utils.apply(
flow.impairment_distribution.drop_type_config.bit_error_rate.set(coef=1, exp=1),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.drop_type_config.schedule.set(duration=0, period=0), #continuous
)
# Random Rate distribution for impairment Drop
await utils.apply(
flow.impairment_distribution.drop_type_config.random_rate.set(probability=10_000),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.drop_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gilbert Elliot distribution for impairment Drop
await utils.apply(
flow.impairment_distribution.drop_type_config.ge.set(good_state_prob=0, good_state_trans_prob=0, bad_state_prob=0, bad_state_trans_prob=0),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.drop_type_config.schedule.set(duration=0, period=0), #continuous
)
# Uniform distribution for impairment Drop
await utils.apply(
flow.impairment_distribution.drop_type_config.uniform.set(minimum=1, maximum=1),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.drop_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gaussian distribution for impairment Drop
await utils.apply(
flow.impairment_distribution.drop_type_config.gaussian.set(mean=1, std_deviation=1),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.drop_type_config.schedule.set(duration=0, period=0), #continuous
)
# Poisson distribution for impairment Drop
await utils.apply(
flow.impairment_distribution.drop_type_config.poisson.set(mean=9),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=1), # repeat pattern
flow.impairment_distribution.drop_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gamma distribution for impairment Drop
await utils.apply(
flow.impairment_distribution.drop_type_config.gamma.set(shape=1, scale=1),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=1), # repeat pattern
flow.impairment_distribution.drop_type_config.schedule.set(duration=0, period=0), #continuous
)
# Custom distribution for impairment Drop
data_x=[0, 1] * 256
await port.custom_distributions.assign(0)
await port.custom_distributions[0].comment.set(comment="Example Custom Distribution")
await port.custom_distributions[0].definition.set(linear=enums.OnOff.OFF, symmetric=enums.OnOff.OFF, entry_count=len(data_x), data_x=data_x)
await utils.apply(
flow.impairment_distribution.drop_type_config.custom.set(cust_id=0),
flow.impairment_distribution.drop_type_config.schedule.set(duration=1, period=1),
flow.impairment_distribution.drop_type_config.schedule.set(duration=0, period=0), #continuous
)
# Set distribution and start impairment Drop
await flow.impairment_distribution.drop_type_config.enable.set_on()
await flow.impairment_distribution.drop_type_config.enable.set_off()
# Configure impairment - Misordering
# Fixed Burst distribution for impairment Misordering
# dist = distributions.misordering.FixedBurst(burst_size=1)
# dist.repeat(period=5)
# dist.one_shot()
# Fixed Burst distribution for impairment Misordering
await utils.apply(
flow.impairment_distribution.misorder_type_config.fixed_burst.set(burst_size=5),
flow.impairment_distribution.misorder_type_config.schedule.set(duration=1, period=5), #repeat
flow.impairment_distribution.misorder_type_config.schedule.set(duration=1, period=0), #one shot
)
# Fixed Rate distribution for impairment Misordering
await utils.apply(
flow.impairment_distribution.misorder_type_config.fixed_rate.set(probability=10_000),
flow.impairment_distribution.misorder_type_config.schedule.set(duration=1, period=1), # repeat pattern
flow.impairment_distribution.misorder_type_config.schedule.set(duration=0, period=0), #continuous
)
# Set distribution and start impairment Misordering
await flow.misordering.set(depth=1)
await flow.impairment_distribution.misorder_type_config.enable.set_on()
await flow.impairment_distribution.misorder_type_config.enable.set_off()
# Configure impairment - Latency & Jitter
# Fixed Burst distribution for impairment Latency & Jitter
await flow.impairment_distribution.latency_jitter_type_config.constant_delay.set(delay=100)
# Random Burst distribution for impairment Latency & Jitter
await utils.apply(
flow.impairment_distribution.latency_jitter_type_config.accumulate_and_burst.set(delay=1300),
flow.impairment_distribution.latency_jitter_type_config.schedule.set(duration=1, period=1), #repeat (duration = 1, period = x)
flow.impairment_distribution.latency_jitter_type_config.schedule.set(duration=1, period=0), #one shot
)
# Step distribution for impairment Latency & Jitter
await utils.apply(
flow.impairment_distribution.latency_jitter_type_config.step.set(low=1300, high=77000),
flow.impairment_distribution.latency_jitter_type_config.schedule.set(duration=0, period=0), #continuous
)
await flow.impairment_distribution.corruption_type_config.off.set()
# Uniform distribution for impairment Latency & Jitter
await utils.apply(
flow.impairment_distribution.latency_jitter_type_config.uniform.set(minimum=1, maximum=1),
flow.impairment_distribution.latency_jitter_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gaussian distribution for impairment Latency & Jitter
await utils.apply(
flow.impairment_distribution.latency_jitter_type_config.gaussian.set(mean=1, std_deviation=1),
flow.impairment_distribution.latency_jitter_type_config.schedule.set(duration=0, period=0), #continuous
)
resp = await flow.latency_range.get()
resp.min
resp.max
# Poisson distribution for impairment Latency & Jitter
await utils.apply(
flow.impairment_distribution.latency_jitter_type_config.poisson.set(mean=1),
flow.impairment_distribution.latency_jitter_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gamma distribution for impairment Latency & Jitter
await utils.apply(
flow.impairment_distribution.latency_jitter_type_config.gamma.set(shape=1, scale=1),
flow.impairment_distribution.latency_jitter_type_config.schedule.set(duration=0, period=0), #continuous
)
# Custom distribution for impairment Latency & Jitter
data_x=[0, 1] * 256
await port.custom_distributions.assign(0)
await port.custom_distributions[0].comment.set(comment="Example Custom Distribution")
await port.custom_distributions[0].definition.set(linear=enums.OnOff.OFF, symmetric=enums.OnOff.OFF, entry_count=len(data_x), data_x=data_x)
await utils.apply(
flow.impairment_distribution.latency_jitter_type_config.custom.set(cust_id=0),
flow.impairment_distribution.latency_jitter_type_config.schedule.set(duration=0, period=0), #continuous
)
# Set distribution and start impairment Latency & Jitter
await flow.impairment_distribution.latency_jitter_type_config.enable.set_on()
await flow.impairment_distribution.latency_jitter_type_config.enable.set_off()
# Configure impairment - Duplication
# Fixed Burst distribution for impairment Duplication
# dist.one_shot()
await utils.apply(
flow.impairment_distribution.duplication_type_config.fixed_burst.set(burst_size=1300),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1), #repeat (duration = 1, period = x)
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=0), #one shot
)
# Random Burst distribution for impairment Duplication
await utils.apply(
flow.impairment_distribution.duplication_type_config.random_burst.set(minimum=1, maximum=1, probability=10_0000),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.duplication_type_config.schedule.set(duration=0, period=0), #continuous
)
# Fixed Rate distribution for impairment Duplication
await utils.apply(
flow.impairment_distribution.duplication_type_config.fixed_rate.set(probability=10_000),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1), # repeat pattern
flow.impairment_distribution.duplication_type_config.schedule.set(duration=0, period=0), #continuous
)
# Bit Error Rate distribution for impairment Duplication
await utils.apply(
flow.impairment_distribution.duplication_type_config.bit_error_rate.set(coef=1, exp=1),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.duplication_type_config.schedule.set(duration=0, period=0), #continuous
)
# Random Rate distribution for impairment Duplication
await utils.apply(
flow.impairment_distribution.duplication_type_config.random_rate.set(probability=10_000),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.duplication_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gilbert Elliot distribution for impairment Duplication
await utils.apply(
flow.impairment_distribution.duplication_type_config.ge.set(good_state_prob=0, good_state_trans_prob=0, bad_state_prob=0, bad_state_trans_prob=0),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.duplication_type_config.schedule.set(duration=0, period=0), #continuous
)
# Uniform distribution for impairment Duplication
await utils.apply(
flow.impairment_distribution.duplication_type_config.uniform.set(minimum=1, maximum=1),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.duplication_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gaussian distribution for impairment Duplication
await utils.apply(
flow.impairment_distribution.duplication_type_config.gaussian.set(mean=1, std_deviation=1),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.duplication_type_config.schedule.set(duration=0, period=0), #continuous
)
# Poisson distribution for impairment Duplication
await utils.apply(
flow.impairment_distribution.duplication_type_config.poisson.set(mean=9),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1), # repeat pattern
flow.impairment_distribution.duplication_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gamma distribution for impairment Duplication
await utils.apply(
flow.impairment_distribution.duplication_type_config.gamma.set(shape=1, scale=1),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1), # repeat pattern
flow.impairment_distribution.duplication_type_config.schedule.set(duration=0, period=0), #continuous
)
# Custom distribution for impairment Duplication
data_x=[0, 1] * 256
await port.custom_distributions.assign(0)
await port.custom_distributions[0].comment.set(comment="Example Custom Distribution")
await port.custom_distributions[0].definition.set(linear=enums.OnOff.OFF, symmetric=enums.OnOff.OFF, entry_count=len(data_x), data_x=data_x)
await utils.apply(
flow.impairment_distribution.duplication_type_config.custom.set(cust_id=0),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=1, period=1),
flow.impairment_distribution.duplication_type_config.schedule.set(duration=0, period=0), #continuous
)
# Set distribution and start impairment Duplication
await flow.impairment_distribution.duplication_type_config.enable.set_on()
await flow.impairment_distribution.duplication_type_config.enable.set_off()
# Configure impairment - Corruption
# Fixed Burst distribution for impairment Corruption
await utils.apply(
flow.impairment_distribution.corruption_type_config.fixed_burst.set(burst_size=1300),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1), #repeat (duration = 1, period = x)
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=0), #one shot
)
# Random Burst distribution for impairment Corruption
await utils.apply(
flow.impairment_distribution.corruption_type_config.random_burst.set(minimum=1, maximum=1, probability=10_0000),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.corruption_type_config.schedule.set(duration=0, period=0), #continuous
)
# Fixed Rate distribution for impairment Corruption
await utils.apply(
flow.impairment_distribution.corruption_type_config.fixed_rate.set(probability=10_000),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1), # repeat pattern
flow.impairment_distribution.corruption_type_config.schedule.set(duration=0, period=0), #continuous
)
# Bit Error Rate distribution for impairment Corruption
await utils.apply(
flow.impairment_distribution.corruption_type_config.bit_error_rate.set(coef=1, exp=1),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.corruption_type_config.schedule.set(duration=0, period=0), #continuous
)
# Random Rate distribution for impairment Corruption
await utils.apply(
flow.impairment_distribution.corruption_type_config.random_rate.set(probability=10_000),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.corruption_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gilbert Elliot distribution for impairment Corruption
await utils.apply(
flow.impairment_distribution.corruption_type_config.ge.set(good_state_prob=0, good_state_trans_prob=0, bad_state_prob=0, bad_state_trans_prob=0),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.corruption_type_config.schedule.set(duration=0, period=0), #continuous
)
# Uniform distribution for impairment Corruption
await utils.apply(
flow.impairment_distribution.corruption_type_config.uniform.set(minimum=1, maximum=1),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.corruption_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gaussian distribution for impairment Corruption
await utils.apply(
flow.impairment_distribution.corruption_type_config.gaussian.set(mean=1, std_deviation=1),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1),# repeat pattern
flow.impairment_distribution.corruption_type_config.schedule.set(duration=0, period=0), #continuous
)
# Poisson distribution for impairment Corruption
await utils.apply(
flow.impairment_distribution.corruption_type_config.poisson.set(mean=9),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1), # repeat pattern
flow.impairment_distribution.corruption_type_config.schedule.set(duration=0, period=0), #continuous
)
# Gamma distribution for impairment Corruption
await utils.apply(
flow.impairment_distribution.corruption_type_config.gamma.set(shape=1, scale=1),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1), # repeat pattern
flow.impairment_distribution.corruption_type_config.schedule.set(duration=0, period=0), #continuous
)
# Custom distribution for impairment Corruption
data_x=[0, 1] * 256
await port.custom_distributions.assign(0)
await port.custom_distributions[0].comment.set(comment="Example Custom Distribution")
await port.custom_distributions[0].definition.set(linear=enums.OnOff.OFF, symmetric=enums.OnOff.OFF, entry_count=len(data_x), data_x=data_x)
await utils.apply(
flow.impairment_distribution.corruption_type_config.custom.set(cust_id=0),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=1, period=1),
flow.impairment_distribution.corruption_type_config.schedule.set(duration=0, period=0), #continuous
)
# Set distribution and start impairment Corruption
await flow.corruption.set(corruption_type=enums.CorruptionType.OFF)
await flow.corruption.set(corruption_type=enums.CorruptionType.ETH)
await flow.corruption.set(corruption_type=enums.CorruptionType.IP)
await flow.corruption.set(corruption_type=enums.CorruptionType.TCP)
await flow.corruption.set(corruption_type=enums.CorruptionType.UDP)
await flow.corruption.set(corruption_type=enums.CorruptionType.BER)
await flow.impairment_distribution.corruption_type_config.enable.set_on()
await flow.impairment_distribution.corruption_type_config.enable.set_off()
await flow.impairment_distribution.corruption_type_config.enable.set_on()
resp = await flow.impairment_distribution.corruption_type_config.one_shot_status.get()
resp.one_shot_status
# Configure bandwidth control - Policer
await flow.bandwidth_control.policer.set(on_off=enums.OnOff.ON, mode=enums.PolicerMode.L1, cir=10_000, cbs=1_000)
await flow.bandwidth_control.policer.set(on_off=enums.OnOff.ON, mode=enums.PolicerMode.L2, cir=10_000, cbs=1_000)
# Configure bandwidth control - Shaper
# Set and start bandwidth control Shaper
await flow.bandwidth_control.shaper.set(on_off=enums.OnOff.ON, mode=enums.PolicerMode.L1, cir=10_000, cbs=1_000, buffer_size=1_000)
await flow.bandwidth_control.shaper.set(on_off=enums.OnOff.ON, mode=enums.PolicerMode.L2, cir=10_000, cbs=1_000, buffer_size=1_000)
"""Per-Flow Impairment Statistics"""
# Flow Rx statistics
rx_total = await flow.statistics.rx.total.get()
rx_total.byte_count
rx_total.packet_count
rx_total.l2_bps
rx_total.pps
# Flow Tx statistics
tx_total = await flow.statistics.tx.total.get()
tx_total.byte_count
tx_total.packet_count
tx_total.l2_bps
tx_total.pps
# Flow overall statistics
flow_drop_total = await flow.statistics.total.drop_packets.get()
flow_drop_total.pkt_drop_count_total
flow_drop_total.pkt_drop_count_programmed
flow_drop_total.pkt_drop_count_bandwidth
flow_drop_total.pkt_drop_count_other
flow_drop_total.pkt_drop_ratio_total
flow_drop_total.pkt_drop_ratio_programmed
flow_drop_total.pkt_drop_ratio_bandwidth
flow_drop_total.pkt_drop_ratio_other
# Flow corrupted statistics
flow_corrupted_total = await flow.statistics.total.corrupted_packets.get()
flow_corrupted_total.fcs_corrupted_pkt_count
flow_corrupted_total.fcs_corrupted_pkt_ratio
flow_corrupted_total.ip_corrupted_pkt_count
flow_corrupted_total.ip_corrupted_pkt_ratio
flow_corrupted_total.tcp_corrupted_pkt_count
flow_corrupted_total.tcp_corrupted_pkt_ratio
flow_corrupted_total.total_corrupted_pkt_count
flow_corrupted_total.total_corrupted_pkt_ratio
flow_corrupted_total.udp_corrupted_pkt_count
flow_corrupted_total.udp_corrupted_pkt_ratio
# Flow latency and jitter statistics
flow_delayed_total = await flow.statistics.total.latency_packets.get()
flow_delayed_total.pkt_count
flow_delayed_total.ratio
flow_jittered_total = await flow.statistics.total.jittered_packets.get()
flow_jittered_total.pkt_count
flow_jittered_total.ratio
# Flow duplication and misordering statistics
flow_duplicated_total = await flow.statistics.total.duplicated_packets.get()
flow_duplicated_total.pkt_count
flow_duplicated_total.ratio
flow_misordered_total = await flow.statistics.total.mis_ordered_packets.get()
flow_misordered_total.pkt_count
flow_misordered_total.ratio
# Clear flow statistics
await flow.statistics.tx.clear.set()
await flow.statistics.rx.clear.set()
await flow.statistics.clear.set()
"""Per-Port Impairment Statistics"""
# Port statistics
# Port drop statistics
port_drop = await port.emulation.statistics.drop.get()
port_drop.pkt_drop_count_total
port_drop.pkt_drop_count_programmed
port_drop.pkt_drop_count_bandwidth
port_drop.pkt_drop_count_other
port_drop.pkt_drop_ratio_total
port_drop.pkt_drop_ratio_programmed
port_drop.pkt_drop_ratio_bandwidth
port_drop.pkt_drop_ratio_other
# Port corrupt statistics
port_corrupted = await port.emulation.statistics.corrupted.get()
port_corrupted.fcs_corrupted_pkt_count
port_corrupted.fcs_corrupted_pkt_ratio
port_corrupted.ip_corrupted_pkt_count
port_corrupted.ip_corrupted_pkt_ratio
port_corrupted.tcp_corrupted_pkt_count
port_corrupted.tcp_corrupted_pkt_ratio
port_corrupted.total_corrupted_pkt_count
port_corrupted.total_corrupted_pkt_ratio
port_corrupted.udp_corrupted_pkt_count
port_corrupted.udp_corrupted_pkt_ratio
# Port latency and jitter statistics
port_delayed = await port.emulation.statistics.latency.get()
port_delayed.pkt_count
port_delayed.ratio
port_jittered = await port.emulation.statistics.jittered.get()
port_jittered.pkt_count
port_jittered.ratio
# Port duplication and misordering statistics
port_duplicated = await port.emulation.statistics.duplicated.get()
port_duplicated.pkt_count
port_duplicated.ratio
port_misordered = await port.emulation.statistics.mis_ordered.get()
port_misordered.pkt_count
port_misordered.ratio
# Clear port statistics
await port.emulation.clear.set()