src package

Subpackages

Module contents

DLT645协议Python实现库

这个库提供了DLT645通信协议的完整实现，包括： - TCP和RTU通信方式 - 客户端和服务端功能 - 电能表数据读写操作 - 电能量、最大需量、变量数据类型支持 - 实时报文捕获功能

src.configure_logging(enabled=True, cmdlevel=None, filelevel=None, filename=<object object>, backup_count=None, limit=None, when=None, colorful=None, compression=None, is_backtrace=None)[源代码]

配置 dlt645 库的日志输出。

库默认**不输出任何日志**；调用本函数启用日志或调整日志配置。

参数:

enabled (bool) -- True 启用日志输出，False 关闭（移除所有日志 handler）。
cmdlevel (Optional[str]) -- 控制台日志级别（如 "DEBUG"/"INFO"/"WARNING"/"ERROR"）。 None 表示回退到各模块默认级别。
filelevel (Optional[str]) -- 文件日志级别。None 表示回退到各模块默认级别。
filename (Union[str, None, object]) -- 统一日志文件路径；指定后所有模块日志输出到该文件。传 None 恢复"各模块各自文件"；不传（默认）保持当前配置。
backup_count (Optional[int]) -- 日志文件保留数量（天）。
limit (Union[int, str, None]) -- 单文件大小限制（如 "10 MB"）。
when (Optional[str]) -- 按时间轮转（如 "midnight"、"D"、"H"）。
colorful (Optional[bool]) -- 控制台输出是否带颜色。
compression (Optional[str]) -- 轮转文件压缩格式（如 "zip"）。
is_backtrace (Optional[bool]) -- 是否显示调用位置（文件:函数:行号）。

返回类型:

None

示例:

from dlt645 import configure_logging

# 启用日志，控制台 INFO、文件 DEBUG
configure_logging(cmdlevel="INFO", filelevel="DEBUG")

# 所有日志统一写入指定文件
configure_logging(filename="logs/meter.log")

# 恢复各模块各自文件
configure_logging(filename=None)

# 关闭日志
configure_logging(enabled=False)

src.enable_logging(**kwargs)[源代码]

启用 dlt645 库日志输出（可传 configure_logging() 的参数）。

返回类型:: None

src.disable_logging()[源代码]

关闭 dlt645 库日志输出。

返回类型:: None

class src.MeterServerService(server, address=None, password_manager=None)[源代码]

基类：object

电表服务端服务类。

用于模拟 DLT645 电表设备，响应客户端的数据读写请求。

变量:

server -- 通信服务器（TCP 或 RTU）。
address -- 设备地址（6字节）。
password_manager -- 密码管理器。
clear_meter_event_records -- 电表清零事件记录列表。
event_records -- 事件记录列表。

初始化电表服务端服务。

参数:

server (Union[TcpServer, RtuServer]) -- 通信服务器实例（TcpServer 或 RtuServer）。
address (Optional[bytearray]) -- 设备地址，默认为全零。
password_manager (Optional[PasswordManager]) -- 密码管理器，默认创建新实例。

classmethod new_tcp_server(ip, port, timeout=5.0)[源代码]

创建 TCP 服务器

参数:

ip (str) -- IP 地址
port (int) -- 端口
timeout (float) -- 超时时间

返回类型:

MeterServerService

返回:

classmethod new_rtu_server(port, data_bits, stop_bits, baud_rate, parity, timeout)[源代码]

创建 RTU 服务器

参数:

port (str) -- 端口
data_bits (int) -- 数据位
stop_bits (int) -- 停止位
baud_rate (int) -- 波特率
parity (str) -- 校验位
timeout (float) -- 超时时间

返回类型:

MeterServerService

返回:

classmethod new_meter_server_service(server)[源代码]

创建新的MeterServerService实例

参数:: server (Union[TcpServer, RtuServer]) -- 服务器实例（TCP或RTU）
返回类型:: MeterServerService
返回:: MeterServerService实例

set_time(data_bytes)[源代码]

广播校时：解析 YYMMDDhhmmss 并设置表内时钟。

按 DL/T645-2007 第6节：广播校时数据域为 6 字节压缩 BCD 码，依次为年（后两位）、月、日、时、分、秒（自然顺序）。

参数:: data_bytes (bytearray) -- 数据域（6字节）。

set_address(address)[源代码]

写通讯地址

参数:: address (Union[str, bytes, bytearray])
返回类型:: bool
返回:

set_password(password)[源代码]

