"""Port Commands - Ultra Ethernet"""
from __future__ import annotations
from dataclasses import dataclass
import ipaddress
import typing
import functools
from xoa_driver.internals.core.builders import (
build_get_request,
build_set_request
)
from xoa_driver.internals.core import interfaces
from xoa_driver.internals.core.token import Token
from xoa_driver.internals.core.transporter.registry import register_command
from xoa_driver.internals.core.transporter.protocol.payload import (
field,
RequestBodyStruct,
ResponseBodyStruct,
XmpByte,
XmpHex,
XmpInt,
XmpIPv4Address,
XmpIPv6Address,
XmpLong,
XmpShort,
XmpMacAddress,
XmpSequence,
XmpStr,
Hex,
XmpJson,
)
from .enums import (
OnOff,
YesNo,
UecCtlosClearDirection,
UecLinkOptionLlr,
UecLlrBehavior,
UecLlrInitEchoMode,
UecLlrEchoCheck,
UecLlrTxErrType,
UecLlrTxErrPattern,
UecLlrTxFsmState,
UecLlrRxFsmState,
UecCbfcCreditLimitMethod,
UecCbfcVcType,
UecCbfcVcMapping,
UecCbfcInjectErrType,
UecCbfcInjectErrPattern,
UecCbfcClearDirection,
UecCbfcMode,
)
[docs]
@register_command
@dataclass
class P_UE_CTLOS_CLEAR:
"""
Clear UE CtlOS counters in the specified direction(s).
"""
code: typing.ClassVar[int] = 1020
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class SetDataAttr(RequestBodyStruct):
direction: UecCtlosClearDirection = field(XmpByte())
"""UE CtlOS clear direction."""
[docs]
def set(self, direction: UecCtlosClearDirection) -> Token[None]:
"""Set the UE CtlOS clear direction.
:param direction: the UE CtlOS clear direction
:type direction: UecCtlosClearDirection
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, direction=direction))
clear_none = functools.partialmethod(set, UecCtlosClearDirection.NONE)
"""Not clear UE CtlOS counters
"""
clear_rx = functools.partialmethod(set, UecCtlosClearDirection.RX)
"""Clear RX UE CtlOS counters
"""
clear_tx = functools.partialmethod(set, UecCtlosClearDirection.TX)
"""Clear TX UE CtlOS counters
"""
clear_all = functools.partialmethod(set, UecCtlosClearDirection.ALL)
"""Clear all UE CtlOS counters (RX and TX).
"""
[docs]
@register_command
@dataclass
class P_UE_CTLOS_RX_STATS:
"""
CtlOS Rx statistics, including counters of all UE CtlOS message type values.
"""
code: typing.ClassVar[int] = 1012
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
llr_init_cnt: int = field(XmpLong())
"""The number of LLR_INIT CtlOS received."""
llr_init_echo_cnt: int = field(XmpLong())
"""The number of LLR_INIT_ECHO CtlOS received."""
llr_ack_cnt: int = field(XmpLong())
"""The number of LLR_ACK CtlOS received."""
llr_nack_cnt: int = field(XmpLong())
"""The number of LLR_NACK CtlOS received."""
llr_init_seq: Hex = field(XmpHex(size=3))
"""The init_seq (20 bits) of the most recently received LLR_INIT CtlOS."""
llr_init_data: Hex = field(XmpHex(size=2))
"""The init_data (16 bits) of the most recently received LLR_INIT CtlOS."""
llr_init_echo_seq: Hex = field(XmpHex(size=3))
"""The init_seq (20 bits) of the most recently received LLR_INIT_ECHO CtlOS."""
llr_init_echo_data: Hex = field(XmpHex(size=2))
"""The init_data (16 bits) of the most recently received LLR_INIT_ECHO CtlOS."""
llr_init_echo_mismatch: int = field(XmpLong())
"""The number of Rx LLR_INIT_ECHO CtlOS with a sequence number or data that does not match the most recently transmitted LLR_INIT CtlOS."""
cbfc_cf_update_cnt: int = field(XmpLong())
"""The number of CF_UPDATE CtlOS received."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get UE CtlOS Rx statistics.
:return: the UE CtlOS Rx statistics
:rtype: P_UE_CTLOS_RX_STATS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_CTLOS_TX_STATS:
"""
CtlOS Tx statistics, including counters of all UE CtlOS message type values.
"""
code: typing.ClassVar[int] = 1011
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
llr_init_cnt: int = field(XmpLong())
"""The number of LLR_INIT CtlOS transmitted."""
llr_init_echo_cnt: int = field(XmpLong())
"""The number of LLR_INIT_ECHO CtlOS transmitted."""
llr_ack_cnt: int = field(XmpLong())
"""The number of LLR_ACK CtlOS transmitted."""
llr_nack_cnt: int = field(XmpLong())
"""The number of LLR_NACK CtlOS transmitted."""
llr_init_seq: Hex = field(XmpHex(size=3))
"""The LLR_INIT sequence number. 3 bytes in hex format."""
llr_init_data: Hex = field(XmpHex(size=2))
"""The LLR_INIT data. 2 bytes in hex format."""
llr_init_echo_seq: Hex = field(XmpHex(size=3))
"""The LLR_INIT_ECHO sequence number. 3 bytes in hex format."""
llr_init_echo_data: Hex = field(XmpHex(size=2))
"""The LLR_INIT_ECHO data. 2 bytes in hex format."""
cbfc_cf_update_cnt: int = field(XmpLong())
"""The number of CF_UPDATE CtlOS transmitted."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get UE CtlOS Tx statistics.
:return: the UE CtlOS Tx statistics
:rtype: P_UE_CTLOS_TX_STATS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_LINKNEG_OPTIONS:
"""
This command selects which LLDP agent to use for UE Link Negotiation Options, and configures the desired LLR-W (local) setting. After the set operation, the UE Link Negotiation Options TLV will be added to the LLDPDU (if it is not already present in the LLDPDU) or replace the existing one (if Options TLV was previously added in LLDP configurataion).
The UE Link Negotiation Options TLV is used to discover and configure advanced UE link capabilities. Currently, the Options TLV is used to advertise the following UEC Link features:
* Link Layer Retry (LLR): Provides link level re-transmissions for packets with errors.
.. note::
In Link Negotiation for LLR, the R/W access of the four flags are:
* LLR-W (local): RW
* LLR-W (remote): RO
* LLR-E (local): RO
* LLR-E (remote): RO
When Link Negotiation is enabled on a LLDP agent, the only configurable parameter is ``LLR-W (local)``. To read ``LLR-W (remote)``, use ``P_UE_LINKNEG_OPTIONS_STATUS``. To read ``LLR-E (local)`` and ``LLR-E (remote)``, use ``P_UE_LLR_MODE``
To disable Link Negotation for UE Link Options, send the command with no parameters. This will remove the Options TLV from the LLDPDU and disable Link Negotiation for UE Link Options.
"""
code: typing.ClassVar[int] = 1016
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
lldp_agent_index: int = field(XmpByte())
"""integer, The LLDP agent index to use for advertising the advanced UE link capabilities. An LLDP agent must be created and configured separately using the LLDP commands before it can be used with this command.
Values: 0 to 7.
"""
llr_wanted_local: UecLinkOptionLlr = field(XmpByte())
"""coded integer, Indicates whether the local port wants LLR.
"""
[docs]
class SetDataAttr(RequestBodyStruct):
lldp_agent_index: int = field(XmpByte())
"""integer, The LLDP agent index to use for advertising the advanced UE link capabilities. An LLDP agent must be created and configured separately using the LLDP commands before it can be used with this command.
Values: 0 to 7.
"""
llr_wanted_local: UecLinkOptionLlr = field(XmpByte())
"""coded integer, Indicates whether the local port wants LLR.
"""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the description of the port.
:return: the description of the port
:rtype: P_UE_LINKNEG_OPTIONS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, lldp_agent_index: int, llr_wanted_local: UecLinkOptionLlr) -> Token[None]:
"""Set the UE Link Negotiation Options.
:param lldp_agent_index: The LLDP agent index to use for advertising the advanced UE link capabilities. Values: 0 to 7.
:type lldp_agent_index: int
:param llr_wanted_local: Indicates whether the local port wants LLR.
:type llr_wanted_local: UecLinkOptionLlr
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, lldp_agent_index=lldp_agent_index, llr_wanted_local=llr_wanted_local))
[docs]
@register_command
@dataclass
class P_UE_LINKNEG_OPTIONS_STATUS:
"""
Read ``LLR-W (local)`` and ``LLR-W (remote)`` of the port, which indicates whether the local and remote ports want to use LLR. This command is for read-only purposes and does not change any configuration or behavior of the LLR.
"""
code: typing.ClassVar[int] = 1029
pushed: typing.ClassVar[bool] = True
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
llr_wanted_local: UecLinkOptionLlr = field(XmpByte())
"""coded integer, Indicates whether the local port wants LLR.
"""
llr_wanted_remote: UecLinkOptionLlr = field(XmpByte())
"""coded integer, Indicates whether the remote port wants LLR.
"""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the current UE link negotiation options status.
:return: the current UE link negotiation options status
:rtype: P_UE_LINKNEG_OPTIONS_STATUS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_LLR_MODE:
"""
Configures the mode of operation of the LLR. Only when both local and remote ports have LLR ``ON``, the LLR will be active. If either port has LLR disabled, the LLR will be inactive.
"""
code: typing.ClassVar[int] = 1013
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
llr_mode_local: OnOff = field(XmpByte())
"""coded integer, LLR mode of the local port."""
llr_mode_remote: OnOff = field(XmpByte())
"""coded integer, LLR mode of the remote port."""
[docs]
class SetDataAttr(RequestBodyStruct):
llr_mode_local: OnOff = field(XmpByte())
"""coded integer, LLR mode of the local port."""
llr_mode_remote: OnOff = field(XmpByte())
"""coded integer, LLR mode of the remote port."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the LLR mode of the local port.
:return: the LLR mode of the local port
:rtype: P_UE_LLR_MODE.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, llr_mode_local: OnOff, llr_mode_remote: OnOff) -> Token[None]:
"""Set the LLR mode of the local port.
:param llr_mode_local: the LLR mode of the local port
:type llr_mode_local: OnOff
:param llr_mode_remote: the LLR mode of the remote port
:type llr_mode_remote: OnOff
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, llr_mode_local=llr_mode_local, llr_mode_remote=llr_mode_remote))
set_on = functools.partialmethod(set, llr_mode_local=OnOff.ON, llr_mode_remote=OnOff.ON)
"""Set LLR local to ON, LLR remote to ON."""
set_off = functools.partialmethod(set, llr_mode_local=OnOff.OFF, llr_mode_remote=OnOff.OFF)
"""Set LLR local to OFF, LLR remote to OFF."""
[docs]
@register_command
@dataclass
class P_UE_LLR_RX_STATS:
"""
Get the LLR Rx link-layer traffic statistics of the port.
"""
code: typing.ClassVar[int] = 1024
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
rx_llr_eligible_good_fcs: int = field(XmpLong())
"""The number of LLR-eligible frames received with a good FCS."""
rx_llr_eligible_poisoned_fcs: int = field(XmpLong())
"""The number of LLR-eligible frames received with a poisoned FCS."""
rx_llr_eligible_bad_fcs: int = field(XmpLong())
"""The number of LLR-eligible frames received with a bad FCS."""
rx_llr_eligible_total: int = field(XmpLong())
"""The total number of LLR-eligible frames received."""
rx_llr_eligible_good_fcs_exp_seq: int = field(XmpLong())
"""The number of LLR-eligible frames received with a good FCS that had the expected sequence number."""
rx_llr_eligible_poisoned_fcs_exp_seq: int = field(XmpLong())
"""The number of LLR-eligible frames received with a poisoned FCS that had the expected sequence number."""
rx_llr_eligible_bad_fcs_exp_seq: int = field(XmpLong())
"""The number of LLR-eligible frames received with a bad FCS that had the expected sequence number."""
rx_llr_eligible_total_exp_seq: int = field(XmpLong())
"""The total number of LLR-eligible frames received with the expected sequence number (irrespective of FCS status)."""
rx_llr_eligible_missing_seq: int = field(XmpLong())
"""The number of LLR-eligible frames received that had a sequence number that indicated a missing LLR-eligible frame in the sequence (irrespective of FCS status)."""
rx_llr_eligible_duplicate_seq: int = field(XmpLong())
"""The number of LLR-eligible frames received that had a duplicate sequence number (irrespective of FCS status)."""
rx_llr_ineligible_good_fcs: int = field(XmpLong())
"""The number of LLR-ineligible frames received with a good FCS."""
rx_llr_ineligible_bad_fcs: int = field(XmpLong())
"""The number of LLR-ineligible frames received with a bad FCS."""
rx_llr_ineligible_total: int = field(XmpLong())
"""The total number of LLR-ineligible frames received."""
rx_replay_operations: int = field(XmpLong())
"""The number of replay operations detected."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the LLR Rx link-layer traffic statistics of the port.
:return: the LLR Rx link-layer traffic statistics
:rtype: P_UE_LLR_RX_STATS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_LLR_TX_STATS:
"""
Get the LLR Tx link-layer traffic statistics of the port.
"""
code: typing.ClassVar[int] = 1023
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
tx_llr_eligible_good_fcs: int = field(XmpLong())
"""The number of LLR-eligible frames transmitted with a good FCS."""
tx_llr_eligible_poisoned_fcs: int = field(XmpLong())
"""The number of LLR-eligible frames transmitted with a poisoned FCS."""
tx_llr_eligible_bad_fcs: int = field(XmpLong())
"""The number of LLR-eligible frames transmitted with a bad FCS."""
tx_llr_eligible_discard: int = field(XmpLong())
"""The number of LLR-eligible frames discarded by the LLR TX when the TX state machine is in the INIT state and the llr_init_behavior is set to DISCARD, or when the TX state machine is in the FLUSH state and the llr_flush_behavior is set to DISCARD."""
tx_llr_eligible_total: int = field(XmpLong())
"""The total number of LLR-eligible frames transmitted."""
tx_llr_ineligible_good_fcs: int = field(XmpLong())
"""The number of LLR-ineligible frames transmitted with a good FCS."""
tx_llr_ineligible_bad_fcs: int = field(XmpLong())
"""The number of LLR-ineligible frames transmitted with a bad FCS."""
tx_llr_ineligible_total: int = field(XmpLong())
"""The total number of LLR-ineligible frames transmitted."""
tx_replayed: int = field(XmpLong())
"""The number of frames transmitted by the LLR TX that were replayed from the LLR replay buffer."""
tx_llr_eligible_good_fcs_good_seq: int = field(XmpLong())
"""The number of LLR-eligible frames transmitted with a good FCS and a good sequence number."""
tx_llr_eligible_poisoned_fcs_good_seq: int = field(XmpLong())
"""The number of LLR-eligible frames transmitted with a poisoned FCS and a good sequence number."""
tx_llr_eligible_bad_fcs_good_seq: int = field(XmpLong())
"""The number of LLR-eligible frames transmitted with a bad FCS and a good sequence number."""
tx_llr_eligible_missing_seq: int = field(XmpLong())
"""The number of LLR-eligible frames that with missing sequence number (irrespective of the FCS)."""
tx_llr_eligible_duplicate_seq: int = field(XmpLong())
"""The number of LLR-eligible frames that with duplicate sequence number (irrespective of the FCS)."""
tx_replay_operations: int = field(XmpLong())
"""The number of times the LLR TX state machine has exited the REPLAY state (replay operations)."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the LLR Tx link-layer traffic statistics of the port.
:return: the LLR Tx link-layer traffic statistics
:rtype: P_UE_LLR_TX_STATS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_CTLOS_SPACING:
"""
Configures the CtlOS spacing parameters of the port.
Only ``min_spacing`` is configurable in this release. ``target_spacing``,
``min_sop_spacing`` and ``min_inframe_spacing`` must still be supplied, but they are
accepted without any range check, never reach the hardware, and always read back as
their fixed defaults 1600, 256 and 2048.
"""
code: typing.ClassVar[int] = 1010
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
target_spacing: int = field(XmpInt())
"""integer, the target number of bytes between two CtlOS. Not configurable in this release, always reads 1600."""
min_spacing: int = field(XmpInt())
"""integer, the minimum number of bytes between two CtlOS. Either 400 or 0 (CtlOS may appear back-to-back)."""
min_sop_spacing: int = field(XmpInt())
"""integer, the minimum number of bytes between the SOP and the first CtlOS in that frame. Not configurable in this release, always reads 256."""
min_inframe_spacing: int = field(XmpInt())
"""integer, the minimum number of bytes between two CtlOS within the same frame. Not configurable in this release, always reads 2048."""
[docs]
class SetDataAttr(RequestBodyStruct):
target_spacing: int = field(XmpInt())
"""integer, the target number of bytes between two CtlOS. Not configurable in this release, accepted and ignored."""
min_spacing: int = field(XmpInt())
"""integer, the minimum number of bytes between two CtlOS. Must be either 400 or 0 (CtlOS may appear back-to-back); any other value is rejected."""
min_sop_spacing: int = field(XmpInt())
"""integer, the minimum number of bytes between the SOP and the first CtlOS in that frame. Not configurable in this release, accepted and ignored."""
min_inframe_spacing: int = field(XmpInt())
"""integer, the minimum number of bytes between two CtlOS within the same frame. Not configurable in this release, accepted and ignored."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CtlOS spacing parameters of the port.
:return: the CtlOS spacing parameters of the port
:rtype: P_UE_CTLOS_SPACING.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, target_spacing: int, min_spacing: int, min_sop_spacing: int, min_inframe_spacing: int) -> Token[None]:
"""Set the CtlOS spacing parameters of the port.
:param target_spacing: the target number of bytes between two CtlOS, not configurable in this release, accepted and ignored
:type target_spacing: int
:param min_spacing: the minimum number of bytes between two CtlOS, either 400 or 0 (CtlOS may appear back-to-back)
:type min_spacing: int
:param min_sop_spacing: the minimum number of bytes between the SOP and the first CtlOS in that frame, not configurable in this release, accepted and ignored
:type min_sop_spacing: int
:param min_inframe_spacing: the minimum number of bytes between two CtlOS within the same frame, not configurable in this release, accepted and ignored
:type min_inframe_spacing: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, target_spacing=target_spacing, min_spacing=min_spacing, min_sop_spacing=min_sop_spacing, min_inframe_spacing=min_inframe_spacing))
[docs]
@register_command
@dataclass
class P_UE_LLR_REPLAY:
"""
Configures the LLR replay parameters of the port.
"""
code: typing.ClassVar[int] = 1014
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
outstanding_seq_max: int = field(XmpInt())
"""integer, the maximum number of outstanding sequence numbers."""
outstanding_data_max: int = field(XmpInt())
"""integer, the maximum amount of outstanding data."""
replay_timer_max: int = field(XmpInt())
"""integer, the maximum value of the replay timer."""
replay_ct_max: int = field(XmpInt())
"""integer, the maximum replay count."""
pcs_lost_status_timer_max: int = field(XmpLong())
"""long integer, the maximum value of the PCS lost status timer."""
data_age_timer_max: int = field(XmpLong())
"""long integer, the maximum value of the data age timer."""
[docs]
class SetDataAttr(RequestBodyStruct):
outstanding_seq_max: int = field(XmpInt())
"""integer, the maximum number of outstanding sequence numbers."""
outstanding_data_max: int = field(XmpInt())
"""integer, the maximum amount of outstanding data."""
replay_timer_max: int = field(XmpInt())
"""integer, the maximum value of the replay timer."""
replay_ct_max: int = field(XmpInt())
"""integer, the maximum replay count."""
pcs_lost_status_timer_max: int = field(XmpLong())
"""long integer, the maximum value of the PCS lost status timer."""
data_age_timer_max: int = field(XmpLong())
"""long integer, the maximum value of the data age timer."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the LLR replay parameters of the port.
:return: the LLR replay parameters of the port
:rtype: P_UE_LLR_REPLAY.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, outstanding_seq_max: int, outstanding_data_max: int, replay_timer_max: int, replay_ct_max: int, pcs_lost_status_timer_max: int, data_age_timer_max: int) -> Token[None]:
"""Set the LLR replay parameters of the port.
:param outstanding_seq_max: the maximum number of outstanding sequence numbers
:type outstanding_seq_max: int
:param outstanding_data_max: the maximum amount of outstanding data
:type outstanding_data_max: int
:param replay_timer_max: the maximum value of the replay timer
:type replay_timer_max: int
:param replay_ct_max: the maximum replay count
:type replay_ct_max: int
:param pcs_lost_status_timer_max: the maximum value of the PCS lost status timer
:type pcs_lost_status_timer_max: int
:param data_age_timer_max: the maximum value of the data age timer
:type data_age_timer_max: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, outstanding_seq_max=outstanding_seq_max, outstanding_data_max=outstanding_data_max, replay_timer_max=replay_timer_max, replay_ct_max=replay_ct_max, pcs_lost_status_timer_max=pcs_lost_status_timer_max, data_age_timer_max=data_age_timer_max))
[docs]
@register_command
@dataclass
class P_UE_LLR_BEHAVIOR:
"""
Configures the LLR behavior of the port for the INIT and FLUSH states.
"""
code: typing.ClassVar[int] = 1015
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
llr_init_behavior: UecLlrBehavior = field(XmpByte())
"""coded byte, the LLR behavior for LLR-eligible frames when the TX state machine is in the INIT state."""
llr_flush_behavior: UecLlrBehavior = field(XmpByte())
"""coded byte, the LLR behavior for LLR-eligible frames when the TX state machine is in the FLUSH state."""
re_init_on_discard: YesNo = field(XmpByte())
"""coded byte, whether to re-initialize the LLR on discard."""
[docs]
class SetDataAttr(RequestBodyStruct):
llr_init_behavior: UecLlrBehavior = field(XmpByte())
"""coded byte, the LLR behavior for LLR-eligible frames when the TX state machine is in the INIT state."""
llr_flush_behavior: UecLlrBehavior = field(XmpByte())
"""coded byte, the LLR behavior for LLR-eligible frames when the TX state machine is in the FLUSH state."""
re_init_on_discard: YesNo = field(XmpByte())
"""coded byte, whether to re-initialize the LLR on discard."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the LLR behavior of the port.
:return: the LLR behavior of the port
:rtype: P_UE_LLR_BEHAVIOR.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, llr_init_behavior: UecLlrBehavior, llr_flush_behavior: UecLlrBehavior, re_init_on_discard: YesNo) -> Token[None]:
"""Set the LLR behavior of the port.
:param llr_init_behavior: the LLR behavior for LLR-eligible frames when the TX state machine is in the INIT state
:type llr_init_behavior: UecLlrBehavior
:param llr_flush_behavior: the LLR behavior for LLR-eligible frames when the TX state machine is in the FLUSH state
:type llr_flush_behavior: UecLlrBehavior
:param re_init_on_discard: whether to re-initialize the LLR on discard
:type re_init_on_discard: YesNo
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, llr_init_behavior=llr_init_behavior, llr_flush_behavior=llr_flush_behavior, re_init_on_discard=re_init_on_discard))
[docs]
@register_command
@dataclass
class P_UE_LLR_INIT:
"""
Configures the LLR INIT parameters of the port.
"""
code: typing.ClassVar[int] = 1017
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
init_seq: Hex = field(XmpHex(size=3))
"""3 bytes in hex format, the 24-bit LLR_INIT sequence number."""
init_data: Hex = field(XmpHex(size=2))
"""2 bytes in hex format, the 16-bit LLR_INIT data."""
min_spacing_multiplier: int = field(XmpInt())
"""integer, the minimum multiplier of the number of bytes between transmission of successive LLR_INIT CtlOS. Range 4 to 256."""
[docs]
class SetDataAttr(RequestBodyStruct):
init_seq: Hex = field(XmpHex(size=3))
"""3 bytes in hex format, the 24-bit LLR_INIT sequence number."""
init_data: Hex = field(XmpHex(size=2))
"""2 bytes in hex format, the 16-bit LLR_INIT data."""
min_spacing_multiplier: int = field(XmpInt())
"""integer, the minimum multiplier of the number of bytes between transmission of successive LLR_INIT CtlOS. Range 4 to 256."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the LLR INIT parameters of the port.
:return: the LLR INIT parameters of the port
:rtype: P_UE_LLR_INIT.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, init_seq: Hex, init_data: Hex, min_spacing_multiplier: int) -> Token[None]:
"""Set the LLR INIT parameters of the port.
:param init_seq: the 24-bit LLR_INIT sequence number (3 bytes)
:type init_seq: Hex
:param init_data: the 16-bit LLR_INIT data (2 bytes)
:type init_data: Hex
:param min_spacing_multiplier: the minimum spacing multiplier
:type min_spacing_multiplier: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, init_seq=init_seq, init_data=init_data, min_spacing_multiplier=min_spacing_multiplier))
[docs]
@register_command
@dataclass
class P_UE_LLR_INIT_ECHO:
"""
Configures the LLR INIT_ECHO parameters of the port.
"""
code: typing.ClassVar[int] = 1018
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
init_seq_mode: UecLlrInitEchoMode = field(XmpByte())
"""coded byte, the source mode of the INIT_ECHO sequence number."""
init_seq: Hex = field(XmpHex(size=3))
"""3 bytes in hex format, the 24-bit INIT_ECHO sequence number when manual mode is used."""
init_data_mode: UecLlrInitEchoMode = field(XmpByte())
"""coded byte, the source mode of the INIT_ECHO data."""
init_data: Hex = field(XmpHex(size=2))
"""2 bytes in hex format, the 16-bit INIT_ECHO data when manual mode is used."""
[docs]
class SetDataAttr(RequestBodyStruct):
init_seq_mode: UecLlrInitEchoMode = field(XmpByte())
"""coded byte, the source mode of the INIT_ECHO sequence number."""
init_seq: Hex = field(XmpHex(size=3))
"""3 bytes in hex format, the 24-bit INIT_ECHO sequence number when manual mode is used."""
init_data_mode: UecLlrInitEchoMode = field(XmpByte())
"""coded byte, the source mode of the INIT_ECHO data."""
init_data: Hex = field(XmpHex(size=2))
"""2 bytes in hex format, the 16-bit INIT_ECHO data when manual mode is used."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the LLR INIT_ECHO parameters of the port.
:return: the LLR INIT_ECHO parameters of the port
:rtype: P_UE_LLR_INIT_ECHO.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, init_seq_mode: UecLlrInitEchoMode, init_seq: Hex, init_data_mode: UecLlrInitEchoMode, init_data: Hex) -> Token[None]:
"""Set the LLR INIT_ECHO parameters of the port.
:param init_seq_mode: the source mode of the INIT_ECHO sequence number
:type init_seq_mode: UecLlrInitEchoMode
:param init_seq: the 24-bit INIT_ECHO sequence number when manual mode is used (3 bytes)
:type init_seq: Hex
:param init_data_mode: the source mode of the INIT_ECHO data
:type init_data_mode: UecLlrInitEchoMode
:param init_data: the 16-bit INIT_ECHO data when manual mode is used (2 bytes)
:type init_data: Hex
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, init_seq_mode=init_seq_mode, init_seq=init_seq, init_data_mode=init_data_mode, init_data=init_data))
[docs]
@register_command
@dataclass
class P_UE_LLR_ACKNACK:
"""
Configures the LLR ACK/NACK spacing parameters of the port.
"""
code: typing.ClassVar[int] = 1019
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
target_spacing: int = field(XmpInt())
"""integer, the target ACK/NACK spacing."""
min_spacing: int = field(XmpInt())
"""integer, the minimum ACK/NACK spacing."""
max_spacing: int = field(XmpInt())
"""integer, the maximum ACK/NACK spacing."""
[docs]
class SetDataAttr(RequestBodyStruct):
target_spacing: int = field(XmpInt())
"""integer, the target ACK/NACK spacing."""
min_spacing: int = field(XmpInt())
"""integer, the minimum ACK/NACK spacing."""
max_spacing: int = field(XmpInt())
"""integer, the maximum ACK/NACK spacing."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the LLR ACK/NACK spacing parameters of the port.
:return: the LLR ACK/NACK spacing parameters of the port
:rtype: P_UE_LLR_ACKNACK.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, target_spacing: int, min_spacing: int, max_spacing: int) -> Token[None]:
"""Set the LLR ACK/NACK spacing parameters of the port.
:param target_spacing: the target ACK/NACK spacing
:type target_spacing: int
:param min_spacing: the minimum ACK/NACK spacing
:type min_spacing: int
:param max_spacing: the maximum ACK/NACK spacing
:type max_spacing: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, target_spacing=target_spacing, min_spacing=min_spacing, max_spacing=max_spacing))
[docs]
@register_command
@dataclass
class P_UE_LLR_INJECT_ERR:
"""
LLR TX error injection of the port.
"""
code: typing.ClassVar[int] = 1025
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class SetDataAttr(RequestBodyStruct):
error_type: UecLlrTxErrType = field(XmpByte())
"""coded byte, the type of error to inject."""
pattern: UecLlrTxErrPattern = field(XmpByte())
"""coded byte, the pattern of error injection."""
burst_size: int = field(XmpInt())
"""integer, the number of errors in a burst."""
burst_interval: int = field(XmpInt())
"""integer, the interval between bursts."""
[docs]
def set(self, error_type: UecLlrTxErrType, pattern: UecLlrTxErrPattern, burst_size: int, burst_interval: int) -> Token[None]:
"""Set the LLR TX error injection configuration of the port.
:param error_type: the type of error to inject
:type error_type: UecLlrTxErrType
:param pattern: the pattern of error injection
:type pattern: UecLlrTxErrPattern
:param burst_size: the number of errors in a burst
:type burst_size: int
:param burst_interval: the interval between bursts
:type burst_interval: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, error_type=error_type, pattern=pattern, burst_size=burst_size, burst_interval=burst_interval))
inject_seq_drop = functools.partialmethod(set, UecLlrTxErrType.SEQ_DROP, UecLlrTxErrPattern.ONCE)
"""Inject a single sequence drop error.
"""
inject_seq_duplicate = functools.partialmethod(set, UecLlrTxErrType.SEQ_DUP, UecLlrTxErrPattern.ONCE)
"""Inject a single sequence duplicate error.
"""
inject_fcs_bad = functools.partialmethod(set, UecLlrTxErrType.FCS_BAD, UecLlrTxErrPattern.ONCE)
"""Inject a single FCS bad error.
"""
inject_fcs_poison = functools.partialmethod(set, UecLlrTxErrType.FCS_POISONED, UecLlrTxErrPattern.ONCE)
"""Inject a single FCS poison error.
"""
inject_llr_ack_drop = functools.partialmethod(set, UecLlrTxErrType.LLR_ACK_DROP, UecLlrTxErrPattern.ONCE)
"""Inject a single LLR ACK drop error.
"""
inject_llr_nack_drop = functools.partialmethod(set, UecLlrTxErrType.LLR_NACK_DROP, UecLlrTxErrPattern.ONCE)
"""Inject a single LLR NACK drop error.
"""
[docs]
@register_command
@dataclass
class P_UE_LLR_POISONFCS:
"""
Configures the poisoned FCS pattern used by the LLR TX error injection of the port.
"""
code: typing.ClassVar[int] = 1028
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
pattern: Hex = field(XmpHex(size=4))
"""4 bytes in hex format, the 32-bit poison FCS pattern."""
[docs]
class SetDataAttr(RequestBodyStruct):
pattern: Hex = field(XmpHex(size=4))
"""4 bytes in hex format, the 32-bit poison FCS pattern."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the poison FCS pattern of the port.
:return: the poison FCS pattern of the port
:rtype: P_UE_LLR_TXERR_POISONFCS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, pattern: Hex) -> Token[None]:
"""Set the poison FCS pattern of the port.
:param pattern: the 32-bit poison FCS pattern (4 bytes)
:type pattern: Hex
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, pattern=pattern))
set_aaaaaaaa = functools.partialmethod(set, Hex("AAAAAAAA"))
"""Set the poisoned FCS pattern of the port to 0xAAAAAAAA.
"""
set_55555555 = functools.partialmethod(set, Hex("55555555"))
"""Set the poisoned FCS pattern of the port to 0x55555555.
"""
[docs]
@register_command
@dataclass
class P_UE_LLR_INIT_ECHO_CHK:
"""
Configures the validation of the LLR INIT_ECHO sequence number and data of the port.
"""
code: typing.ClassVar[int] = 1031
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
validate_init_seq: UecLlrEchoCheck = field(XmpByte())
"""coded byte, whether to validate the INIT_ECHO sequence number."""
validate_init_data: UecLlrEchoCheck = field(XmpByte())
"""coded byte, whether to validate the INIT_ECHO data."""
[docs]
class SetDataAttr(RequestBodyStruct):
validate_init_seq: UecLlrEchoCheck = field(XmpByte())
"""coded byte, whether to validate the INIT_ECHO sequence number."""
validate_init_data: UecLlrEchoCheck = field(XmpByte())
"""coded byte, whether to validate the INIT_ECHO data."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the LLR INIT_ECHO validation configuration of the port.
:return: the LLR INIT_ECHO validation configuration of the port
:rtype: P_UE_LLR_INIT_ECHO_CHK.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, validate_init_seq: UecLlrEchoCheck, validate_init_data: UecLlrEchoCheck) -> Token[None]:
"""Set the LLR INIT_ECHO validation configuration of the port.
:param validate_init_seq: whether to validate the INIT_ECHO sequence number
:type validate_init_seq: UecLlrEchoCheck
:param validate_init_data: whether to validate the INIT_ECHO data
:type validate_init_data: UecLlrEchoCheck
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, validate_init_seq=validate_init_seq, validate_init_data=validate_init_data))
validate_init_seq = functools.partialmethod(set, UecLlrEchoCheck.ENABLE, UecLlrEchoCheck.DISABLE)
"""Only validate the init_seq of the INIT_ECHO.
"""
validate_init_data = functools.partialmethod(set, UecLlrEchoCheck.DISABLE, UecLlrEchoCheck.ENABLE)
"""Only validate the init_data of the INIT_ECHO.
"""
validate_both = functools.partialmethod(set, UecLlrEchoCheck.ENABLE, UecLlrEchoCheck.ENABLE)
"""Validate both the init_seq and init_data of the INIT_ECHO.
"""
validate_none = functools.partialmethod(set, UecLlrEchoCheck.DISABLE, UecLlrEchoCheck.DISABLE)
"""Do not validate the init_seq nor the init_data of the INIT_ECHO.
"""
[docs]
@register_command
@dataclass
class P_UE_LLR_TXFSM_STATE:
"""
Return the current state of the LLR Tx FSM that the port is in, along with the current values of the relevant timers and counters for that state. The possible states are: LLR_OFF, INIT, ADVANCE, REPLAY, and FLUSH. The specific timers and counters returned depend on the current state of the FSM.
"""
code: typing.ClassVar[int] = 1021
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
txfsm_state: UecLlrTxFsmState = field(XmpByte())
"""coded byte, the current state of the LLR Tx FSM that the port is in."""
ns_in_llr_off: int = field(XmpLong())
"""long integer, the number of nanoseconds the FSM has been in the LLR_OFF state since the last time it was cleared."""
ns_in_init: int = field(XmpLong())
"""long integer, the number of nanoseconds the FSM has been in the INIT state since the last time it was cleared."""
ns_in_advance: int = field(XmpLong())
"""long integer, the number of nanoseconds the FSM has been in the ADVANCE state since the last time it was cleared."""
ns_in_replay: int = field(XmpLong())
"""long integer, the number of nanoseconds the FSM has been in the REPLAY state since the last time it was cleared."""
ns_in_flush: int = field(XmpLong())
"""long integer, the number of nanoseconds the FSM has been in the FLUSH state since the last time it was cleared."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the state of the LLR TX finite state machine of the port.
:return: the state of the LLR TX finite state machine of the port
:rtype: P_UE_LLR_TXFSM_STATE.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_LLR_RXFSM_STATE:
"""
Return the current state of the LLR Rx FSM that the port is in, along with the current values of the relevant timers and counters for that state. The possible states are: OFF, SEND_ACKS, SEND_NACK, and NACK_SENT. The specific timers and counters returned depend on the current state of the FSM.
"""
code: typing.ClassVar[int] = 1022
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
rxfsm_state: UecLlrRxFsmState = field(XmpByte())
"""coded byte, the current state of the LLR Rx FSM that the port is in."""
ns_in_off: int = field(XmpLong())
"""long integer, the number of nanoseconds the FSM has been in the OFF state since the last time it was cleared."""
ns_in_send_acks: int = field(XmpLong())
"""long integer, the number of nanoseconds the FSM has been in the SEND_ACKS state since the last time it was cleared."""
ns_in_send_nack: int = field(XmpLong())
"""long integer, the number of nanoseconds the FSM has been in the SEND_NACK state since the last time it was cleared."""
ns_in_nack_sent: int = field(XmpLong())
"""long integer, the number of nanoseconds the FSM has been in the NACK_SENT state since the last time it was cleared."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the state of the LLR RX finite state machine of the port.
:return: the state of the LLR RX finite state machine of the port
:rtype: P_UE_LLR_RXFSM_STATE.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_LLR_STATUS:
"""
Get the active status of the LLR of the port.
"""
code: typing.ClassVar[int] = 1030
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
status: YesNo = field(XmpByte())
"""coded byte, whether the LLR is active."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the active status of the LLR of the port.
:return: the active status of the LLR of the port
:rtype: P_UE_LLR_STATUS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_CTLOS_RX_ERRORS:
"""
Get the CtlOS Rx error counters of the port.
"""
code: typing.ClassVar[int] = 1034
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
ctlos_msg_type_errors: int = field(XmpLong())
"""long integer, the number of CtlOS message type errors."""
ctlos_padding_errors: int = field(XmpLong())
"""long integer, the number of CtlOS padding errors."""
llr_init_echo_invalid_init_seq: int = field(XmpLong())
"""long integer, the number of LLR_INIT_ECHO with an invalid INIT sequence number."""
llr_init_echo_invalid_init_data: int = field(XmpLong())
"""long integer, the number of LLR_INIT_ECHO with invalid INIT data."""
llr_init_interval_low: int = field(XmpLong())
"""long integer, the number of LLR_INIT received with an interval that is too low."""
llr_ack_nack_interval_high: int = field(XmpLong())
"""long integer, the number of LLR_ACK/NACK received with an interval that is too high."""
llr_ack_nack_interval_low: int = field(XmpLong())
"""long integer, the number of LLR_ACK/NACK received with an interval that is too low."""
llr_ctlos_padding_error: int = field(XmpLong())
"""long integer, the number of LLR CtlOS padding errors."""
llr_ack_nack_seq_error_cnt: int = field(XmpLong())
"""long integer, the number of LLR_ACK/NACK received with a sequence error."""
cf_update_interval_high: int = field(XmpLong())
"""long integer, the number of CF_UPDATE received with an interval that is too high."""
cf_update_interval_low: int = field(XmpLong())
"""long integer, the number of CF_UPDATE received with an interval that is too low."""
cf_update_padding_error: int = field(XmpLong())
"""long integer, the number of CF_UPDATE received with a padding error."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CtlOS Rx error counters of the port.
:return: the CtlOS Rx error counters of the port
:rtype: P_UE_CTLOS_RX_ERRORS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_CBFC_CCUPDATE_HDR:
"""
The header fields of the CBFC CC_Update message generated by the port.
"""
code: typing.ClassVar[int] = 1035
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
dmac: Hex = field(XmpMacAddress())
"""six hex bytes, the destination MAC address of the CC_Update message.
Destination MAC address for the CC_Update message. Default set to "0180C2000001" or the 48-bit individual address of the destination station.
"""
smac: Hex = field(XmpMacAddress())
"""six hex bytes, the source MAC address of the CC_Update message."""
ethertype: Hex = field(XmpHex(size=2))
"""two hex bytes, the EtherType of the CC_Update message.
Default set to "88B5".
"""
opcode: Hex = field(XmpHex(size=2))
"""two hex bytes, the opcode of the CC_Update message.
Default set to "FFFE".
"""
cid: Hex = field(XmpHex(size=3))
"""three hex bytes, the Company ID of the CC_Update message.
Default set to "FA7ACB".
"""
[docs]
class SetDataAttr(RequestBodyStruct):
dmac: Hex = field(XmpMacAddress())
"""six hex bytes, the destination MAC address of the CC_Update message.
Destination MAC address for the CC_Update message. Default set to "0180C2000001" or the 48-bit individual address of the destination station.
"""
smac: Hex = field(XmpMacAddress())
"""six hex bytes, the source MAC address of the CC_Update message."""
ethertype: Hex = field(XmpHex(size=2))
"""two hex bytes, the EtherType of the CC_Update message.
Default set to "88B5".
"""
opcode: Hex = field(XmpHex(size=2))
"""two hex bytes, the opcode of the CC_Update message.
Default set to "FFFE".
"""
cid: Hex = field(XmpHex(size=3))
"""three hex bytes, the Company ID of the CC_Update message.
Default set to "FA7ACB".
"""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CBFC CC_Update message header of the port.
:return: the CBFC CC_Update message header of the port
:rtype: P_UE_CBFC_CCUPDATE_HDR.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, dmac: Hex, smac: Hex, ethertype: Hex, opcode: Hex, cid: Hex) -> Token[None]:
"""Set the CBFC CC_Update message header of the port.
:param dmac: the destination MAC address of the CC_Update message, default set to "0180C2000001"
:type dmac: Hex
:param smac: the source MAC address of the CC_Update message, default set to port MAC address.
:type smac: Hex
:param ethertype: the EtherType of the CC_Update message, default set to "88B5"
:type ethertype: Hex
:param opcode: the opcode of the CC_Update message, default set to "FFFE"
:type opcode: Hex
:param cid: the Company ID of the CC_Update message, default set to "FA7ACB"
:type cid: Hex
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, dmac=dmac, smac=smac, ethertype=ethertype, opcode=opcode, cid=cid))
[docs]
@register_command
@dataclass
class P_UE_CBFC_CCUPDATE_TIMER:
"""
The period at which the port generates CBFC CC_Update messages.
"""
code: typing.ClassVar[int] = 1036
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
cc_msg_timer: int = field(XmpInt())
"""integer, the CC_Update generation period in microseconds."""
[docs]
class SetDataAttr(RequestBodyStruct):
cc_msg_timer: int = field(XmpInt())
"""integer, the CC_Update generation period in microseconds."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CBFC CC_Update generation period of the port.
:return: the CBFC CC_Update generation period of the port
:rtype: P_UE_CBFC_CCUPDATE_TIMER.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, cc_msg_timer: int) -> Token[None]:
"""Set the CBFC CC_Update generation period of the port.
:param cc_msg_timer: the CC_Update generation period in microseconds
:type cc_msg_timer: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, cc_msg_timer=cc_msg_timer))
[docs]
@register_command
@dataclass
class P_UE_CBFC_CFUPDATE_TIMER:
"""
The interval bounds at which the port generates CBFC CF_Update CtlOS.
"""
code: typing.ClassVar[int] = 1037
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
cf_min_timer: int = field(XmpInt())
"""integer, the minimum number of bytes between two CF_Update CtlOS."""
cf_max_timer: int = field(XmpInt())
"""integer, the maximum number of bytes between two CF_Update CtlOS."""
[docs]
class SetDataAttr(RequestBodyStruct):
cf_min_timer: int = field(XmpInt())
"""integer, the minimum number of bytes between two CF_Update CtlOS."""
cf_max_timer: int = field(XmpInt())
"""integer, the maximum number of bytes between two CF_Update CtlOS."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CBFC CF_Update timers of the port.
:return: the CBFC CF_Update timers of the port
:rtype: P_UE_CBFC_CFUPDATE_TIMER.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, cf_min_timer: int, cf_max_timer: int) -> Token[None]:
"""Set the CBFC CF_Update timers of the port.
:param cf_min_timer: the minimum number of bytes between two CF_Update CtlOS. Default: 800. Allowed range: 800 bytes to 16384 bytes. Resolution: 8 bytes.
:type cf_min_timer: int
:param cf_max_timer: the maximum number of bytes between two CF_Update CtlOS. Default: 16384. Allowed range: 16384 bytes to 1048576 bytes with granularity of 16384 bytes (16 KB to 1 MB with granularity of 16 KB). Resolution: 16384 bytes.
:type cf_max_timer: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, cf_min_timer=cf_min_timer, cf_max_timer=cf_max_timer))
[docs]
@register_command
@dataclass
class P_UE_CBFC_NUMVCS:
"""
The number of CBFC virtual channels in a CBFC configuration data storage.
Only the two local staged storages (0 = local receiver, 1 = local sender) are writable.
Reducing the number of virtual channels resets the dropped ones to their defaults, and on
the local sender storage it also removes the ``PS_UE_CBFC_VC`` mapping of every stream that
pointed at a dropped virtual channel.
"""
code: typing.ClassVar[int] = 1038
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
_storage_xindex: int
[docs]
class GetDataAttr(ResponseBodyStruct):
num_vcs: int = field(XmpInt())
"""integer, the number of virtual channels."""
[docs]
class SetDataAttr(RequestBodyStruct):
num_vcs: int = field(XmpInt())
"""integer, the number of virtual channels, in the range 1 to 32."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the number of CBFC virtual channels of the storage.
:return: the number of CBFC virtual channels of the storage
:rtype: P_UE_CBFC_NUMVCS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port, indices=[self._storage_xindex]))
[docs]
def set(self, num_vcs: int) -> Token[None]:
"""Set the number of CBFC virtual channels of the storage.
:param num_vcs: the number of virtual channels
:type num_vcs: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, num_vcs=num_vcs, indices=[self._storage_xindex]))
[docs]
@register_command
@dataclass
class P_UE_CBFC_LINK:
"""
The CBFC link parameters of a CBFC configuration data storage.
Only the two local staged storages (0 = local receiver, 1 = local sender) are writable.
``UecCbfcCreditLimitMethod.NONE`` is not supported by the hardware and is rejected.
"""
code: typing.ClassVar[int] = 1039
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
_storage_xindex: int
[docs]
class GetDataAttr(ResponseBodyStruct):
credit_size: int = field(XmpInt())
"""integer, the size of a credit in bytes."""
credit_limit_method: UecCbfcCreditLimitMethod = field(XmpByte())
"""coded byte, how the receive buffer is distributed across the virtual channels."""
total_credit_limit: int = field(XmpInt())
"""integer, the credit limit shared by all lossless virtual channels."""
packet_overhead: int = field(XmpInt())
"""integer, the per-packet overhead in bytes added when accounting credits."""
[docs]
class SetDataAttr(RequestBodyStruct):
credit_size: int = field(XmpInt())
"""integer, the size of a credit in bytes."""
credit_limit_method: UecCbfcCreditLimitMethod = field(XmpByte())
"""coded byte, how the receive buffer is distributed across the virtual channels."""
total_credit_limit: int = field(XmpInt())
"""integer, the credit limit shared by all lossless virtual channels, in the range 0 to 524287."""
packet_overhead: int = field(XmpInt())
"""integer, the per-packet overhead in bytes added when accounting credits, in the range -16 to 12."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CBFC link parameters of the storage.
:return: the CBFC link parameters of the storage
:rtype: P_UE_CBFC_LINK.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port, indices=[self._storage_xindex]))
[docs]
def set(self, credit_size: int, credit_limit_method: UecCbfcCreditLimitMethod, total_credit_limit: int, packet_overhead: int) -> Token[None]:
"""Set the CBFC link parameters of the storage.
:param credit_size: the size of a credit in bytes
:type credit_size: int
:param credit_limit_method: how the receive buffer is distributed across the virtual channels
:type credit_limit_method: UecCbfcCreditLimitMethod
:param total_credit_limit: the credit limit shared by all lossless virtual channels
:type total_credit_limit: int
:param packet_overhead: the per-packet overhead in bytes added when accounting credits
:type packet_overhead: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, credit_size=credit_size, credit_limit_method=credit_limit_method, total_credit_limit=total_credit_limit, packet_overhead=packet_overhead, indices=[self._storage_xindex]))
[docs]
@register_command
@dataclass
class P_UE_CBFC_VC_TYPE:
"""
Whether a CBFC virtual channel is lossless or best-effort.
Only the two local staged storages (0 = local receiver, 1 = local sender) are writable.
"""
code: typing.ClassVar[int] = 1040
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
_vc_xindex: int
_storage_xindex: int
[docs]
class GetDataAttr(ResponseBodyStruct):
vc_type: UecCbfcVcType = field(XmpByte())
"""coded byte, whether the virtual channel is lossless."""
[docs]
class SetDataAttr(RequestBodyStruct):
vc_type: UecCbfcVcType = field(XmpByte())
"""coded byte, whether the virtual channel is lossless."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the type of the CBFC virtual channel.
:return: the type of the CBFC virtual channel
:rtype: P_UE_CBFC_VC_TYPE.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port, indices=[self._vc_xindex, self._storage_xindex]))
[docs]
def set(self, vc_type: UecCbfcVcType) -> Token[None]:
"""Set the type of the CBFC virtual channel.
:param vc_type: whether the virtual channel is lossless
:type vc_type: UecCbfcVcType
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, vc_type=vc_type, indices=[self._vc_xindex, self._storage_xindex]))
set_best_effort = functools.partialmethod(set, UecCbfcVcType.NO)
"""Make the virtual channel best-effort.
"""
set_lossless = functools.partialmethod(set, UecCbfcVcType.YES)
"""Make the virtual channel lossless.
"""
[docs]
@register_command
@dataclass
class P_UE_CBFC_VC:
"""
The credit limit and the packet classification of a CBFC virtual channel.
Only the two local staged storages (0 = local receiver, 1 = local sender) are writable.
``UecCbfcVcMapping.HANDLE`` is not supported by the classifier and is rejected, so ``handle``
is range checked but never programmed.
"""
code: typing.ClassVar[int] = 1041
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
_vc_xindex: int
_storage_xindex: int
[docs]
class GetDataAttr(ResponseBodyStruct):
credit_limit: int = field(XmpInt())
"""integer, the credit limit of the virtual channel."""
mapping: UecCbfcVcMapping = field(XmpByte())
"""coded byte, the packet field used to map a packet to the virtual channel."""
mask_pcp_dei: Hex = field(XmpHex(size=1))
"""one hex byte, the mask applied to the VLAN PCP/DEI field."""
value_pcp_dei: Hex = field(XmpHex(size=1))
"""one hex byte, the value the masked VLAN PCP/DEI field is matched against."""
mask_dscp: Hex = field(XmpHex(size=1))
"""one hex byte, the mask applied to the IP DSCP field."""
value_dscp: Hex = field(XmpHex(size=1))
"""one hex byte, the value the masked IP DSCP field is matched against."""
handle: int = field(XmpInt())
"""integer, the UE packet handle mapped to the virtual channel."""
[docs]
class SetDataAttr(RequestBodyStruct):
credit_limit: int = field(XmpInt())
"""integer, the credit limit of the virtual channel, in the range 0 to 524287. Ignored by a best-effort virtual channel."""
mapping: UecCbfcVcMapping = field(XmpByte())
"""coded byte, the packet field used to map a packet to the virtual channel. HANDLE is not supported in this release."""
mask_pcp_dei: Hex = field(XmpHex(size=1))
"""one hex byte, the mask applied to the VLAN PCP/DEI field. At most 0x0F."""
value_pcp_dei: Hex = field(XmpHex(size=1))
"""one hex byte, the value the masked VLAN PCP/DEI field is matched against. At most 0x0F."""
mask_dscp: Hex = field(XmpHex(size=1))
"""one hex byte, the mask applied to the IP DSCP field. At most 0x3F."""
value_dscp: Hex = field(XmpHex(size=1))
"""one hex byte, the value the masked IP DSCP field is matched against. At most 0x3F."""
handle: int = field(XmpInt())
"""integer, the UE packet handle mapped to the virtual channel, at most 4095. Not programmed in this release."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the configuration of the CBFC virtual channel.
:return: the configuration of the CBFC virtual channel
:rtype: P_UE_CBFC_VC.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port, indices=[self._vc_xindex, self._storage_xindex]))
[docs]
def set(self, credit_limit: int, mapping: UecCbfcVcMapping, mask_pcp_dei: Hex, value_pcp_dei: Hex, mask_dscp: Hex, value_dscp: Hex, handle: int) -> Token[None]:
"""Set the configuration of the CBFC virtual channel.
:param credit_limit: the credit limit of the virtual channel
:type credit_limit: int
:param mapping: the packet field used to map a packet to the virtual channel
:type mapping: UecCbfcVcMapping
:param mask_pcp_dei: the mask applied to the VLAN PCP/DEI field
:type mask_pcp_dei: Hex
:param value_pcp_dei: the value the masked VLAN PCP/DEI field is matched against
:type value_pcp_dei: Hex
:param mask_dscp: the mask applied to the IP DSCP field
:type mask_dscp: Hex
:param value_dscp: the value the masked IP DSCP field is matched against
:type value_dscp: Hex
:param handle: the UE packet handle mapped to the virtual channel
:type handle: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, credit_limit=credit_limit, mapping=mapping, mask_pcp_dei=mask_pcp_dei, value_pcp_dei=value_pcp_dei, mask_dscp=mask_dscp, value_dscp=value_dscp, handle=handle, indices=[self._vc_xindex, self._storage_xindex]))
[docs]
@register_command
@dataclass
class P_UE_CBFC_CLEAR:
"""
Clear UE CBFC counters and statistics in the specified direction(s).
"""
code: typing.ClassVar[int] = 1055
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class SetDataAttr(RequestBodyStruct):
direction: UecCbfcClearDirection = field(XmpByte())
"""coded byte, direction of the counters to clear."""
[docs]
def set(self, direction: UecCbfcClearDirection) -> Token[None]:
"""Clear the CBFC counters in the given direction.
:param direction: direction of the counters to clear
:type direction: UecCbfcClearDirection
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, direction=direction))
clear_none = functools.partialmethod(set, UecCbfcClearDirection.NONE)
"""Clear no CBFC counters.
"""
clear_rx = functools.partialmethod(set, UecCbfcClearDirection.RX)
"""Clear all CBFC RX counters.
"""
clear_tx = functools.partialmethod(set, UecCbfcClearDirection.TX)
"""Clear all CBFC TX counters.
"""
clear_all = functools.partialmethod(set, UecCbfcClearDirection.ALL)
"""Clear all CBFC RX and TX counters.
"""
[docs]
@register_command
@dataclass
class P_UE_CBFC_MODE:
"""
The CBFC mode of operation of the port, for the sender (TX) and the receiver (RX). Both
directions are ``ON`` by default.
Turning the sender ``OFF`` stops CC_Update generation and makes every packet bypass the
credit check, so traffic is forwarded uncredited, and the credit wallet is zeroed. Turning
the receiver ``OFF`` bypasses credit accounting altogether. Either way that direction's
CBFC statistics are cleared, while the CBFC link message configuration and the CBFC
virtual channel configuration are left untouched.
"""
code: typing.ClassVar[int] = 1056
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
rx_mode: UecCbfcMode = field(XmpByte())
"""coded byte, the CBFC receiver mode."""
tx_mode: UecCbfcMode = field(XmpByte())
"""coded byte, the CBFC sender mode."""
[docs]
class SetDataAttr(RequestBodyStruct):
rx_mode: UecCbfcMode = field(XmpByte())
"""coded byte, the CBFC receiver mode."""
tx_mode: UecCbfcMode = field(XmpByte())
"""coded byte, the CBFC sender mode."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CBFC mode of the port.
:return: the CBFC receiver and sender modes of the port
:rtype: P_UE_CBFC_MODE.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, rx_mode: UecCbfcMode, tx_mode: UecCbfcMode) -> Token[None]:
"""Set the CBFC mode of the port.
:param rx_mode: the CBFC receiver mode
:type rx_mode: UecCbfcMode
:param tx_mode: the CBFC sender mode
:type tx_mode: UecCbfcMode
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, rx_mode=rx_mode, tx_mode=tx_mode))
enable_rx_only = functools.partialmethod(set, UecCbfcMode.ON, UecCbfcMode.OFF)
"""Enable the CBFC receiver while disabling the sender.
"""
enable_tx_only = functools.partialmethod(set, UecCbfcMode.OFF, UecCbfcMode.ON)
"""Enable the CBFC sender while disabling the receiver.
"""
enable_both = functools.partialmethod(set, UecCbfcMode.ON, UecCbfcMode.ON)
"""Enable both the CBFC receiver and sender.
"""
disable_both = functools.partialmethod(set, UecCbfcMode.OFF, UecCbfcMode.OFF)
"""Disable both the CBFC receiver and sender.
"""
[docs]
@register_command
@dataclass
class P_UE_CBFC_VC_STREAMS:
"""
The streams that are mapped to a CBFC virtual channel. The stream-to-virtual-channel
mapping is N:1 and is set per stream with ``PS_UE_CBFC_VC``.
"""
code: typing.ClassVar[int] = 1054
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
_vc_xindex: int
[docs]
class GetDataAttr(ResponseBodyStruct):
stream_indices: typing.List[int] = field(XmpSequence(types_chunk=[XmpInt()]))
"""list of integers, the indices of the streams mapped to the virtual channel."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the streams mapped to the CBFC virtual channel.
:return: the indices of the streams mapped to the virtual channel
:rtype: P_UE_CBFC_VC_STREAMS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port, indices=[self._vc_xindex]))
[docs]
@register_command
@dataclass
class P_UE_CBFC_APPLY:
"""
Copy a staged CBFC configuration data storage into its active counterpart, and push it to
the hardware. Applying the local sender storage also re-programs the whole stream-to-virtual-
channel table, so a ``P_UE_CBFC_NUMVCS`` reduction is reflected in the hardware.
"""
code: typing.ClassVar[int] = 1042
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class SetDataAttr(RequestBodyStruct):
storage_index: int = field(XmpInt())
"""integer, the staged configuration data storage to apply."""
[docs]
def set(self, storage_index: int) -> Token[None]:
"""Apply a staged CBFC configuration data storage.
:param storage_index: the staged configuration data storage to apply, 0 (local receiver) or 1 (local sender)
:type storage_index: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, storage_index=storage_index))
local_rx = functools.partialmethod(set, storage_index=0)
"""Apply the local receiver staged configuration."""
local_tx = functools.partialmethod(set, storage_index=1)
"""Apply the local sender staged configuration."""
[docs]
@register_command
@dataclass
class P_UE_CBFC_INJECT_ERR:
"""
CBFC error injection of the port.
No burst pattern is defined for CBFC in this release. ``burst_size`` and ``burst_interval``
must still be supplied, but they are accepted without any range check and never reach the
hardware.
"""
code: typing.ClassVar[int] = 1043
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class SetDataAttr(RequestBodyStruct):
error_type: UecCbfcInjectErrType = field(XmpByte())
"""coded byte, the type of error to inject."""
pattern: UecCbfcInjectErrPattern = field(XmpByte())
"""coded byte, the pattern of error injection."""
burst_size: int = field(XmpInt())
"""integer, the number of errors in a burst. Not supported in this release, accepted and ignored."""
burst_interval: int = field(XmpInt())
"""integer, the interval between bursts. Not supported in this release, accepted and ignored."""
[docs]
def set(self, error_type: UecCbfcInjectErrType, pattern: UecCbfcInjectErrPattern, burst_size: int, burst_interval: int) -> Token[None]:
"""Set the CBFC error injection configuration of the port.
:param error_type: the type of error to inject
:type error_type: UecCbfcInjectErrType
:param pattern: the pattern of error injection
:type pattern: UecCbfcInjectErrPattern
:param burst_size: the number of errors in a burst, not supported in this release, accepted and ignored
:type burst_size: int
:param burst_interval: the interval between bursts, not supported in this release, accepted and ignored
:type burst_interval: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, error_type=error_type, pattern=pattern, burst_size=burst_size, burst_interval=burst_interval))
cc_update_inc = functools.partialmethod(set, UecCbfcInjectErrType.CC_UPDATE_INC, UecCbfcInjectErrPattern.ONCE, 1, 1)
"""Increase the CC_Update credits consumed counter once, creating a credit leak.
"""
# inject_cf_update_drop = functools.partialmethod(set, UecCbfcInjectErrType.CF_UPDATE_DROP, UecCbfcInjectErrPattern.ONCE, 1, 1)
#"""Drop the next CF_Update once.
#"""
cc_update_bad_fcs = functools.partialmethod(set, UecCbfcInjectErrType.CC_UPDATE_BAD_FCS, UecCbfcInjectErrPattern.ONCE, 1, 1)
"""Inject a bad FCS in the next CC_Update message.
"""
cc_update_poisoned_fcs = functools.partialmethod(set, UecCbfcInjectErrType.CC_UPDATE_POISONED_FCS, UecCbfcInjectErrPattern.ONCE, 1, 1)
"""Inject a poisoned FCS in the next CC_Update message.
"""
[docs]
@register_command
@dataclass
class P_UE_CBFC_CC_INC:
"""
The number of credits added to the CC_Update credits consumed counter by the
CC_UPDATE_INC error injection, and the virtual channel it is applied to.
CC_Update is generated by the sender, so ``vc_index`` is resolved against the local sender
staged storage: it must be below its ``P_UE_CBFC_NUMVCS``, and a non-zero ``credits`` also
requires that virtual channel to be lossless.
"""
code: typing.ClassVar[int] = 1044
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
vc_index: int = field(XmpInt())
"""integer, the virtual channel the credit leak is injected on."""
credit_increment: int = field(XmpInt())
"""integer, the number of credits added to the credits consumed counter."""
[docs]
class SetDataAttr(RequestBodyStruct):
vc_index: int = field(XmpInt())
"""integer, the virtual channel the credit leak is injected on."""
credit_increment: int = field(XmpInt())
"""integer, the number of credits added to the credits consumed counter, in the range 0 to 1048575."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CC_Update credit increment of the port.
:return: the CC_Update credit increment of the port
:rtype: P_UE_CBFC_CC_INC.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
def set(self, vc_index: int, credit_increment: int) -> Token[None]:
"""Set the CC_Update credit increment of the port.
:param vc_index: the virtual channel the credit leak is injected on
:type vc_index: int
:param credit_increment: the number of credits added to the credits consumed counter, in the range 0 to 1048575
:type credit_increment: int
"""
return Token(self._connection, build_set_request(self, module=self._module, port=self._port, vc_index=vc_index, credit_increment=credit_increment))
[docs]
@register_command
@dataclass
class P_UE_CBFC_TX_STATS:
"""
CBFC Tx statistics of the port.
"""
code: typing.ClassVar[int] = 1046
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
cc_update_count: int = field(XmpLong())
"""long integer, the number of CC_Update messages transmitted since the last clear."""
cc_update_interval_min: int = field(XmpLong())
"""long integer, the minimum interval in microseconds between two consecutive Tx CC_Update since the last clear."""
cc_update_interval_max: int = field(XmpLong())
"""long integer, the maximum interval in microseconds between two consecutive Tx CC_Update since the last clear."""
cc_update_interval_avg: int = field(XmpLong())
"""long integer, the average interval in microseconds between two consecutive Tx CC_Update since the last clear."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CBFC Tx statistics of the port.
:return: the CBFC Tx statistics of the port
:rtype: P_UE_CBFC_TX_STATS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_CBFC_RX_STATS:
"""
CBFC Rx statistics of the port.
"""
code: typing.ClassVar[int] = 1047
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
cc_update_count: int = field(XmpLong())
"""long integer, unsigned, the number of CC_Update messages received since the last clear."""
cc_update_interval_min: int = field(XmpLong())
"""long integer, the minimum interval in microseconds between two consecutive Rx CC_Update since the last clear."""
cc_update_interval_max: int = field(XmpLong())
"""long integer, the maximum interval in microseconds between two consecutive Rx CC_Update since the last clear."""
cc_update_interval_avg: int = field(XmpLong())
"""long integer, the average interval in microseconds between two consecutive Rx CC_Update since the last clear."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CBFC Rx statistics of the port.
:return: the CBFC Rx statistics of the port
:rtype: P_UE_CBFC_RX_STATS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_CBFC_RX_ERRORS:
"""
CBFC Rx error counters of the port.
"""
code: typing.ClassVar[int] = 1049
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
[docs]
class GetDataAttr(ResponseBodyStruct):
lost_credits: int = field(XmpLong())
"""long integer, signed, the accumulative per-port Lost Credits since the last clear."""
cc_update_bad_cid: int = field(XmpLong())
"""long integer, unsigned, the number of CC_Update received with a CID that is not FA-7A-CB, since the last clear."""
cc_update_bad_opcode: int = field(XmpLong())
"""long integer, unsigned, the number of CC_Update received with an opcode that is not 0xFFFE, since the last clear."""
cc_update_unknown_msg_type: int = field(XmpLong())
"""long integer, unsigned, the number of CC_Update received with a message type value that is not 0x01 or 0x02, since the last clear."""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CBFC Rx error counters of the port.
:return: the CBFC Rx error counters of the port
:rtype: P_UE_CBFC_RX_ERRORS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port))
[docs]
@register_command
@dataclass
class P_UE_CBFC_VC_TX_STATS:
"""
CBFC Tx statistics of a virtual channel.
"""
code: typing.ClassVar[int] = 1050
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
_vc_xindex: int
[docs]
class GetDataAttr(ResponseBodyStruct):
bits_last_sec: int = field(XmpLong())
"""long integer, the number of bits transmitted in the last second."""
bytes_last_sec: int = field(XmpLong())
"""long integer, the number of bytes transmitted in the last second."""
pkts_last_sec: int = field(XmpLong())
"""long integer, the number of packets transmitted in the last second."""
bytes_total: int = field(XmpLong())
"""long integer, the total number of bytes transmitted."""
pkts_total: int = field(XmpLong())
"""long integer, the total number of packets transmitted."""
credits_consumed: int = field(XmpLong())
"""long integer, signed, sender VC credits consumed (S_VC_CC) since the last clear.
A best-effort VC always returns -1.
"""
credits_freed: int = field(XmpLong())
"""long integer, signed, sender VC credits freed (S_VC_CF) since the last clear.
A best-effort VC always returns -1.
"""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CBFC Tx statistics of the virtual channel.
:return: the CBFC Tx statistics of the virtual channel
:rtype: P_UE_CBFC_VC_TX_STATS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port, indices=[self._vc_xindex]))
[docs]
@register_command
@dataclass
class P_UE_CBFC_VC_RX_STATS:
"""
CBFC Rx statistics of a virtual channel.
"""
code: typing.ClassVar[int] = 1051
pushed: typing.ClassVar[bool] = False
_connection: 'interfaces.IConnection'
_module: int
_port: int
_vc_xindex: int
[docs]
class GetDataAttr(ResponseBodyStruct):
bits_last_sec: int = field(XmpLong())
"""long integer, the number of bits received in the last second."""
bytes_last_sec: int = field(XmpLong())
"""long integer, the number of bytes received in the last second."""
pkts_last_sec: int = field(XmpLong())
"""long integer, the number of packets received in the last second."""
bytes_total: int = field(XmpLong())
"""long integer, the total number of bytes received."""
pkts_total: int = field(XmpLong())
"""long integer, the total number of packets received."""
credits_consumed: int = field(XmpLong())
"""long integer, signed, receiver VC credits consumed (R_VC_CC) since the last clear.
A best-effort VC always returns -1.
"""
credits_freed: int = field(XmpLong())
"""long integer, signed, receiver VC credits freed (R_VC_CF) since the last clear.
A best-effort VC always returns -1.
"""
[docs]
def get(self) -> Token[GetDataAttr]:
"""Get the CBFC Rx statistics of the virtual channel.
:return: the CBFC Rx statistics of the virtual channel
:rtype: P_UE_CBFC_VC_RX_STATS.GetDataAttr
"""
return Token(self._connection, build_get_request(self, module=self._module, port=self._port, indices=[self._vc_xindex]))
__all__ = [
"P_UE_CTLOS_CLEAR",
"P_UE_CTLOS_RX_STATS",
"P_UE_CTLOS_TX_STATS",
"P_UE_CTLOS_SPACING",
"P_UE_CTLOS_RX_ERRORS",
"P_UE_LINKNEG_OPTIONS",
"P_UE_LINKNEG_OPTIONS_STATUS",
"P_UE_LLR_MODE",
"P_UE_LLR_RX_STATS",
"P_UE_LLR_TX_STATS",
"P_UE_LLR_REPLAY",
"P_UE_LLR_BEHAVIOR",
"P_UE_LLR_INIT",
"P_UE_LLR_INIT_ECHO",
"P_UE_LLR_ACKNACK",
"P_UE_LLR_INJECT_ERR",
"P_UE_LLR_POISONFCS",
"P_UE_LLR_INIT_ECHO_CHK",
"P_UE_LLR_TXFSM_STATE",
"P_UE_LLR_RXFSM_STATE",
"P_UE_LLR_STATUS",
"P_UE_CBFC_MODE",
"P_UE_CBFC_CCUPDATE_HDR",
"P_UE_CBFC_CCUPDATE_TIMER",
"P_UE_CBFC_CFUPDATE_TIMER",
"P_UE_CBFC_NUMVCS",
"P_UE_CBFC_LINK",
"P_UE_CBFC_VC_STREAMS",
"P_UE_CBFC_VC_TYPE",
"P_UE_CBFC_VC",
"P_UE_CBFC_APPLY",
"P_UE_CBFC_INJECT_ERR",
"P_UE_CBFC_CC_INC",
"P_UE_CBFC_CLEAR",
"P_UE_CBFC_TX_STATS",
"P_UE_CBFC_RX_STATS",
"P_UE_CBFC_RX_ERRORS",
"P_UE_CBFC_VC_TX_STATS",
"P_UE_CBFC_VC_RX_STATS",
]