"""
The Advanced Layer 1 functions
"""
from __future__ import annotations
import asyncio
from typing import (
TYPE_CHECKING,
Any,
Union,
List,
Tuple,
)
if TYPE_CHECKING:
from xoa_driver.ports import Z800FreyaPort, Z1600EdunPort
from xoa_driver.modules import Z800FreyaModule, Z1600EdunModule
FreyaEdunModule = Union[Z800FreyaModule, Z1600EdunModule]
FreyaEdunPort = Union[Z800FreyaPort, Z1600EdunPort]
from ..utils import apply
from ..enums import (
OnOff,
PcsErrorInjectionType,
)
import warnings
[docs]
async def get_tx_freq_curr(port: "FreyaEdunPort") -> int:
"""
Get the current Tx frequency in Hz of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The current Tx frequency in Hz.
:rtype: int
"""
resp = await port.layer1_adv.tx_freq.get()
return resp.current
[docs]
async def get_tx_freq_min(port: "FreyaEdunPort") -> int:
"""
Get the minimum Tx frequency in Hz of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The minimum Tx frequency in Hz.
:rtype: int
"""
resp = await port.layer1_adv.tx_freq.get()
return resp.minimum
[docs]
async def get_tx_freq_max(port: "FreyaEdunPort") -> int:
"""
Get the maximum Tx frequency in Hz of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The maximum Tx frequency in Hz.
:rtype: int
"""
resp = await port.layer1_adv.tx_freq.get()
return resp.maximum
[docs]
async def get_tx_freq_all(port: "FreyaEdunPort") -> Tuple[int, int, int]:
"""
Get the current, minimum, and maximum Tx frequency in Hz of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The current, minimum, and maximum Tx frequency in Hz.
:rtype: Tuple[int, int, int]
"""
resp = await port.layer1_adv.tx_freq.get()
return (resp.current, resp.minimum, resp.maximum)
[docs]
async def get_tx_ppm_curr(port: "FreyaEdunPort") -> int:
"""
Get the current Tx PPM of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The current Tx PPM.
:rtype: int
"""
resp = await port.layer1_adv.tx_ppm.get()
return resp.current
[docs]
async def get_tx_ppm_min(port: "FreyaEdunPort") -> int:
"""
Get the minimum Tx PPM of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The minimum Tx PPM.
:rtype: int
"""
resp = await port.layer1_adv.tx_ppm.get()
return resp.minimum
[docs]
async def get_tx_ppm_max(port: "FreyaEdunPort") -> int:
"""
Get the maximum Tx PPM of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The maximum Tx PPM.
:rtype: int
"""
resp = await port.layer1_adv.tx_ppm.get()
return resp.maximum
[docs]
async def get_tx_ppm_all(port: "FreyaEdunPort") -> Tuple[int, int, int]:
"""
Get the current, minimum, and maximum Tx PPM of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The current, minimum, and maximum Tx PPM.
:rtype: Tuple[int, int, int]
"""
resp = await port.layer1_adv.tx_ppm.get()
return (resp.current, resp.minimum, resp.maximum)
[docs]
async def get_tx_freq(port: "FreyaEdunPort") -> Tuple[int, int, int, int, int, int]:
"""
Get the current, minimum, and maximum Tx frequency (Hz) and frequency offset (ppm) of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: Tx frequency current, minimum, and maximum, and frequency offset (ppm) current, minimum, and maximum.
:rtype: Tuple[int, int, int, int, int, int]
"""
freq_resp, ppm_resp = await apply(
port.layer1_adv.tx_freq.get(),
port.layer1_adv.tx_ppm.get(),
)
return (freq_resp.current, freq_resp.minimum, freq_resp.maximum,
ppm_resp.current, ppm_resp.minimum, ppm_resp.maximum)
[docs]
async def get_rx_freq_curr(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[int]:
"""
Get the current Rx frequency in Hz of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The current Rx frequency in Hz of the specified Serdes.
:rtype: List[int]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_freq.get())
resps = await apply(*cmds)
for resp in resps:
results.append(resp.current)
return results
[docs]
async def get_rx_freq_min(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[int]:
"""
Get the minimum Rx frequency in Hz since last query of the specified Serdes.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The minimum Rx frequency in Hz since last query of the specified Serdes.
:rtype: List[int]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_freq.get())
resps = await apply(*cmds)
for resp in resps:
results.append(resp.minimum)
return results
[docs]
async def get_rx_freq_max(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[int]:
"""
Get the maximum Rx frequency in Hz since last query of the specified Serdes.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The maximum Rx frequency in Hz since last query of the specified Serdes.
:rtype: List[int]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_freq.get())
resps = await apply(*cmds)
for resp in resps:
results.append(resp.maximum)
return results
[docs]
async def get_rx_freq_all(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[Tuple[int, int, int]]:
"""
Get the current, minimum, and maximum Rx frequencies in Hz of the specified Serdes.
The minimum and maximum values are since last query.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The current, minimum, and maximum Rx frequencies in Hz of the specified Serdes.
:rtype: List[Tuple[int, int, int]]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_freq.get())
resps = await apply(*cmds)
for resp in resps:
results.append((resp.current, resp.minimum, resp.maximum))
return results
[docs]
async def get_rx_ppm_curr(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[int]:
"""
Get the current Rx PPM of the specified Serdes.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The current Rx PPM of the specified Serdes.
:rtype: List[int]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_ppm.get())
resps = await apply(*cmds)
for resp in resps:
results.append(resp.current)
return results
[docs]
async def get_rx_ppm_min(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[int]:
"""
Get the minimum Rx PPM since last query of the specified Serdes.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The minimum Rx PPM since last query of the specified Serdes.
:rtype: List[int]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_ppm.get())
resps = await apply(*cmds)
for resp in resps:
results.append(resp.minimum)
return results
[docs]
async def get_rx_ppm_max(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[int]:
"""
Get the maximum Rx PPM since last query of the specified Serdes.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The maximum Rx PPM since last query of the specified Serdes.
:rtype: List[int]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_ppm.get())
resps = await apply(*cmds)
for resp in resps:
results.append(resp.maximum)
return results
[docs]
async def get_rx_ppm_all(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[Tuple[int, int, int]]:
"""
Get the current, minimum, and maximum Rx PPM of the specified Serdes.
The minimum and maximum values are since last query.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The current, minimum, and maximum Rx PPM of the specified Serdes.
:rtype: List[Tuple[int, int, int]]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_ppm.get())
resps = await apply(*cmds)
for resp in resps:
results.append((resp.current, resp.minimum, resp.maximum))
return results
[docs]
async def get_rx_freq(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[Tuple[int, int, int, int, int, int]]:
"""
Get the current, minimum, and maximum Rx frequency (Hz) and frequency offset (ppm) of the specified Serdes.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: Rx frequency current, minimum, and maximum, and frequency offset (ppm) current, minimum, and maximum of the specified Serdes.
:rtype: List[Tuple[int, int, int, int, int, int]]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_freq.get())
cmds.append(port.layer1_adv.serdes[serdes_id].rx_ppm.get())
resps = await apply(*cmds)
for i in range(0, len(resps), 2):
freq_resp = resps[i]
ppm_resp = resps[i + 1]
results.append((freq_resp.current, freq_resp.minimum, freq_resp.maximum,
ppm_resp.current, ppm_resp.minimum, ppm_resp.maximum))
return results
[docs]
async def get_tx_datarate_curr(port: "FreyaEdunPort") -> int:
"""
Get the current Tx datarate in bps of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The current Tx datarate in bps.
:rtype: int
"""
resp = await port.layer1_adv.tx_datarate.get()
return resp.current
[docs]
async def get_tx_datarate_min(port: "FreyaEdunPort") -> int:
"""
Get the minimum Tx datarate in bps since last query.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The minimum Tx datarate in bps since last query.
:rtype: int
"""
resp = await port.layer1_adv.tx_datarate.get()
return resp.minimum
[docs]
async def get_tx_datarate_max(port: "FreyaEdunPort") -> int:
"""
Get the maximum Tx datarate in bps since last query.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The maximum Tx datarate in bps since last query.
:rtype: int
"""
resp = await port.layer1_adv.tx_datarate.get()
return resp.maximum
[docs]
async def get_tx_datarate_all(port: "FreyaEdunPort") -> Tuple[int, int, int]:
"""
Get the current, minimum, and maximum Tx datarate in bps of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The current, minimum, and maximum Tx datarate in bps.
:rtype: Tuple[int, int, int]
"""
resp = await port.layer1_adv.tx_datarate.get()
return (resp.current, resp.minimum, resp.maximum)
[docs]
async def get_rx_datarate_curr(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[int] | int:
"""
Get the current Rx datarate in bps of the specified port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The current Rx datarate in bps. If multiple Serdes indices are provided, a list of datarates is returned.
:rtype: int or List[int]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_datarate.get())
resps = await apply(*cmds)
for resp in resps:
results.append(resp.current)
if len(results) == 1:
return results[0]
return results
[docs]
async def get_rx_datarate_min(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[int] | int:
"""
Get the minimum Rx datarate in bps since last query.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The minimum Rx datarate in bps since last query. If multiple Serdes indices are provided, a list of minimum datarates is returned.
:rtype: int or List[int]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_datarate.get())
resps = await apply(*cmds)
for resp in resps:
results.append(resp.minimum)
if len(results) == 1:
return results[0]
return results
[docs]
async def get_rx_datarate_max(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[int] | int:
"""
Get the maximum Rx datarate in bps since last query.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The maximum Rx datarate in bps since last query. If multiple Serdes indices are provided, a list of maximum datarates is returned.
:rtype: int or List[int]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_datarate.get())
resps = await apply(*cmds)
for resp in resps:
results.append(resp.maximum)
if len(results) == 1:
return results[0]
return results
[docs]
async def get_rx_datarate_all(port: "FreyaEdunPort", serdes_indices: List[int] = [0]) -> List[Tuple[int, int, int]]:
"""
Get the current, minimum, and maximum Rx datarates in bps of the specified Serdes.
The minimum and maximum values are since last query.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: The current, minimum, and maximum Rx datarates in bps of the specified Serdes.
:rtype: List[Tuple[int, int, int]]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_datarate.get())
resps = await apply(*cmds)
for resp in resps:
results.append((resp.current, resp.minimum, resp.maximum))
return results
[docs]
async def get_cdr_lol_status(port: "FreyaEdunPort", serdes_indices: List[int]) -> List[Tuple[bool, bool]]:
"""
Get the current and latched CDR LOL status of the specified Serdes.
True means error condition is present, while False means error condition is not present.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: A list of tuples containing current and latched CDR LOL status for each Serdes.
:rtype: List[Tuple[bool, bool]]
"""
results = []
cmds = []
for serdes_id in serdes_indices:
cmds.append(port.layer1_adv.serdes[serdes_id].rx_cdr_lol.get())
resps = await apply(*cmds)
for resp in resps:
curr = True if resp.current.value == 1 else False
latched = True if resp.latched.value == 1 else False
results.append((curr, latched))
return results
[docs]
async def get_rx_pcsl_skew(port: "FreyaEdunPort", lane_indices: List[int]) -> List[Tuple[int, int]]:
"""Get Rx relative skew measured in bits of the specified PCS lanes.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param lane_indices: The indices of the PCS lanes.
:type lane_indices: List[int]
:return: PCSL and relative skew of the specified PCS lanes measured in bits
:rtype: List[Tuple[int, int]]
"""
results = []
cmds = []
for lane in lane_indices:
cmds.append(port.layer1.pcs.lane[lane].rx_status.status.get())
resps = await apply(*cmds)
for resp in resps:
results.append((resp.virtual_lane, resp.skew))
return results
[docs]
async def get_hi_ber_status(port: "FreyaEdunPort") -> Tuple[bool, bool]:
"""
Get the current and latched HI-BER status of the specified port.
True means error condition is present, while False means error condition is not present.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing current and latched HI-BER status.
:rtype: Tuple[bool, bool]
"""
resp = await port.layer1_adv.pcs.hi_ber.status.get()
curr = True if resp.current.value == 1 else False
latched = True if resp.latched.value == 1 else False
return (curr, latched)
[docs]
async def get_hi_ser_status(port: "FreyaEdunPort") -> Tuple[bool, bool, bool]:
"""
Get the current and latched HI-SER status of the specified port.
True means error condition is present, while False means error condition is not present.
HI-SER is signalled if 5560 RS-FEC symbol errors are detected in contiguous block of 8192 non-overlapping RS-FEC codewords.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing alarm state, current and latched HI-SER status.
:rtype: Tuple[bool, bool, bool]
"""
resp = await port.layer1_adv.pcs.hi_ser.status.get()
alarm_state = True if resp.alarm_state == OnOff.ON else False
curr = True if resp.current.value == 1 else False
latched = True if resp.latched.value == 1 else False
return (alarm_state, curr, latched)
[docs]
async def get_deg_ser_status(port: "FreyaEdunPort") -> Tuple[bool, bool]:
"""
Get the current and latched Degraded SER status of the specified port.
True means error condition is present, while False means error condition is not present.
The thresholds for signaling Degraded SER is programmable using
:py:func:`set_deg_ser_thresholds`.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing current and latched Degraded SER status.
:rtype: Tuple[bool, bool]
"""
resp = await port.layer1_adv.pcs.deg_ser.status.get()
curr = True if resp.current.value == 1 else False
latched = True if resp.latched.value == 1 else False
return (curr, latched)
[docs]
async def set_deg_ser_thresholds(port: "FreyaEdunPort", activate_threshold: int, deactivate_threshold: int, interval: int) -> None:
"""
Configure signaling Degraded SER thresholds of the port.
If more than `activate_threshold` number of RS-FEC symbol errors are detected
in a contiguous block of `interval` RS-FEC codewords, Degraded SER
is signalled on the port.
If less than `deactivate_threshold` number of RS-FEC symbol errors are detected
in a contiguous block of `interval` RS-FEC codewords, Degraded SER
is no longer signalled on the port.
`interval` must be an even number and a multiple of the number of PCS flows:
- 100G: 2 (one flow, but must be even)
- 200G/400G: 2 (two flows)
- 800G/1.6T: 4 (four flows)
An uncorrectable codeword is counted as 16 erroneous RS-FEC symbols.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param activate_threshold: The activate threshold.
:type activate_threshold: int
:param deactivate_threshold: The deactivate threshold.
:type deactivate_threshold: int
:param interval: The interval.
:type interval: int
"""
await port.layer1_adv.pcs.deg_ser.threshold.set(
act_thresh=activate_threshold,
deact_thresh=deactivate_threshold,
degrade_interval=interval,
)
[docs]
async def get_deg_ser_thresholds(
port: "FreyaEdunPort",
) -> Tuple[int, int, int]:
"""
Get signaling Degraded SER thresholds of the port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing activate threshold, deactivate threshold, and interval.
:rtype: Tuple[int, int, int]
"""
resp = await port.layer1_adv.pcs.deg_ser.threshold.get()
return (resp.act_thresh, resp.deact_thresh, resp.degrade_interval)
[docs]
async def get_lf_status(port: "FreyaEdunPort") -> Tuple[bool, bool]:
"""
Get the current and latched Local Fault status of the specified port.
True means error condition is present, while False means error condition is not present.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing current and latched Local Fault status.
:rtype: Tuple[bool, bool]
"""
resp = await port.layer1.rs_fault.status.get()
curr = True if resp.lf_current.value == 1 else False
latched = True if resp.lf_latched.value == 1 else False
return (curr, latched)
[docs]
async def get_rf_status(port: "FreyaEdunPort") -> Tuple[bool, bool]:
"""
Get the current and latched Remote Fault status of the specified port.
True means error condition is present, while False means error condition is not present.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing current and latched Remote Fault status.
:rtype: Tuple[bool, bool]
"""
resp = await port.layer1.rs_fault.status.get()
curr = True if resp.rf_current.value == 1 else False
latched = True if resp.rf_latched.value == 1 else False
return (curr, latched)
[docs]
async def get_lf_rf_status(port: "FreyaEdunPort") -> Tuple[bool, bool, bool, bool]:
"""
Get the current and latched Local Fault and Remote Fault status of the specified port.
True means error condition is present, while False means error condition is not present.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing current and latched Local Fault and Remote Fault status.
:rtype: Tuple[bool, bool, bool, bool]
"""
resp = await port.layer1.rs_fault.status.get()
lf_curr = True if resp.lf_current.value == 1 else False
lf_latched = True if resp.lf_latched.value == 1 else False
rf_curr = True if resp.rf_current.value == 1 else False
rf_latched = True if resp.rf_latched.value == 1 else False
return (lf_curr, lf_latched, rf_curr, rf_latched)
[docs]
async def get_link_down_status(port: "FreyaEdunPort") -> Tuple[bool, bool]:
"""
Get the current and latched Link Down status of the specified port.
True means error condition is present, while False means error condition is not present.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing current and latched Link Down status.
:rtype: Tuple[bool, bool]
"""
resp = await port.layer1_adv.pcs.link_down.status.get()
curr = True if resp.current.value == 1 else False
latched = True if resp.latched.value == 1 else False
return (curr, latched)
[docs]
async def get_rx_errors_since_clear(port: "FreyaEdunPort") -> Tuple[int, int, int, int, int, int]:
"""
Get the Rx number of the number of LOA, 256b/257 ITBs, 64b/66b erroneous codewords, link down, local fault, and remote fault events since the last counter clear.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing the received number of LOA, 256b/257 ITBs, 64b/66b erroneous codewords, link down, local fault, and remote fault events, since the last counter clear per port.
:rtype: Tuple[int, int, int, int, int, int]
"""
resp1, resp2 = await apply(
port.layer1_adv.pcs.err_inject.rx_cnt.get(),
port.layer1.rs_fault.stats.get(),
)
return (resp1.loa_count, resp1.itb_count, resp1.err_cw_count, resp1.link_down_count, resp2.lf_count, resp2.rf_count)
[docs]
async def get_tx_errors_since_clear(port: "FreyaEdunPort") -> Tuple[int, int, int, int]:
"""
Get the inject (Tx) number of LOA, HI-SER, 256b/257 ITBs, 64b/66b erroneous codewords since the last counter clear.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing the injected number of LOA, HI-SER, 256b/257 ITBs, 64b/66b erroneous codewords, since the last counter clear per port.
:rtype: Tuple[int, int, int, int]
"""
resp = await port.layer1_adv.pcs.err_inject.tx_cnt.get()
return (resp.loa_count, resp.hi_ser_count, resp.itb_count, resp.err_cw_count)
[docs]
async def inject_errcwd_once(port: "FreyaEdunPort") -> None:
"""
Inject a 64b/66b codeword error from the port immediately when called.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
"""
await port.layer1_adv.pcs.err_inject.inject.set(type=PcsErrorInjectionType.ERRCWD)
[docs]
async def inject_itb_once(port: "FreyaEdunPort") -> None:
"""
Inject an invalid 256b/257b transcode block (ITB) from the port immediately when called.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
"""
await port.layer1_adv.pcs.err_inject.inject.set(type=PcsErrorInjectionType.ITB)
[docs]
async def inject_loa_once(port: "FreyaEdunPort") -> None:
"""
Inject a Loss of Alignment (LOA) event from the port immediately when called.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
"""
await port.layer1_adv.pcs.err_inject.inject.set(type=PcsErrorInjectionType.LOA)
[docs]
async def inject_hi_ser_once(port: "FreyaEdunPort") -> None:
"""
Inject a High SER (HI-SER) event from the port immediately when called.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
"""
await port.layer1_adv.pcs.err_inject.inject.set(type=PcsErrorInjectionType.HISER)
[docs]
async def set_hi_ser_alarm(port: "FreyaEdunPort", on: bool) -> None:
"""
Set the HI-SER alarm on or off on the port.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param on: Set to `True` to enable the HI-SER alarm, or `False` to disable the HI-SER alarm.
"""
if on:
await port.layer1_adv.pcs.hi_ser.alarm.set_on()
else:
await port.layer1_adv.pcs.hi_ser.alarm.set_off()
[docs]
async def clear_rx_err_cnt(port: "FreyaEdunPort") -> None:
"""
Clear Rx Layer 1 advanced statistics error counters on the port.
"""
await port.layer1_adv.clear.set_rx()
[docs]
async def clear_tx_err_cnt(port: "FreyaEdunPort") -> None:
"""
Clear Tx Layer 1 advanced statistics error counters on the port.
"""
await port.layer1_adv.clear.set_tx()
# Deprecated functions
[docs]
async def get_cdr_lol(port: "FreyaEdunPort", serdes_indices: List[int]) -> List[Tuple[bool, bool]]:
"""
:py:func:`get_cdr_lol` is deprecated and will be removed in a future release. Please use :py:func:`get_cdr_lol_status` instead.
.. deprecated:: 1.8
Get the current and latched CDR LOL status of the specified Serdes.
True means error condition is present, while False means error condition is not present.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param serdes_indices: The indices of the Serdes.
:type serdes_indices: List[int]
:return: A list of tuples containing the current and latched CDR LOL status for each specified Serdes.
:rtype: List[Tuple[bool, bool]]
"""
warnings.warn(message="get_cdr_lol is deprecated and will be removed in a future release. Please use get_cdr_lol_status instead.", category=DeprecationWarning, stacklevel=2)
return await get_cdr_lol_status(port, serdes_indices)
[docs]
async def get_rx_lane_skew(port: "FreyaEdunPort", lane_indices: List[int]) -> List[Tuple[int, int]]:
"""
:py:func:`get_rx_lane_skew` is deprecated and will be removed in a future release. Please use :py:func:`get_rx_pcsl_skew` instead.
.. deprecated:: 1.8
Get Rx relative skew measured in bits of the specified PCS lanes.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:param lane_indices: The indices of the PCS lanes.
:type lane_indices: List[int]
:return: A list of tuples containing the current and latched RX lane skew for each specified PCS lane.
:rtype: List[Tuple[int, int]]
"""
warnings.warn(message="get_rx_lane_skew is deprecated and will be removed in a future release. Please use get_rx_pcsl_skew instead.", category=DeprecationWarning, stacklevel=2)
return await get_rx_pcsl_skew(port, lane_indices)
[docs]
async def get_hi_ber(port: "FreyaEdunPort") -> Tuple[bool, bool]:
"""
:py:func:`get_hi_ber` is deprecated and will be removed in a future release. Please use :py:func:`get_hi_ber_status` instead.
.. deprecated:: 1.8
Get the current and latched HI-BER status of the specified port.
True means error condition is present, while False means error condition is not present.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing the current and latched HI-BER status of the port.
:rtype: Tuple[bool, bool]
"""
warnings.warn(message="get_hi_ber is deprecated and will be removed in a future release. Please use get_hi_ber_status instead.", category=DeprecationWarning, stacklevel=2)
return await get_hi_ber_status(port)
[docs]
async def get_hi_ser(port: "FreyaEdunPort") -> Tuple[bool, bool, bool]:
"""
:py:func:`get_hi_ser` is deprecated and will be removed in a future release. Please use :py:func:`get_hi_ser_status` instead.
.. deprecated:: 1.8
Get the current and latched HI-SER status of the specified port.
True means error condition is present, while False means error condition is not present.
HI-SER is signalled if 5560 RS-FEC symbol errors are detected in contiguous block of 8192 non-overlapping RS-FEC codewords.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing the current and latched HI-SER status of the port.
:rtype: Tuple[bool, bool, bool]
"""
warnings.warn(message="get_hi_ser is deprecated and will be removed in a future release. Please use get_hi_ser_status instead.", category=DeprecationWarning, stacklevel=2)
return await get_hi_ser_status(port)
[docs]
async def get_deg_ser(port: "FreyaEdunPort") -> Tuple[bool, bool]:
"""
:py:func:`get_deg_ser` is deprecated and will be removed in a future release. Please use :py:func:`get_deg_ser_status` instead.
.. deprecated:: 1.8
Get the current and latched DEG-SER status of the specified port.
True means error condition is present, while False means error condition is not present.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: A tuple containing the current and latched DEG-SER status of the port.
:rtype: Tuple[bool, bool]
"""
warnings.warn(message="get_deg_ser is deprecated and will be removed in a future release. Please use get_deg_ser_status instead.", category=DeprecationWarning, stacklevel=2)
return await get_deg_ser_status(port)
[docs]
async def set_cw_err(port: "FreyaEdunPort") -> None:
"""
:py:func:`set_cw_err` is deprecated and will be removed in a future release. Please use :py:func:`inject_errcwd_once` instead.
.. deprecated:: 1.8
Inject a 64b/66b codeword error from the port immediately when called.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort
"""
warnings.warn(message="set_cw_err is deprecated and will be removed in a future release. Please use inject_errcwd_once instead.", category=DeprecationWarning, stacklevel=2)
await inject_errcwd_once(port)
[docs]
async def set_itb(port: "FreyaEdunPort") -> None:
"""
:py:func:`set_itb` is deprecated and will be removed in a future release. Please use :py:func:`inject_itb_once` instead.
.. deprecated:: 1.8
Inject an invalid 256b/257b transcode block (ITB) from the port immediately when called.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
"""
warnings.warn(message="set_itb is deprecated and will be removed in a future release. Please use inject_itb_once instead.", category=DeprecationWarning, stacklevel=2)
await inject_itb_once(port)
[docs]
async def get_cw_err_since_last(port: "FreyaEdunPort") -> int:
"""
:py:func:`get_cw_err_since_last` is deprecated and will be removed in a future release. Please use :py:func:`get_rx_errors_since_clear` instead.
.. deprecated:: 1.8
Get the number of 64b/66b erroneous codewords received since last counter clear.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The number of 64b/66b erroneous codewords received since last counter clear.
:rtype: int
"""
warnings.warn(message="get_cw_err_since_last is deprecated and will be removed in a future release. Please use get_rx_errors_since_clear instead.", category=DeprecationWarning, stacklevel=2)
resp = await port.layer1_adv.pcs.err_inject.rx_cnt.get()
return resp.err_cw_count
[docs]
async def get_itb_since_last(port: "FreyaEdunPort") -> int:
"""
:py:func:`get_itb_since_last` is deprecated and will be removed in a future release. Please use :py:func:`get_rx_errors_since_clear` instead.
.. deprecated:: 1.8
Get the number of 256b/257 ITBs received since last counter clear.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The number of 256b/257 ITBs received since last counter clear.
:rtype: int
"""
warnings.warn(message="get_itb_since_last is deprecated and will be removed in a future release. Please use get_rx_errors_since_clear instead.", category=DeprecationWarning, stacklevel=2)
resp = await port.layer1_adv.pcs.err_inject.rx_cnt.get()
return resp.itb_count
[docs]
async def get_total_loa_since_last(port: "FreyaEdunPort") -> int:
"""
:py:func:`get_total_loa_since_last` is deprecated and will be removed in a future release. Please use :py:func:`get_rx_errors_since_clear` instead.
.. deprecated:: 1.8
Get the number of LOA events received since last counter clear.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The number of LOA events received since last counter clear.
:rtype: int
"""
warnings.warn(message="get_total_loa_since_last is deprecated and will be removed in a future release. Please use get_rx_errors_since_clear instead.", category=DeprecationWarning, stacklevel=2)
resp = await port.layer1_adv.pcs.err_inject.rx_cnt.get()
return resp.loa_count
[docs]
async def get_link_down_since_last(port: "FreyaEdunPort") -> int:
"""
:py:func:`get_link_down_since_last` is deprecated and will be removed in a future release. Please use :py:func:`get_rx_errors_since_clear` instead.
.. deprecated:: 1.8
Get the number of link down events received since last counter clear.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The number of link down events received since last counter clear.
:rtype: int
"""
warnings.warn(message="get_link_down_since_last is deprecated and will be removed in a future release. Please use get_rx_errors_since_clear instead.", category=DeprecationWarning, stacklevel=2)
resp = await port.layer1_adv.pcs.err_inject.rx_cnt.get()
return resp.link_down_count
[docs]
async def get_local_fault_since_last(port: "FreyaEdunPort") -> int:
"""
:py:func:`get_local_fault_since_last` is deprecated and will be removed in a future release. Please use :py:func:`get_rx_errors_since_clear` instead.
.. deprecated:: 1.8
Get the number of local fault events received since last counter clear.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The number of local fault events received since last counter clear.
:rtype: int
"""
warnings.warn(message="get_local_fault_since_last is deprecated and will be removed in a future release. Please use get_rx_errors_since_clear instead.", category=DeprecationWarning, stacklevel=2)
resp = await port.layer1.rs_fault.stats.get()
return resp.lf_count
[docs]
async def get_remote_fault_since_last(port: "FreyaEdunPort") -> int:
"""
:py:func:`get_remote_fault_since_last` is deprecated and will be removed in a future release. Please use :py:func:`get_rx_errors_since_clear` instead.
.. deprecated:: 1.8
Get the number of remote fault events received since last counter clear.
:param port: The port instance.
:type port: Union[Z800FreyaPort, Z1600EdunPort]
:return: The number of remote fault events received since last counter clear.
:rtype: int
"""
warnings.warn(message="get_remote_fault_since_last is deprecated and will be removed in a future release. Please use get_rx_errors_since_clear instead.", category=DeprecationWarning, stacklevel=2)
resp = await port.layer1.rs_fault.stats.get()
return resp.rf_count
__all__ = (
"get_tx_freq_curr",
"get_tx_freq_min",
"get_tx_freq_max",
"get_tx_freq_all",
"get_tx_ppm_curr",
"get_tx_ppm_min",
"get_tx_ppm_max",
"get_tx_ppm_all",
"get_tx_freq",
"get_rx_freq_curr",
"get_rx_freq_min",
"get_rx_freq_max",
"get_rx_freq_all",
"get_rx_ppm_curr",
"get_rx_ppm_min",
"get_rx_ppm_max",
"get_rx_ppm_all",
"get_rx_freq",
"get_tx_datarate_curr",
"get_tx_datarate_min",
"get_tx_datarate_max",
"get_tx_datarate_all",
"get_rx_datarate_curr",
"get_rx_datarate_min",
"get_rx_datarate_max",
"get_rx_datarate_all",
"get_cdr_lol_status",
"get_rx_pcsl_skew",
"get_hi_ber_status",
"get_hi_ser_status",
"get_deg_ser_status",
"set_deg_ser_thresholds",
"get_deg_ser_thresholds",
"get_lf_status",
"get_rf_status",
"get_lf_rf_status",
"get_link_down_status",
"get_rx_errors_since_clear",
"get_tx_errors_since_clear",
"inject_errcwd_once",
"inject_itb_once",
"inject_loa_once",
"inject_hi_ser_once",
"set_hi_ser_alarm",
"clear_rx_err_cnt",
"clear_tx_err_cnt",
# Deprecated functions
"get_cdr_lol",
"get_rx_lane_skew",
"get_hi_ber",
"get_hi_ser",
"get_deg_ser",
"set_cw_err",
"set_itb",
# Functions to get individual Rx error counters since last clear.
"get_cw_err_since_last",
"get_itb_since_last",
"get_total_loa_since_last",
"get_link_down_since_last",
"get_local_fault_since_last",
"get_remote_fault_since_last",
)