写密码

参数:: password (str)
返回类型:: bool
返回:

set_00(di, value)[源代码]

写电能量

参数:

di (int) -- 数据项
value (float) -- 值

返回类型:

bool

返回:

set_01(di, demand)[源代码]

写最大需量及发生时间

参数:

di (int) -- 数据项
demand (Demand) -- 值

返回类型:

bool

返回:

set_02(di, value)[源代码]

写变量

参数:

di (int) -- 数据项
value (float) -- 值

返回类型:

bool

返回:

set_03(di, value)[源代码]

写事件记录

参数:

di (int) -- 数据项
value (list[str, tuple[str, str]]) -- 值

返回类型:

bool

返回:

set_04(di, value)[源代码]

写参变量

参数:

di (int) -- 数据项
value (str | list) -- 值

返回类型:

bool

返回:

get_data_item(di)[源代码]

获取数据项

参数:: di (int) -- 数据项
返回类型:: Optional[DataItem]
返回:

handle_request(frame)[源代码]

处理读数据请求

参数:: frame
返回:

start()[源代码]

启动底层服务端。

返回类型:: bool

stop()[源代码]

停止底层服务端。

返回类型:: bool

close()

停止底层服务端。

返回类型:: bool

enable_message_capture(queue_size=100)[源代码]

启用报文捕获功能。

参数:: queue_size (int) -- 报文队列大小，默认100
返回类型:: None

disable_message_capture()[源代码]

禁用报文捕获功能。

返回类型:: None

get_captured_messages(count=0)[源代码]

获取捕获的报文列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 报文列表
返回类型:: List[MessageRecord]

get_captured_tx_messages(count=0)[源代码]

获取捕获的发送报文列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 发送报文列表
返回类型:: List[MessageRecord]

get_captured_rx_messages(count=0)[源代码]

获取捕获的接收报文列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 接收报文列表
返回类型:: List[MessageRecord]

get_captured_pairs(count=0)[源代码]

获取捕获的TX/RX配对列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 配对列表
返回类型:: List[MessagePair]

clear_captured_messages()[源代码]

清空所有捕获的报文。

返回类型:: None

get_message_capture_stats()[源代码]

获取报文捕获统计信息。

返回:: 统计信息字典
返回类型:: dict

class src.MeterClientService(client)[源代码]

基类：object

电表客户端服务类。

用于与 DLT645 电表设备进行通信，提供数据读写等业务功能。

变量:

address -- 设备地址（6字节）。
password_manager -- 密码管理器。
operation_code -- 操作码（4字节）。
client -- 通信客户端（TCP 或 RTU）。

初始化电表客户端服务。

参数:: client (Union[TcpClient, RtuClient]) -- 通信客户端实例（TcpClient 或 RtuClient）。

classmethod new_tcp_client(ip, port, timeout=30.0)[源代码]

创建TCP客户端

返回类型:: MeterClientService

classmethod new_rtu_client(port, baudrate, databits, stopbits, parity, timeout)[源代码]

创建RTU客户端

返回类型:: MeterClientService

classmethod new_meter_client_service(client)[源代码]

创建新的MeterService实例

返回类型:: MeterClientService

get_time(t)[源代码]

从字节数据获取时间

返回类型:: datetime

set_address(address)[源代码]

设置设备地址

返回类型:: bool

set_password(password)[源代码]

设置设备密码, 修改数据的命令会带上密码发送出去

返回类型:: bool

change_password(old_password, new_password)[源代码]

修改设备密码

返回类型:: bool

read_00(di)[源代码]

读取电能

返回类型:: Optional[DataItem]

read_01(di)[源代码]

读取最大需量及发生时间

返回类型:: Optional[DataItem]

read_02(di)[源代码]

读取变量

返回类型:: Optional[DataItem]

read_03(di)[源代码]

读取事件记录

返回类型:: Optional[DataItem]

read_04(di)[源代码]

读取参变量

返回类型:: Optional[DataItem]

write_04(di, value, password)[源代码]

写参变量

返回类型:: Optional[DataItem]

read_address()[源代码]

读取通讯地址

返回类型:: Optional[DataItem]

write_address(new_address)[源代码]

写通讯地址

返回类型:: Optional[DataItem]

broadcast_time_sync(dt=None)[源代码]

广播校时命令（C=08H，数据域YYMMDDhhmmss）。

