MOKA/pqAutomationApp
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

1.1.0版本

Browse Source
This commit is contained in:
xinzhu.yin
2026-04-16 16:51:05 +08:00
commit c157e774e5
333 changed files with 70759 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
UniTAP/dev/ports/modules/fec/__init__.py Normal file
View File

@@ -0,0 +1,3 @@
from .fec_tx import FecTx
from .fec_rx import FecRx
from .fec_shared import FECCounters, FECErrorType128b132b, FECErrorType8b10b

131
UniTAP/dev/ports/modules/fec/fec_rx.py Normal file
View File

@@ -0,0 +1,131 @@
from UniTAP.libs.lib_tsi.tsi import *
from UniTAP.libs.lib_tsi.tsi_io import PortIO
from .fec_shared import FECCounters
from UniTAP.dev.ports.modules.dpcd.dpcd import DPCDRegisters
class FecRx:
"""
Class `FecRx` allows working with FEC functionality from Sink (RX - receiver) side. You can:
- Check capable FEC or not `is_capable`.
- Check enabled FEC or not `is_enabled`.
- Enable/Disable FEC `enable`.
- Enable/Disable calculating sum of errors `aggregate_errors`.
- Get error counters `get_error_counters`.
- Clear all errors `clear`.
"""
def __init__(self, port_io: PortIO, dpcd: DPCDRegisters):
self.__io = port_io
self.__dpcd = dpcd
self.__aggregate_error = 0
def is_enabled(self) -> bool:
"""
Returns status of FEC, is enabled or not.
Returns:
object of `bool` type.
"""
result = self.__io.get(TSI_DPRX_FEC_STATUS_R, c_int)
status_fec = ((result[1] & 0x1) != 0)
return bool(status_fec)
def is_capable(self) -> bool:
"""
Returns status of FEC, is capable or not.
Returns:
object of `bool` type.
"""
result = self.__io.get(TSI_DPRX_FEC_CTRL, c_int)
enabled_fec = (result[1] & 0x1) != 0
return enabled_fec
def enable(self, enable: bool):
"""
Enable/Disable FEC.
Args:
enable (bool) - enable (True) or disable (False)
"""
val = 0x1 if enable else 0x0
self.__io.set(TSI_DPRX_FEC_CTRL, val)
def aggregate_errors(self, enable: bool):
"""
Enable/Disable calculating sum of errors.
Args:
enable (bool) - enable (True) or disable (False)
"""
result = self.__io.get(TSI_DPRX_FEC_CONTROL, c_int)
val = result[1]
if enable:
val |= 0x2
self.__aggregate_error = 1
else:
val &= ~0x2
self.__aggregate_error = 0
self.__io.set(TSI_DPRX_FEC_CONTROL, val)
def get_error_counters(self) -> FECCounters:
"""
Get current error counters.
Returns:
object of `FECCounters` type
"""
result = FECCounters()
lane_count = self.__io.get(TSI_R_DPRX_LINK_LANE_COUNT, c_int)[1]
if lane_count == 4:
lane_count += 1 if self.__aggregate_error else 0
dpcd = self.__dpcd.read(0x120, 1).data[0]
dpcdaggr = bool(dpcd & (1 << 6))
for i in range(lane_count):
if i == 0 and dpcdaggr:
dpcd &= ~(1 << 6)
if i == 4 and dpcdaggr:
dpcd |= 1 << 6
elif i == 4:
continue
if i < 4:
dpcd &= ~(0x3 << 4)
dpcd |= (i << 4)
for j in range(1, 6):
dpcd &= ~(0x7 << 1)
v = dpcd | (j << 1)
self.__dpcd.write(0x120, v)
val = int.from_bytes(self.__dpcd.read(0x281, 2).data, 'little')
if j == 1:
result.uncorrectedBlockErrors[i] = val & 0x7FFF if val & 0x8000 else None
elif j == 2:
result.correctedBlockErrors[i] = val & 0x7FFF if val & 0x8000 else None
elif j == 3:
result.bitErrors[i] = val & 0x7FFF if val & 0x8000 else None
elif j == 4:
result.parityBlockErrors[i] = val & 0x7FFF if val & 0x8000 else None
elif j == 5:
result.parityBitErrors[i] = val & 0x7FFF if val & 0x8000 else None
return result
def clear(self):
"""
Clear all errors.
"""
address = 0x120
dpcd = self.__dpcd.read(address, 1).data[0]
old_fec = dpcd
fec_ready = dpcd & ~0xE
dpcd = fec_ready
self.__dpcd.write(address, dpcd)
dpcd = old_fec
self.__dpcd.write(address, dpcd)

