src package

Subpackages

Module contents

DLT645协议Python实现库

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

class src.MeterServerService(server, address=bytearray(b'\\x00\\x00\\x00\\x00\\x00\\x00'), password_manager=<src.model.types.dlt645_type.PasswordManager object>)[源代码]

基类：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)[源代码]
set_address(address)[源代码]

写通讯地址

参数:

address (str)

返回:

set_password(password)[源代码]

写密码

参数:

password (str)

返回类型:

None

返回:

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)[源代码]

写事件记录

参数:
返回类型:

bool

返回:

set_04(di, value)[源代码]

写参变量

参数:
返回类型:

bool

返回:

get_data_item(di)[源代码]

获取数据项

参数:

di (int) -- 数据项

返回类型:

Optional[DataItem]

返回:

handle_request(frame)[源代码]

处理读数据请求

参数:

frame

返回:

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客户端

返回类型:

Optional[MeterClientService]

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

创建RTU客户端

返回类型:

Optional[MeterClientService]

classmethod new_meter_client_service(client)[源代码]

创建新的MeterService实例

返回类型:

Optional[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]

send_and_handle_request(frame_bytes)[源代码]

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

返回类型:

Optional[DataItem]

Args:

frame_bytes: 要发送的帧数据

Returns:

DataItem: 成功时返回数据项 None: 超时或失败时返回

handle_response(frame)[源代码]

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

返回类型:

Optional[DataItem]

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

DLT645 协议实现类。

该类实现了 DLT645 协议的核心功能，包括数据编码/解码、校验和计算、 帧构建和解析等操作。所有方法均为类方法，可直接调用。

classmethod decode_data(data)[源代码]

数据域解码（-0x33H 转换）。

按照 DLT645 协议规定，对数据域进行解码处理， 每个字节减去 0x33H，并使用模 256 运算确保结果在 0-255 范围内。

参数:

data (bytes) -- 需要解码的数据字节。

返回:

解码后的原始数据。

返回类型:

bytes

classmethod calculate_checksum(data)[源代码]

校验和计算（模 256 求和）。

按照 DLT645 协议规定，计算数据的校验和， 对所有字节进行累加，然后对结果取模 256。

参数:

data (bytes) -- 需要计算校验和的数据字节。

返回:

计算得到的校验和值（0-255 范围）。

返回类型:

int

classmethod encode_data(data)[源代码]

数据域编码（+0x33H 转换）。

按照 DLT645 协议规定，对数据域进行编码处理， 每个字节加上 0x33H，并使用模 256 运算确保结果在 0-255 范围内。

参数:

data (bytes) -- 需要编码的原始数据字节。

返回:

编码后的数据。

返回类型:

bytes

classmethod build_frame(addr, ctrl_code, data)[源代码]

构建 DLT645 协议帧。

按照 DLT645 协议规定，构建完整的协议帧，包括前导字节、 地址域、控制码、数据域、校验和和结束标志。

参数:

  • addr (bytes) -- 设备地址（6字节）。

  • ctrl_code (int) -- 控制码。

  • data (bytes) -- 原始数据域（未编码）。

返回:

完整的 DLT645 协议帧。

返回类型:

bytearray

抛出:

ValueError -- 当地址长度不为6字节时抛出。

classmethod deserialize(raw)[源代码]

将字节数据反序列化为 Frame 对象。

从原始字节数据中解析出 DLT645 协议帧。如果无法找到完整的帧， 则抛出异常。

参数:

raw (bytes) -- 输入的原始字节数据。

返回:

解析出的 Frame 对象。

返回类型:

Optional[Frame]

抛出:

Exception -- 当无法找到完整帧时抛出。

classmethod deserialize_with_remaining(raw)[源代码]

将字节数据反序列化为 Frame 对象，并返回未解析的剩余数据。

从原始字节数据中解析出 DLT645 协议帧，并返回未解析的剩余数据。 如果数据不完整或无法解析，返回 (原始数据, None)。

参数:

raw (bytes) -- 输入的原始字节数据。

返回:

元组 (未解析的剩余数据, 解析出的帧)。 如果数据不完整或无法解析，返回 (原始数据, None)。

返回类型:

tuple[bytes, Optional[Frame]]

classmethod serialize(frame)[源代码]

将 Frame 对象序列化为字节数据。

按照 DLT645 协议规定，将 Frame 对象序列化为完整的协议帧字节。

参数:

frame (Frame) -- 需要序列化的 Frame 对象。

返回:

序列化后的协议帧字节。

返回类型:

Optional[bytes]

抛出:

Exception -- 当起始或结束标志无效时抛出。

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

基类：IntEnum

控制码枚举。

DLT645 协议中控制码用于标识帧的功能类型。

变量:

  • BroadcastTimeSync -- 广播校时命令（0x08）

  • ClearDemand -- 需量清零命令（0x10）

  • ReadData -- 读数据命令（0x11）

  • ReadAddress -- 读通讯地址命令（0x13）

  • WriteData -- 写数据命令（0x14）

  • WriteAddress -- 写通讯地址命令（0x15）

  • FreezeCmd -- 冻结命令（0x16）

  • ChangeBaudRate -- 修改通信速率命令（0x17）

  • ChangePassword -- 修改密码命令（0x18）