按 DL/T645-2007 第6节：广播校时使用广播地址 999999999999H，数据域为 6 字节压缩 BCD 码（YY MM DD hh mm ss，自然顺序），从站不要求应答，因此只发送不等待响应。

参数:: dt (Optional[datetime]) -- 校时时间，默认当前时间。
返回:: 发送成功返回 True，失败返回 False。
返回类型:: bool

freeze(month=None, day=None, hour=None, minute=None, broadcast=False)[源代码]

冻结命令（C=16H，数据域MMDDhhmm）。

按 DL/T645-2007 第7节：数据域为 4 字节压缩 BCD 码（MM DD hh mm），各字段缺省时取当前时间；可传 99 表示周期定时冻结： - 99DDhhmm：以月为周期定时冻结 - 9999hhmm：以日为周期定时冻结 - 999999mm：以小时为周期定时冻结 - 99999999：瞬时冻结

参数:

month (Optional[int]) -- 月（1-12 或 99）。
day (Optional[int]) -- 日（1-31 或 99）。
hour (Optional[int]) -- 时（0-23 或 99）。
minute (Optional[int]) -- 分（0-59 或 99）。
broadcast (bool) -- 是否广播冻结（使用广播地址，从站不要求应答）。

返回:

成功返回 DataItem，失败返回 None。

返回类型:

Optional[DataItem]

change_baud_rate(baud)[源代码]

更改通信速率命令（C=17H，数据域1字节通信速率特征字）。

按 DL/T645-2007 第8节：数据域为 1 字节特征字（见附录C），从站正常应答帧中的数据与请求帧中的特征字必须相同。标准速率：1200/2400/4800/9600/19200 bps。

参数:: baud (int) -- 目标通信速率（bps）。
返回:: 成功返回 DataItem，失败返回 None。
返回类型:: Optional[DataItem]

clear_demand(di, password)[源代码]

最大需量清零命令（C=19H，数据域DI+密码）。

按 DL/T645-2007 第10节：数据域为 4 字节数据标识 + 4 字节密码，将当前最大需量及发生时间数据清零。需 04 级及以上权限。

参数:

di (int) -- 需量数据标识。
password (str) -- 密码（4字节BCD，首字节为权限级别，数值越小权限越高，需 0-4 级）。

返回:

成功返回 DataItem，失败返回 None。

返回类型:

Optional[DataItem]

clear_meter(password)[源代码]

电表清零命令（C=1AH，数据域DI+密码）。

按 DL/T645-2007 第11节：数据域为 4 字节数据标识（00000000H） + 4 字节密码，清空电能量、最大需量及发生时间、冻结量、事件记录、负荷记录等数据（电表清零事件记录不被清除）。

参数:: password (str) -- 密码（4字节BCD，首字节为权限级别，数值越小权限越高，需 0-2 级）。
返回:: 成功返回 DataItem，失败返回 None。
返回类型:: Optional[DataItem]

clear_event(password, operator_code='00000000', di=4294967295)[源代码]

事件清零命令（C=1BH，数据域密码+操作者代码+数据标识）。

按 DL/T645-2007 第12节：数据域为 4 字节密码 + 4 字节操作者代码 + 4 字节数据标识。数据标识为 FFFFFFFFH 表示事件总清零；分项事件清零时数据标识为事件记录数据标识（DI0 用 FF 表示）。

参数:

password (str) -- 密码（4字节BCD，首字节为权限级别，数值越小权限越高，需 0-2 级）。
operator_code (str) -- 操作者代码（4字节BCD）。
di (int) -- 数据标识，默认 FFFFFFFFH（事件总清零）。

返回:

成功返回 DataItem，失败返回 None。

返回类型:

Optional[DataItem]

send_and_handle_request(frame_bytes)[源代码]

发送请求并处理响应（带超时控制）

返回类型:: Optional[DataItem]

Args:: frame_bytes: 要发送的帧数据
Returns:: DataItem: 成功时返回数据项 None: 超时或失败时返回

handle_response(frame)[源代码]

处理响应帧，包括异常帧检测

返回类型:: Optional[DataItem]

connect()[源代码]

建立底层连接。

返回类型:: bool

disconnect()[源代码]

关闭底层连接。

返回类型:: bool

close()

关闭底层连接。

返回类型:: bool

enable_message_capture(queue_size=100)[源代码]

启用报文捕获功能。

参数:: queue_size (int) -- 报文队列大小，默认100
返回类型:: None

disable_message_capture()[源代码]

禁用报文捕获功能。

返回类型:: None

get_captured_messages(count=0)[源代码]