38
UniTAP/dev/ports/modules/fec/fec_shared.py Normal file
View File

@@ -0,0 +1,38 @@
from enum import IntEnum
class FECCounters:
"""
Class `FECCounters` possible errors:
- Uncorrected block errors.
- Corrected block errors.
- Bit errors.
- Parity block errors.
- Parity bit errors.
"""
def __init__(self):
self.uncorrectedBlockErrors = [None, None, None, None, None]
self.correctedBlockErrors = [None, None, None, None, None]
self.bitErrors = [None, None, None, None, None]
self.parityBlockErrors = [None, None, None, None, None]
self.parityBitErrors = [None, None, None, None, None]
class FECErrorType8b10b(IntEnum):
"""
Describes possible FEC 8b/10b errors.
"""
UNCORRECTED_BLOCK = 0
CORRECTED_BLOCK = 1
CORRECTED_PARITY = 2
CORRECTED_BLOCK_1 = 3
CORRECTED_PARITY_1 = 4
class FECErrorType128b132b(IntEnum):
"""
Describes possible FEC 128b/132b errors.
"""
UNCORRECTED_BLOCK = 0
CORRECTED_BLOCK_4 = 1
CORRECTED_BLOCK_2 = 3

204
UniTAP/dev/ports/modules/fec/fec_tx.py Normal file
View File