BroadcastTimeSync = 8
ClearDemand = 16
ReadData = 17
ReadAddress = 19
WriteData = 20
WriteAddress = 21
FreezeCmd = 22
ChangeBaudRate = 23
ChangePassword = 24
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=datetime.datetime(2026, 2, 21, 1, 19, 18, 296140))[源代码]

基类：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, service)[源代码]

基类：object

TCP 服务器类，用于与 DLT645 客户端进行 TCP 通信。

变量:

  • ip -- 服务器 IP 地址。

  • port -- 服务器端口号。

  • timeout -- 连接超时时间（秒）。

  • ln -- 监听套接字。

  • service -- 服务实例，用于处理业务逻辑。

初始化 TCP 服务器。

参数:

  • ip (str) -- 服务器 IP 地址（如 '0.0.0.0'）。

  • port (int) -- 服务器端口号。

  • timeout (float) -- 连接超时时间（秒）。

  • service (Any) -- 服务实例，用于处理业务逻辑。

start()[源代码]

启动TCP服务器（非阻塞，在后台线程中运行）

stop()[源代码]

停止TCP服务器

is_running()[源代码]

检查服务器是否正在运行

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

基类：object

RTU 服务器类，用于与 DLT645 客户端进行串口通信。

该类实现了 RTU（Remote Terminal Unit）服务器功能， 支持与 DLT645 协议客户端进行串口通信。

变量:

  • port -- 串口名称。

  • data_bits -- 数据位。

  • stop_bits -- 停止位。

  • baud_rate -- 波特率。

  • parity -- 校验位。

  • timeout -- 超时时间（秒）。

  • service -- 服务实例，用于处理业务逻辑。

  • conn -- 串口连接对象。

初始化 RTU 服务器。

参数:

  • port (str) -- 串口端口名（如 '/dev/ttyUSB0'）。

  • data_bits (int) -- 数据位，默认 8。

  • stop_bits (int) -- 停止位，默认 1。

  • baud_rate (int) -- 波特率，默认 9600。

  • parity (str) -- 校验位，默认无校验。

  • timeout (float) -- 超时时间（秒），默认 5.0。

  • service (Any) -- 服务实例，用于处理业务逻辑。

start()[源代码]

启动RTU服务器（非阻塞，在后台线程中运行）

返回类型:

bool

stop()[源代码]

停止RTU服务器

返回类型:

bool

is_running()[源代码]

检查服务器是否正在运行

返回类型:

bool

handle_connection(conn)[源代码]

处理单个串口连接

返回类型:

None

Args:

conn: 串口连接对象，必须是 serial.Serial 实例

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

基类：object

TCP 客户端类，用于与 DLT645 设备进行 TCP 通信。

变量:

  • ip -- 服务器 IP 地址。

  • port -- 服务器端口号。

  • timeout -- 连接超时时间（秒）。

  • conn -- socket 连接对象。

初始化 TCP 客户端。

参数:

  • ip (str) -- 服务器 IP 地址（如 '0.0.0.0'）。

  • port (int) -- 服务器端口号。

  • timeout (float) -- 连接超时时间（秒），默认 5.0。

connect()[源代码]

连接到服务器

返回类型:

bool

disconnect()[源代码]

断开与服务器的连接

返回类型:

bool

send_request(data, write_timeout=2.0, read_timeout=5.0, total_timeout=10.0, min_response_len=1, retries=1)[源代码]

增强版TCP请求-响应（支持超时控制和分片数据处理）

返回类型:

Optional[bytes]

Args:

data: 要发送的请求数据 write_timeout: 数据写入超时(秒) read_timeout: 单次recv操作的超时(秒) total_timeout: 整个请求-响应的总超时(秒) min_response_len: 最小有效响应长度 retries: 失败重试次数

Returns:

bytes: 成功接收的响应数据 None: 超时或失败时返回

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

基类：object

RTU 客户端类，用于与 DLT645 设备进行串口通信。

该类实现了 RTU（Remote Terminal Unit）客户端功能， 支持与 DLT645 协议设备进行串口通信。

变量:

  • port -- 串口名称。

  • baud_rate -- 波特率。

  • data_bits -- 数据位。

  • stop_bits -- 停止位。

  • parity -- 校验位。

  • timeout -- 超时时间（秒）。

  • conn -- 串口连接对象。

初始化 RTU 客户端。

参数:

  • port (str) -- 串口名称（如 "COM1" 或 "/dev/ttyUSB0"）。

  • baud_rate (int) -- 波特率，默认 9600。

  • data_bits (int) -- 数据位，默认 8。

  • stop_bits (int) -- 停止位，默认 1。

  • parity (str) -- 校验位，默认无校验。

  • timeout (float) -- 超时时间（秒），默认 1.0。

connect()[源代码]

连接到串口

建立与指定串口的连接，配置为DLT645协议通信所需的参数。

返回类型:

bool

Returns:

bool: 连接成功返回True，失败返回False

disconnect()[源代码]

断开与串口的连接

关闭串口连接并释放资源。

返回类型:

bool

Returns:

bool: 断开连接成功返回True，失败返回False

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

发送请求并接收响应

向DLT645设备发送请求数据，并接收响应。支持超时处理和自动重试机制。

返回类型:

Optional[bytes]

Args:

data: 要发送的请求数据 retries: 失败重试次数，默认1次

Returns:

bytes: 成功接收的响应数据 None: 失败时返回

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