获取捕获的报文列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 报文列表
返回类型:: List[MessageRecord]

get_captured_tx_messages(count=0)[源代码]

获取捕获的发送报文列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 发送报文列表
返回类型:: List[MessageRecord]

get_captured_rx_messages(count=0)[源代码]

获取捕获的接收报文列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 接收报文列表
返回类型:: List[MessageRecord]

get_captured_pairs(count=0)[源代码]

获取捕获的TX/RX配对列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 配对列表
返回类型:: List[MessagePair]

clear_captured_messages()[源代码]

清空所有捕获的报文。

返回类型:: None

get_message_capture_stats()[源代码]

获取报文捕获统计信息。

返回:: 统计信息字典
返回类型:: dict

class src.DLT645Protocol[源代码]

基类：object

无状态的 DL/T 645 帧编解码器。

MAX_DATA_LENGTH = 255

MAX_PREAMBLE_LENGTH = 4

MIN_FRAME_LENGTH = 12

classmethod decode_data(data)[源代码]

移除数据域逐字节增加的 0x33 偏移。

返回类型:: bytes

classmethod encode_data(data)[源代码]

按协议为数据域逐字节增加 0x33 偏移。

返回类型:: bytes

classmethod calculate_checksum(data)[源代码]

计算从第一个 0x68 到数据域末尾的模 256 累加和。

返回类型:: int

classmethod build_frame(addr, ctrl_code, data, preamble_count=4)[源代码]

从解码后的地址、控制码和数据域构建完整帧。

返回类型:: bytearray

classmethod deserialize(raw)[源代码]

解析第一帧；数据不完整或没有合法帧时抛出 ValueError。

返回类型:: Optional[Frame]

classmethod deserialize_with_remaining(raw)[源代码]

从字节流中解析一帧，并返回帧后的剩余数据。

无效噪声和损坏帧会被丢弃；不完整帧会从首个有效候选 0x68 起保留，供下一批数据继续拼接。

返回类型:: tuple[bytes, Optional[Frame]]

classmethod serialize(frame)[源代码]

序列化帧；前导 0xFE 不参与校验和计算。

返回类型:: bytes

class src.CtrlCode(value)[源代码]

基类：IntEnum

控制码枚举。

DLT645 协议中控制码用于标识帧的功能类型，取值遵循 DL/T645-2007 标准。

变量:

BroadcastTimeSync -- 广播校时命令（0x08）
ReadData -- 读数据命令（0x11）
ReadAddress -- 读通讯地址命令（0x13）
WriteData -- 写数据命令（0x14）
WriteAddress -- 写通讯地址命令（0x15）
FreezeCmd -- 冻结命令（0x16）
ChangeBaudRate -- 修改通信速率命令（0x17）
ChangePassword -- 修改密码命令（0x18）
ClearDemand -- 最大需量清零命令（0x19）
ClearMeter -- 电表清零命令（0x1A）
ClearEvent -- 事件清零命令（0x1B）

BroadcastTimeSync = 8

ReadData = 17

ReadAddress = 19

WriteData = 20

WriteAddress = 21

FreezeCmd = 22

ChangeBaudRate = 23

ChangePassword = 24

ClearDemand = 25

ClearMeter = 26

ClearEvent = 27

class src.Demand(value, time)[源代码]

基类：object

需量数据类。

用于表示最大需量及其发生时间。

变量:

value -- 需量值（单位由具体数据项定义）。
time -- 需量发生时间。

初始化 Demand 实例。

参数:

value (float) -- 需量值。
time (datetime) -- 需量发生时间。

class src.DataItem(di, name, data_format, value=0, unit='', update_time=None)[源代码]

基类：object

数据项实体类，表示一个具体的数据项。

变量:

di -- 数据标识 (DI)，4 字节。
name -- 数据项名称。
data_format -- 数据格式字符串（如 "XXXXXX.XX"）。
value -- 数据值。
unit -- 数据单位。
update_time -- 最后更新时间戳。

初始化 DataItem 实例。

参数:

di (int) -- 数据标识 (DI)。
name (str) -- 数据项名称。
data_format (str) -- 定义数据结构的格式。
value (Union[str, float, List], 可选) -- 数据项的实际值，默认为 0。
unit (str, 可选) -- 单位字符串（如 "kWh"），默认为空字符串。
update_time (datetime.datetime, 可选) -- 数据更新的时间戳，默认为当前时间。

class src.DataFormat(value)[源代码]