@@ -0,0 +1,204 @@
from UniTAP.libs.lib_tsi.tsi import *
from UniTAP.libs.lib_tsi.tsi_io import PortIO
from .fec_shared import FECCounters, FECErrorType8b10b, FECErrorType128b132b
from UniTAP.dev.ports.modules.dpcd.dpcd import DPCDRegisters
from typing import Union
class FecTx:
"""
Class `FecTx` allows working with FEC functionality from Source (TX - transmitter) side. You can:
- Check enabled FEC or not `is_enabled`.
- Check state that FEC is prefers after link training `is_prefer_after_lt`.
- Enable/Disable FEC `enable`.
- Enable/Disable intent FEC `enable_intent`.
- Enable/Disable calculating sum of errors `aggregate_errors`.
- Get error counters `get_error_counters`.
- Clear all errors `clear`.
"""
def __init__(self, port_io: PortIO, dpcd: DPCDRegisters):
self.__io = port_io
self.__dpcd = dpcd
self.__aggregate_error = 0
def is_enabled(self) -> bool:
"""
Returns status of FEC, is enabled or not.
Returns:
object of `bool` type.
"""
result = self.__io.get(TSI_DPTX_FEC_STATUS_R, c_int)
status_fec = ((result[1] & 0x1) != 0)
return bool(status_fec)
def is_prefer_after_lt(self) -> bool:
"""
Check state that FEC is prefers after link training.
Returns:
object of `bool` type.
"""
result = self.__io.get(TSI_DPTX_FEC_CTRL, c_int)
enabled_fec = (result[1] & 0x2) != 0
return enabled_fec
def enable(self, enable: bool):
"""
Enable/Disable FEC.
Args:
enable (bool) - enable (True) or disable (False)
"""
result = self.__io.get(TSI_DPTX_FEC_CTRL, c_int)
val = result[1]
val |= 0x8 if enable else 0x10
self.__io.set(TSI_DPTX_FEC_CTRL, val)
def enable_intent(self, enable: bool):
"""
Enable/Disable intent FEC.
Args:
enable (bool) - enable (True) or disable (False)
"""
result = self.__io.get(TSI_DPTX_FEC_CTRL, c_int)
val = result[1]
val |= 1 if enable else 4
self.__io.set(TSI_DPTX_FEC_CTRL, val)
def aggregate_errors(self, enable: bool):
"""
Enable/Disable calculating sum of errors.
Args:
enable (bool) - enable (True) or disable (False)
"""
result = self.__io.get(TSI_DPTX_FEC_CONTROL, c_int)
val = result[1]
if enable:
val |= 0x40
self.__aggregate_error = 1
else:
val &= ~0x40
self.__aggregate_error = 0
self.__io.set(TSI_DPTX_FEC_CONTROL, val)
def generate_errors(self, error_type: Union[FECErrorType8b10b, FECErrorType128b132b], lane: list, ms: int = 100):
"""
Generate FEC errors.
Args:
error_type (Union[`FECErrorType8b10b`, `FECErrorType128b132b`])
lane (list)
ms (int) - time in m seconds
"""
result, status = self.__io.get(TSI_DPTX_LINK_MODE_R, c_int)
resut, hw_caps, size = self.__io.get(TSI_DPTX_HW_CAPS_R, c_uint32, 4)
if status == 0:
if isinstance(error_type, FECErrorType128b132b):
assert False, "This device doesn't support 128b/132b" if (hw_caps[1] & 0x7) == 0 else "Change link mode!"
if status == 1:
if isinstance(error_type, FECErrorType8b10b):
assert False, "This device doesn't support 8b/10b" if (hw_caps[1] & 0x7) == 0 else "Change link mode!"
dpcd = self.__dpcd.read(0x120, 1).data[0]
dpcd |= 2
self.__dpcd.write(0x120, dpcd)
delay = ms * 100
self.__io.set(TSI_MLEG_CONTROL, 0)
nl = self.__io.get(TSI_R_DPTX_LINK_STATUS_LANE_COUNT, c_uint32)[1]
self.__io.set(TSI_MLEG_SYMBOL_REPLACE_A, 0x00010000)
self.__io.set(TSI_MLEG_SYMBOL_REPLACE_MASK_A, 0x00010001)
self.__io.set(TSI_MLEG_SYMBOL_REPLACE_B, 0x00000001)
self.__io.set(TSI_MLEG_SYMBOL_REPLACE_MASK_B, 0x00010001)
self.__io.set(TSI_MLEG_DELAY_COUNTER, delay)
n = lane[0] << 16
self.__io.set(TSI_MLEG_LANE0_REPLACE_COUNTERS, n)
n = lane[1] << 16
self.__io.set(TSI_MLEG_LANE1_REPLACE_COUNTERS, n)
n = lane[2] << 16
self.__io.set(TSI_MLEG_LANE2_REPLACE_COUNTERS, n)
n = lane[3] << 16
self.__io.set(TSI_MLEG_LANE3_REPLACE_COUNTERS, n)
ctrl = 0
ctrl |= (error_type.value << 16)
if nl == 1:
ctrl |= (1 << 13)
ctrl |= (1 << 12)
ctrl |= (0xF << 4)
if lane[0]:
ctrl |= (1 << 0)
if lane[1]:
ctrl |= (1 << 1)
if lane[2]:
ctrl |= (1 << 2)
if lane[3]:
ctrl |= (1 << 3)
self.__io.set(TSI_MLEG_CONTROL, ctrl)
def get_error_counters(self) -> FECCounters:
"""
Get current error counters.
Returns:
object of `FECCounters` type
"""
result = FECCounters()
lane_count = self.__io.get(TSI_R_DPTX_LINK_STATUS_LANE_COUNT, c_uint32)[1]
if lane_count == 4:
lane_count += 1 if self.__aggregate_error else 0
dpcd = self.__dpcd.read(0x120, 1).data[0]
dpcdaggr = bool(dpcd & (1 << 6))
for i in range(lane_count):
if i == 0 and dpcdaggr:
dpcd &= ~(1 << 6)
if i == 4 and dpcdaggr:
dpcd |= 1 << 6
elif i == 4:
continue
if i < 4:
dpcd &= ~(0x3 << 4)
dpcd |= (i << 4)
for j in range(1, 6):
dpcd &= ~(0x7 << 1)
v = dpcd | (j << 1)
self.__dpcd.write(0x120, v)
val = int.from_bytes(self.__dpcd.read(0x281, 2).data, 'little')
if j == 1:
result.uncorrectedBlockErrors[i] = val & 0x7FFF if val & 0x8000 else None
elif j == 2:
result.correctedBlockErrors[i] = val & 0x7FFF if val & 0x8000 else None
elif j == 3:
result.bitErrors[i] = val & 0x7FFF if val & 0x8000 else None
elif j == 4:
result.parityBlockErrors[i] = val & 0x7FFF if val & 0x8000 else None
elif j == 5:
result.parityBitErrors[i] = val & 0x7FFF if val & 0x8000 else None
return result
def clear(self):
"""
Clear all errors.
"""
address = 0x120
dpcd = self.__dpcd.read(address, 1).data[0]
fec_ready = dpcd & ~0xE
dpcd = fec_ready
self.__dpcd.write(address, dpcd)