基类：Enum

表示各种数据格式模板的枚举。

定义说明： - X: 十进制数字 - Y: 年 - M: 月 - D: 日 - W: 星期 - h: 时 - m: 分 - s: 秒 - N: 数字（通用）

XXXXXXXX = 'XXXXXXXX'

XXXXXX_XX = 'XXXXXX.XX'

XXXX_XX = 'XXXX.XX'

XXX_XXX = 'XXX.XXX'

XX_XXXX = 'XX.XXXX'

XXX_X = 'XXX.X'

X_XXX = 'X.XXX'

YYMMDDWW = 'YYMMDDWW'

hhmmss = 'hhmmss'

YYMMDDhhmm = 'YYMMDDhhmm'

NN = 'NN'

NNNN = 'NNNN'

NNNNNNNN = 'NNNNNNNN'

class src.TcpServer(ip, port, timeout=5.0, service=None)[源代码]

基类：object

为每条 TCP 连接创建一个工作线程的服务端。

MAX_BUFFER_SIZE = 4096

start()[源代码]

启动监听并等待绑定完成；重复调用是幂等的。

返回类型:: bool

stop()[源代码]

关闭监听、活动连接和工作线程；重复调用是幂等的。

返回类型:: bool

is_running()[源代码]

返回类型:: bool

handle_connection(conn)[源代码]

返回类型:: None

class src.RtuServer(port, data_bits=8, stop_bits=1, baud_rate=9600, parity='N', timeout=5.0, service=None)[源代码]

基类：object

在后台线程中读取和处理串口帧。

MAX_BUFFER_SIZE = 4096

start()[源代码]

返回类型:: bool

stop()[源代码]

返回类型:: bool

is_running()[源代码]

返回类型:: bool

handle_connection(conn)[源代码]

返回类型:: None

class src.TcpClient(ip='', port=0, timeout=5.0)[源代码]

基类：object

支持分片响应、总超时和串行请求的 TCP 客户端。

MAX_BUFFER_SIZE = 4096

connect()[源代码]

建立连接；已有有效连接时直接复用。

返回类型:: bool

disconnect()[源代码]

关闭连接；重复调用是幂等的。

返回类型:: bool

send_request(data, write_timeout=None, read_timeout=None, total_timeout=None, min_response_len=1, retries=1)[源代码]

发送请求并返回一个经过校验的完整响应帧。

返回类型:: Optional[bytes]

send_only(data, timeout=2.0)[源代码]

只发送数据，不等待响应。

返回类型:: bool

class src.RtuClient(port='', baud_rate=9600, data_bits=8, stop_bits=1, parity='N', timeout=1.0)[源代码]

基类：object

支持分片读取、总超时与重试的串口客户端。

MAX_BUFFER_SIZE = 4096

connect()[源代码]

返回类型:: bool

disconnect()[源代码]

返回类型:: bool

send_request(data, retries=1)[源代码]

返回类型:: Optional[bytes]

send_only(data)[源代码]

返回类型:: bool

class src.MessageRecord(direction, data, id=<factory>, timestamp=<factory>, hex_string='', pair_id=None)[源代码]

基类：object

单条报文记录。

用于存储发送或接收的单条报文数据。

变量:

id -- 唯一标识符 (UUID)
direction -- 报文方向，"TX"（发送）或 "RX"（接收）
data -- 原始报文数据
timestamp -- 时间戳（秒，带小数部分）
hex_string -- 十六进制字符串表示
pair_id -- 配对的报文ID，用于TX/RX关联

direction: str

data: bytes

id: str

timestamp: float

hex_string: str = ''

pair_id: Optional[str] = None

property formatted_time: str

获取格式化的时间字符串（年-月-日时:分:秒.毫秒）。

返回:: 格式化的时间字符串
返回类型:: str

to_dict()[源代码]

转换为字典格式。

返回:: 包含所有字段的字典
返回类型:: dict

class src.MessagePair(id=<factory>, tx=None, rx=None, round_trip_time=None)[源代码]

基类：object

TX/RX配对记录。

用于将发送报文和接收报文配对，便于分析请求-响应对。

变量:

id -- 配对ID
tx -- 发送报文记录
rx -- 接收报文记录
round_trip_time -- 往返时间（秒）

id: str

tx: Optional[MessageRecord] = None

rx: Optional[MessageRecord] = None

round_trip_time: Optional[float] = None

set_tx(record)[源代码]

设置发送报文。

参数:: record (MessageRecord) -- 发送报文记录
返回类型:: None

set_rx(record)[源代码]

设置接收报文。

参数:: record (MessageRecord) -- 接收报文记录
返回类型:: None

is_complete()[源代码]

检查配对是否完整（有TX和RX）。

返回:: 配对是否完整
返回类型:: bool

property tx_time: str | None

获取发送报文的格式化时间。

返回:: 格式化的发送时间，如果没有TX则返回None
返回类型:: Optional[str]

property rx_time: str | None

获取接收报文的格式化时间。

返回:: 格式化的接收时间，如果没有RX则返回None
返回类型:: Optional[str]

to_dict()[源代码]

转换为字典格式。

返回:: 包含所有字段的字典
返回类型:: dict

class src.MessageCapture(enabled=False, queue_size=100)[源代码]

基类：object

报文捕获管理器。

管理TX和RX报文的捕获、配对和存储。默认禁用，需要显式调用enable()启用。

变量:

_enabled -- 是否启用捕获
_queue_size -- 队列大小
_tx_queue -- 发送报文队列
_rx_queue -- 接收报文队列
_pairs -- 配对队列
_pending_pairs -- 等待配对的记录（TX等待RX）

初始化报文捕获管理器。

参数:

enabled (bool) -- 是否启用捕获，默认False
queue_size (int) -- 队列大小，默认100

property enabled: bool: 获取启用状态。

property queue_size: int: 获取队列大小。

enable()[源代码]

启用报文捕获。

返回类型:: None

disable()[源代码]

禁用报文捕获。

返回类型:: None

set_queue_size(size)[源代码]

设置队列大小。

参数:: size (int) -- 新的队列大小
返回类型:: None

capture_tx(data)[源代码]

捕获发送报文。

参数:: data (bytes) -- 发送的原始数据
返回:: 事务ID，用于后续配对RX
返回类型:: Optional[str]

capture_rx(data, tx_id=None)[源代码]

捕获接收报文。

参数:

data (bytes) -- 接收的原始数据
tx_id (Optional[str]) -- 对应的TX事务ID，用于配对

返回类型:

None

capture_rx_for_server(data)[源代码]

捕获服务器端接收报文（作为请求）。

对于服务器端，RX是请求，TX是响应。创建配对并返回事务ID，用于后续配对TX响应。

参数:: data (bytes) -- 接收的原始数据
返回:: 事务ID，用于后续配对TX
返回类型:: Optional[str]

capture_tx_for_server(data, rx_id=None)[源代码]

捕获服务器端发送报文（作为响应）。

参数:

data (bytes) -- 发送的原始数据
rx_id (Optional[str]) -- 对应的RX事务ID，用于配对

返回类型:

None

get_tx_messages(count=0)[源代码]

获取发送报文列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 发送报文列表
返回类型:: List[MessageRecord]

get_rx_messages(count=0)[源代码]

获取接收报文列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 接收报文列表
返回类型:: List[MessageRecord]

get_pairs(count=0)[源代码]

获取配对列表。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 配对列表
返回类型:: List[MessagePair]

get_all_messages(count=0)[源代码]

获取所有报文（TX和RX），按时间排序。

参数:: count (int) -- 要获取的数量，0表示全部
返回:: 所有报文列表
返回类型:: List[MessageRecord]

clear()[源代码]

清空所有捕获的报文。

返回类型:: None

get_stats()[源代码]

获取捕获统计信息。

返回:: 统计信息字典
返回类型:: dict

class src.MessageQueue(maxlen=100)[源代码]

基类：object

线程安全的报文队列。

使用deque实现固定大小的循环队列，当队列满时自动丢弃最旧的记录。

变量:

_queue -- 内部队列
_lock -- 线程锁

初始化报文队列。

参数:: maxlen (int) -- 队列最大长度，默认100

property maxlen: int: 获取队列最大长度。

append(record)[源代码]

添加记录到队列。

参数:: record -- 要添加的记录（MessageRecord或MessagePair）
返回类型:: None

get_all()[源代码]

获取所有记录。

返回:: 所有记录的列表副本
返回类型:: List

get_recent(count)[源代码]

获取最近的N条记录。

参数:: count (int) -- 要获取的记录数量
返回:: 最近的记录列表
返回类型:: List

clear()[源代码]

清空队列。

返回类型:: None

resize(new_maxlen)[源代码]

调整队列大小。

参数:: new_maxlen (int) -- 新的最大长度
返回类型:: None