该库是基于开源的C-DLMS库进行开发的，并且支持 Client / Server 模式。

支持 DLMS 功能的模组型号如下：

系列	型号
EC200U	EC200UCN_AA_DLMS

💡 Tips

• 本库支持Client/Server模式，适用于智能电表等物联网设备。
• 商业应用请走商务流程。

DLMS COSEM Objects

COSEM Objects 共有属性和方法

Event hooks

dlms 模块中的每个对象都是某个子类的实例，为每个对象添加了以下功能：
六个属性（CosemObject，基类）
on_before_read、on_after_read、on_before_write、on_after_write、on_before_action、on_after_action ，每个对象都可以设置为一个回调函数。

import dlms
# 使用clock对象+on_before_read举例子
#on_after_read、on_before_write、on_after_write、on_before_action、on_after_action也类似
clock = dlms.Clock("0.0.1.0.0.255")

def on_before_read_clock(self, event):
    if event.index == 2:  # attr 2 = time
        print("[DLMS] Client is reading clock.time")
        # 返回 True 或 None → 允许读取（C 层已从 RTC 刷新了时间）
        return True

    elif event.index == 3:  # attr 3 = time_zone
        print("[DLMS] Client is reading clock.time_zone")
        return True

    return True  # 默认允许

# 绑定 handler 在client/server通信中才能触发
clock.on_before_read = on_before_read_clock

Per-instance access control

access_dict 属性接受一个与 access 构造器参数格式相同的访问字典，并会覆盖该特定对象实例的类级别默认值。

每个对象都有一个默认的访问字典，该字典由 set_default_access 函数设置。

import dlms
# 使用Clock对象举例子，因为每个对象都可以设置access_dict属性，所以可以针对每个对象设置不同的访问字典
# 类级别：所有 Clock 默认客户端只读
dlms.set_default_access(dlms.Clock, {
    2: (dlms.AccessMode.READ, dlms.Authentication.NONE),   # time 只读
    3: (dlms.AccessMode.READ, dlms.Authentication.NONE),   # time_zone 只读
})

# 两个 Clock 对象
clk_a = dlms.Clock("0.0.1.0.0.255")  # 允许客户端只读
clk_b = dlms.Clock("0.0.2.0.0.255")  # 允许客户端只读

# 但 clk_b 需要允许 HLS 客户端同步时间 → 实例级别覆盖
clk_b.access_dict = {
    2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH),  # time 可写
}

obj.idx(name)+obj.attr_name(index)

每个对象都可以使用这两种方法

import dlms
#Clock对象使用obj.idx和obj.attr_name举例子，每个对象都可以这样参考clock的例子来写
# 1. 创建 Clock 对象
clock = dlms.Clock("0.0.1.0.0.255", time_zone=8, deviation=0)
print("=== COSEM Objects ===")
print(clock)
print()
# 2. idx(name) — Python 名 → DLMS 编号
print("clock.idx('time')        = {}".format(clock.idx('time')))         # 2
print("clock.idx('time_zone')   = {}".format(clock.idx('time_zone')))    # 3
print("clock.idx('status')      = {}".format(clock.idx('status')))       # 4
print("clock.idx('begin')       = {}".format(clock.idx('begin')))        # 5
print("clock.idx('end')         = {}".format(clock.idx('end')))          # 6
print("clock.idx('deviation')   = {}".format(clock.idx('deviation')))    # 7
print("clock.idx('enabled')     = {}".format(clock.idx('enabled')))      # 8
print("clock.idx('base')        = {}".format(clock.idx('base')))         # 9
# 验证：不存在的名字会抛异常
print()
try:
    clock.idx('foobar')
except Exception as e:
    print("clock.idx('foobar') throw exception: {}".format(type(e).__name__, e))

# 3. attr_name(index) — DLMS 编号 → Python 名
print()
print("=== clock.attr_name() — DLMS Number → Python attr ===")
for i in range(2, 10):
    print("clock.attr_name({}) = '{}'".format(i, clock.attr_name(i)))

# 4. attrs() — 可持久化属性列表（跳过 VOLATILE）
print()
print("=== clock.attrs() — Persistent attribute list ===")
attrs_list = clock.attrs()
print("backup {} attributes:".format(len(attrs_list)))
for index, name in attrs_list:
    print("clock.attr_name({}) = '{}'".format(index, name))

print()
print("note: (2, 'time') had been skipped, because it's DLMS_ATTR_VOLATILE")

# 5. 实际使用：用 idx() 配合 client.read()
print()
print("=== real situation ===")
print("example client.read(clock, 2):")
print("  client.read(clock, clock.idx('time'))")
print("  client.read(clock, clock.idx('time_zone'))")
print("This way, the code becomes more readable and there is no need to remember the Blue Book numbers.")

obj.idx(name)+obj.attr_name(index) 实际应用

在事件处理程序和客户端代码中使用这些方法，可以避免硬编码魔术数字，从而无需查阅蓝皮书进行解码。例如，所以clock.idx('time') 返回 client.read(clock, clock.idx('time')) 等价于 client.read(clock, 2)，但具有自文档化特性。 obj.attrs() 返回一个包含所有复杂属性（如 (attr_index, attr_name) 元组）的列表，这些属性将被 BinarySerializer 保存。不包括瞬态属性（如 Clock.time 和 ProfileGeneric.buffer）。该列表在构建自定义序列化器时尤为有用。

import dlms #该例子不完整，仅供说明参考
# × 没有 idx()：必须记住每个属性的 Blue Book 编号,
client.read(clock, 2)       # 2 是什么？time？time_zone？
client.write(clock, 7, 60)  # 7 是什么？deviation？enabled？

# √ 有 idx()：用 Python 属性名，自文档化
client.read(clock, clock.idx('time'))
client.write(clock, clock.idx('deviation'), 60)

---

Data

DLMS 中最通用的数据容器。可存储 任意类型 （整数、字符串、字节、布尔值、空值，甚至引用其他对象的属性），其 COSEM ID=1 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	value	CHOICE（任意类型）	存储的实际数据

构造函数

Data(logical_name: str, nocopy: bool = False, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "1.0.1.8.0.255"
nocopy	bool	False	若为 True ， value 可设为 (obj, attr_index) 元组引用其他属性
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

支持的 value 类型

import dlms
#初始化Data对象
data = dlms.Data("1.0.1.8.0.255",nocopy=False,access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)})
#Data对象有logical_name和value属性
print(data.logical_name)  #输出“1.0.1.8.0.255”
print(data.value)         #输出None

#另一种输出Data对象的属性的方式
print(data)  # <Data ln='1.0.1.8.0.255', value=None>

#设置Data对象的value属性
data.value = 42           # 整数
print(data.value)         # 输出42

data.value = b"\x01\x02"  # 字节串
print(data.value)         # 输出DLMS协议的b'\x01\x02' 

data.value = "hello"      # 字符串
print(data.value)         # 输出DLMS协议的"hello"

data.value = True         # 布尔值
print(data.value)         # 输出True

data.value = None         # 设置为None
print(data.value)         # 输出None

实时引用（BYREF）模式（ nocopy=True ）

以 nocopy=True 构造 Data 对象，并将 (target_object, attribute_index) 元组赋值给 value ，即可启用实时引用模式。此后当客户端读取该 Data 对象时，服务器会在读取的那一刻透明地从目标对象取回被引用属性的 当前值 ，而不做任何复制。该模式最常见的用途是暴露位于 SecuritySetup 对象内部的调用计数器（invocation counter）。

import dlms
data = dlms.Data("1.0.1.8.0.255",nocopy=True,access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)})
sec = dlms.SecuritySetup("0.0.43.0.2.255")
data.value = (sec, 6)       # 获取SecuritySetup对象的in_invocation_counter访问权
print(data.value)           # 输出sec.min_invocation_counter的值，开始未赋值，输出0

sec.min_invocation_counter = 100
print(data.value)           # 输出sec.min_invocation_counter的值，现在为100

on_before_write — 写前验证

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
value	int | bytes | str | bool | None | tuple	√	数据值，支持引用模式
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

访问控制

import dlms

# 构造函数中指定
data = dlms.Data("1.0.1.8.0.255",access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)})
# 输出{2: (AccessMode.READ_WRITE, Authentication.HIGH)} 安全模式的返回值可看安全模型（Security）章节
print(data.access_dict)
# 或事后修改
data.access_dict = {2: (dlms.AccessMode.READ, dlms.Authentication.HIGH)}
 # 输出{2: (AccessMode.READ_WRITE, Authentication.HIGH)}  安全模式的返回值可看安全模型（Security）章节
print(data.access_dict) 

典型用例

场景	示例
存储电表序列号	data.value = 12345678
存储固件版本字符串	data.value = "v1.2.3"
存储自定义状态标志	data.value = True
事件通知的消息载体	data.value = "Tamper detected"
BYREF 联动读写	data.value = (other_obj, 2)

---

Register

DLMS 中表示 单一物理测量量 的标准对象。用于电能读数、瞬时功率、电压、电流及类似测量。内置 scaler （缩放因子）实现原始整数值到物理量的转换， unit 由 OBIS 代码自动派生，其 COSEM ID=3 。

Blue Book 属性

Python 属性	类型	DLMS 属性编号	说明
logical_name	str	1	OBIS 代码，只读
value	int	2	原始整数值，尚未应用 scaler
scaler	int	3（部分）	10 的幂次指数。
unit	int（只读）	3（部分）	物理单位， Unit 枚举值。由 OBIS 代码在构造时自动派生，不可修改

import dlms
reg = dlms.Register(
    "1.0.1.8.0.255",
    0, 
    scaler=-3,
    access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)})
print(reg)               #输出<Register ln='1.0.1.8.0.255', value=0, scaler=-3, unit=Unit.ACTIVE_ENERGY>
#属性修改或者不可修改
print(reg.logical_name)  #输出“1.0.1.8.0.255” reg.logical_name初始化后不可修改
reg.value = 12345678
print(reg.value)         # 输出12345678
reg.scaler = -5
print(reg.scaler)        # 输出-5
print(reg.unit)          # 输出Unit.ACTIVE_ENERGY ,OBIS 代码在构造时自动派生，不可修改

构造函数

Register(logical_name: str, default_value: int, scaler: int = 1, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "1.0.1.8.0.255"
default_value	int	必传	初始值，也是 reset() 恢复的目标值
scaler	int	1	10 的幂指数。 -3 表示原始值 ÷ 1000
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

物理量转换公式

physical_value = raw_value * (10 ** scaler)

scaler	含义	示例
0	原始值即物理值	value=100 → 100 kWh
-3	原始值 ÷ 1000	value=12345678 → 12345.678 kWh
3	原始值 × 1000	value=12 → 12000 kWh

unit （只读）由 OBIS 代码自动派生——例如 1.0.1.8.0.255 自动映射到 Unit.ACTIVE_ENERGY ，无需手动设置。

import dlms
reg = dlms.Register(
    "1.0.1.8.0.255", 
    0, 
    scaler=-3,
    access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)})
reg.value = 1234789
raw_value = reg.value * (10 ** reg.scaler) # 1234.789

方法	COSEM 方法	说明
reset()	1	将 value 重置为构造时的 default_value

如需 reset() 时同步清除硬件状态，使用 on_before_action 拦截：

# 默认行为：C 层直接恢复 value → 0
# reg.reset() 该方法并注册开到python库中，而是在通信时，触发before_reset回调
# 成功return ture则调用C层的reset方法，否则不调用C层的reset方法
import dlms
def before_reset(self, event):
    if event.index == 1:  # method 1 = reset
        # 此处可以调用硬件 API 清零外部计数器
        pass
    return True  # 允许执行

reg.on_before_action = before_reset

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
value	int	√	当前测量值（原始整数，未应用 scaler）
scaler	int	√	10 的幂指数，用于计算物理量
unit	int	×	物理单位（ Unit 枚举），由 OBIS 代码自动派生，不可修改
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

访问控制

import dlms
# 构造函数中指定
reg = dlms.Register("1.0.1.8.0.255", 0, scaler=-3,
    access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)})

# 或事后修改
reg.access_dict = {2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)}

典型用例

场景	示例
总正向有功电能，scaler=-3 换算 kWh	Register("1.0.1.8.0.255", 0, scaler=-3)
单相瞬时电压，scaler=-1 换算 V	Register("1.0.32.7.0.255", 0, scaler=-1)
单相瞬时电流，scaler=-2 换算 A	Register("1.0.31.7.0.255", 0, scaler=-2)
单相有功功率	Register("1.0.21.7.0.255", 0, scaler=0)

ExtendedRegister

Register 的扩展版本。在 value 、 scaler 、 unit 之外增加了 status （状态码）和 capture_time （捕获时间戳）两个属性，用于记录测量值产生时的上下文信息，其 COSEM ID=4 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	value	CHOICE	当前测量值（支持多种数值类型）
3	scaler	int	10 的幂次指数。
4	unit	int	物理单位， Unit 枚举值。由 OBIS 代码在构造时自动派生，不可修改
5	status	int	应用定义的状态码， 0 =OK， None =未记录
6	capture_time	octet-string	值的捕获时间戳

unit （只读）由 OBIS 代码自动派生，逻辑与 Register 完全一致。

import dlms
ext_reg = dlms.ExtendedRegister(
        "1.0.2.8.0.255",
        value=0,
        scaler=-3,
        status=0,
        capture_time=(2025, 1, 1, 0, 0, 0),
        access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)}
)
print(ext_reg) 
# 输出<ExtendedRegister ln='1.0.2.8.0.255', value=0, scaler=-3, unit=Unit.ACTIVE_ENERGY, status=0, capture_time=(2025, 1, 1, 0, 0, 0)>
#同时也可以单个打印：ext_reg.logical_name...
#部分属性可进行修改，例：
print(ext_reg.capture_time) #输出(2025, 1, 1, 0, 0, 0)
ext_reg.capture_time = (2026, 1, 1, 0, 0, 0) #变更为 2026, 1, 1, 0, 0, 0)
#部分属性不可修改，例：
print(ext_reg.logical_name) #输出“1.0.2.8.0.255” ext_reg.logical_name初始化后不可修改

构造函数

ExtendedRegister(logical_name: str, value: int | float | str = 0, scaler: int = 0,
                 status: int | None = None, capture_time: tuple | None = None,
                 access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "1.0.2.8.0.255"
value	int | float | str	0	初始测量值
scaler	int	0	10 的幂指数。 -3 表示原始值 ÷ 1000
status	int | None	None	状态码， 0 =OK， None =未记录状态
capture_time	tuple | None	None	捕获时间戳 (year, month, day, hour, min, sec)
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

物理量转换公式（同 Register）

physical_value = raw_value * (10 ** scaler)

import dlms
ext_reg = dlms.ExtendedRegister(
        "1.0.2.8.0.255",
        value=0,
        scaler=-3,
        status=0,
        capture_time=(2025, 1, 1, 0, 0, 0),
        access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)}
)
ext_reg.value = 1234789
raw_value = ext_reg.value * (10 ** ext_reg.scaler) # 1234.789

方法	COSEM 方法	说明
reset()	1	将 value 重置为构造时的默认值；行为与 Register 一致

# 默认行为：C 层直接恢复 value → 0
# reext_regg.reset() 该方法并注册开到python库中，而是在通信时，触发before_reset回调
# 成功return ture则调用C层的reset方法，否则不调用C层的reset方法
import dlms
def before_reset(self, event):
    if event.index == 1:  # method 1 = reset
        # 此处可以调用硬件 API 清零外部计数器
        pass
    return True  # 允许执行

ext_reg.on_before_action = before_reset

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
value	int | float | str	√	当前测量值（原始值，未应用 scaler）
scaler	int	√	10 的幂指数，用于计算物理量
unit	int	×	物理单位（ Unit 枚举），由 OBIS 代码自动派生
status	int | None	√	状态码， 0 =OK， None =未记录
capture_time	tuple | None	√	捕获时间戳 (year, month, day, hour, min, sec)
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

对比 Register

特性	Register	ExtendedRegister
value 类型	int	int | float | str
status	无	√
capture_time	无	√
reset()	√	√

访问控制

import dlms
# 构造函数中指定
ext_reg = dlms.ExtendedRegister("1.0.2.8.0.255", value=0, scaler=-3,
    access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)})

# 或事后修改
ext_reg.access_dict = {2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)}

典型用例

场景	示例
总反向有功电能，带状态和捕获时间	ExtendedRegister("1.0.2.8.0.255", value=0, scaler=-3, status=0, capture_time=(2025, 1, 15, 10, 30, 0))
瞬时功率，浮动状态码	ExtendedRegister("1.0.21.7.0.255", value=0, scaler=0, status=None)

---

ProfileGeneric

DLMS 标准对象，用于 历史数据存储 （负荷曲线、事件日志、账单记录）。内部维护一个有序行缓冲区，每行是若干捕获对象在某一时刻的快照，其 COSEM ID=7 。

	RAM Buffer （C 侧管理	外部存储 （Python 管理）
数据存放	gxProfileGeneric.buffer （堆内存）	flash / JSONL / EEPROM / 用户自定
自动 capture	cosem_invoke → 写入 RAM buffer	on_before_action → 你的 flash 写入函数
客户端 read	cosem_getValue → 从 RAM buffer 序列化	on_before_read → 你返回 list[list]
排序（FIFO 等）	C 侧自动淘汰旧行	仅协议上报； 实际排序由 Python 实现
持久化	掉电丢失	自己控制
BinarySerializer	N/A	自动跳过（buffer 标记为 COMPLEX）
适用场景	开发调试、小容量测试	生产环境 ── 必须使用

Blue Book 属性

编号	名称	DLMS 类型	Python 接口	说明
1	logical_name	octet-string(6)	logical_name	OBIS 代码，只读
2	buffer	ARRAY of STRUCTURE	由 on_before_read 提供	历史数据行；Python 侧不能直接读写 通过 buffer_size 查看行数）
3	capture_objects	ARRAY of STRUCTURE	capture_objects	定义每行捕获的对象、属性和数据索引
4	capture_period	uint32	capture_period	自动捕获周期（秒）， 0 表示禁用
5	sort_method	enum	sort_method	FIFO、LIFO、LARGEST 或 SMALLEST
6	sort_object	STRUCTURE	见下方三个 Python 属性	LARGEST/SMALLEST 的排序参照
7	entries_in_use	uint32	entries_in_use	当前历史记录数
8	profile_entries	uint32	profile_entries	最大历史记录数

标准属性只有 1 到 8。 sort_object_attribute_index 和 sort_object_data_index 不是属性 7、8，而是标准属性 6 sort_object 的组成部分：

Python 属性	对应 C 字段	默认值
sort_object	sortObject	None
sort_object_attribute_index	sortObjectAttributeIndex	2
sort_object_data_index	sortObjectDataIndex	0

排序方式常量

常量	值	说明
ProfileGeneric.FIFO	1	先进先出（最常用）
ProfileGeneric.LIFO	2	后进先出
ProfileGeneric.LARGEST	3	按 sort_object 排序，取最大值
ProfileGeneric.SMALLEST	4	按 sort_object 排序，取最小值

#使用DLMS RAM Buffer 模式
import dlms
clock = dlms.Clock("0.0.1.0.0.255", time_zone=480)
reg = dlms.Register("1.0.1.8.0.255", 0, scaler=-3)

# 先创建空壳
profile = dlms.ProfileGeneric("1.0.99.1.0.255")
print(profile.logical_name)                 # 输出 "1.0.99.1.0.255"
print(profile.buffer_size)                  #输出 0 当输出等于profile_entries时，表示buffer已满
# DLMS 属性 7，当前行数。每次 capture 自动 +1，上限 = profile_entries。
# 正常只读，不要手动改。
print(profile.entries_in_use)   

# 再逐步填充属性
profile.capture_objects = [clock, reg]       #  裸对象写法
profile.capture_period = 900                 # 15 分钟 更新 capture_objects对象的值
profile.sort_method = dlms.ProfileGeneric.FIFO
profile.profile_entries = 5                  # 5 行

# 需要capture_period设置为LARGEST和SMALLEST才生效，否则不生效
profile.sort_object = reg                    
profile.sort_object_attribute_index = 2      # 默认就是 2，可改
profile.sort_object_data_index = 0           # 默认就是 0，可改

# buffer 变化（FIFO, profile_entries=5）：
# T0 (12:00):  [(2025,8,2,12, 0,0),  100]   ← 第 1 行
# T1 (12:15):  [(2025,8,2,12,15,0),  110]   ← 第 2 行
# T2 (12:30):  [(2025,8,2,12,30,0),  105]   ← 第 3 行
# T3 (12:45):  [(2025,8,2,12,45,0),   98]   ← 第 4 行
# T4 (13:00):  [(2025,8,2,13, 0,0),  130]   ← 第 5 行，buffer 满了
#──────────────────────────────────────────
# T5 (13:15):  [(2025,8,2,13,15,0),  125]   ← 踢掉 T0，T5 加末尾
# T6 (13:30):  [(2025,8,2,13,30,0),  115]   ← 踢掉 T1，T6 加末尾

#使用DLMS RAM Buffer 模式
import dlms
clock = dlms.Clock("0.0.1.0.0.255", time_zone=480)
reg = dlms.Register("1.0.1.8.0.255", 0, scaler=-3)

# 先创建空壳
profile = dlms.ProfileGeneric("1.0.99.1.0.255")
# 再逐步填充属性
profile.capture_objects = [clock, reg]       # 裸对象写法
profile.capture_period = 900                 # 15 分钟 更新 capture_objects所有对象的值
profile.sort_method = dlms.ProfileGeneric.LARGEST
profile.profile_entries = 5                  # 3 行

# 需要capture_period设置为LARGEST和SMALLEST才生效，否则不生效
profile.sort_object = reg                    # 按 reg 的值比大小
profile.sort_object_attribute_index = 2      # reg 的属性 2 = value（默认）
profile.sort_object_data_index = 0           # 取整个值（默认，reg 是整数不需要子索引）


# sort_object_attribute_index和sort_object_data_index的用法
# 按 Clock 的时间比大小，只比"小时"字段
# profile.sort_object = clock
# profile.sort_object_attribute_index = 2      # Clock 的属性 2 = datetime 值
# profile.sort_object_data_index = 3           # datetime=(年,月,日,时,分,秒)，索引 3 = 小时

buffer 演变（LARGEST, profile_entries=3）：
T0 (12:00):  [(12, 0,0),  100]                                   reg 值: 100
T1 (12:15):  [(12,15,0),  110]                                   reg 值: 100, 110
T2 (12:30):  [(12,30,0),  105]                                   reg 值: 100, 110, 105 → 满了！
─────────────────────────────────────────────────────────────────
T3 (12:45):  [(12,45,0),   98]  ← reg=110 最大 → 踢掉 T1 那行
             结果：[(12,0,0),100]  [(12,30,0),105]  [(12,45,0),98]


#sort_object_attribute_index和sort_object_data_index的用法
#sort_method = LARGEST，sort_object = clock，data_index = 3（按"小时"比较）
#T0 (12:00):  [(12, 0,0),  100]   时=12
#T1 (14:00):  [(14, 0,0),  105]   时=14
#T2 (12:30):  [(12,30,0),  110]   时=12  → 满了！
#─────────────────────────────────────
#T3 (16:00):  [(16, 0,0),   98]   ← 时=14 最大 → 踢掉 T1
#            结果：[(12,0,0),100]  [(12,30,0),110]  [(16,0,0),98]

构造函数

ProfileGeneric(logical_name: str, capture_objects: list = None,
               capture_period: int = None, sort_method: int = None,
               sort_object: object = None, profile_entries: int = None,
               entries_in_use: int = None, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "1.0.99.1.0.255"
capture_objects	list	None	每行捕获的内容，支持三种格式（见下方说明）
capture_period	int	0	自动捕获间隔（秒）， 0 禁用，常见值 900 （15 分钟）
sort_method	int	0 （无效）	排序方式（见下方常量表）。 默认 0 不是有效值 ，请始终显式传入常量如 ProfileGeneric.FIFO
sort_object	object	None	排序参照对象，仅 LARGEST / SMALLEST 时需要
profile_entries	int	0	缓冲区最大容量，如 8928 （93 天 × 15 分钟）
entries_in_use	int	0	当前已有行数
access	dict	None	实例级权限（ 仅限关键字参数 ），格式 {attr_index: (AccessMode, Authentication)}

重要说明

• buffer （属性 2）带有 AttributeFlag.COMPLEX 标志， BinarySerializer 不会自动持久化 。缓冲区需单独序列化。
• capture_objects 和 sort_object 也标记为 COMPLEX，序列化器同样跳过。
• 必须 设置 on_before_read handler 来提供历史数据，否则客户端读到空缓冲区：

#使用DLMS 自定义存储 模式
import dlms

clock = dlms.Clock("0.0.1.0.0.255")
reg   = dlms.Register("1.0.1.8.0.255", 0, scaler=-3)

profile = dlms.ProfileGeneric(
    "1.0.99.1.0.255",
    capture_objects=[clock, reg],             # 裸对象 → 默认 attr=2, data=0
    capture_period=900,                       # 15 分钟自动捕获
    sort_method=dlms.ProfileGeneric.FIFO,
    profile_entries=8928,                     # 93 天 × 96 行/天
    access={
        2: (dlms.AccessMode.READ, dlms.Authentication.NONE)
    }
)

# ===== 写：自动 or 手动触发 → 存到外部 =====
def on_capture(self, event):
    if event.action != 1:                    # action=1 = capture
        return True
    row = [clock.time, reg.value]            # 拍快照
    append_row_to_flash(row)                 # 实现append_row_to_flash 写入操作
    return True

# ===== 读：客户端请求 buffer → 从外部取出 =====
def on_read_buffer(self, event):
    if event.index != self.idx('buffer'):    # 非 buffer 属性走默认
        return True
    return load_rows_from_flash()            # 实现append_row_to_flash 读取操作

profile.on_before_action = on_capture
profile.on_before_read   = on_read_buffer

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
capture_objects	list	√	每行捕获 (obj, attr_index, data_index) 三元组列表
capture_period	int	√	自动捕获间隔（秒）
sort_method	int	√	排序方式
sort_object	object	√	排序参照对象
sort_object_attribute_index	int	√	排序参照属性索引
sort_object_data_index	int	√	排序参照数据索引
profile_entries	int	√	缓冲区最大容量
entries_in_use	int	√	当前行数
buffer_size	int	×	缓冲区数据字节数（只读）
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

方法

方法	说明
deinit()	释放 C 侧资源，移除对象前调用

访问控制

import dlms
# 构造函数中指定
profile = dlms.ProfileGeneric("1.0.99.1.0.255",
    access={
        2: (dlms.AccessMode.READ, dlms.Authentication.LOW),
        1: (dlms.AccessMode.AUTHENTICATED_WRITE, dlms.Authentication.HIGH),
    })

# 或事后修改
profile.access_dict = {2: (dlms.AccessMode.READ, dlms.Authentication.LOW)}

Clock

设备的实时时钟对象，典型 OBIS 代码 0.0.1.0.0.255 。包含时区偏移和夏令时（DST）配置。C 层在每次读写 time 属性时自动与硬件 RTC 同步，无需额外 handler 即可实现基本时间管理，其 COSEM ID=8 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	time	octet-string	当前日期和时间（⏩ VOLATILE，每读自动从 RTC 刷新）
3	time_zone	int16	UTC 偏移量（分钟）
4	status	int8	时钟状态标志 （不建议手动更改）
5	begin	octet-string	DST 开始时间
6	end	octet-string	DST 结束时间
7	deviation	int8	DST 生效时的偏移量（分钟）
8	enabled	bool	DST 调整是否启用
9	base	enum	时钟源

默认 DST 配置：

参数	默认值	含义
begin	(-1, 3, -1, 2, 0, 0, 0, 0)	3 月最后一个周日 2:00
end	(-1, 10, -1, 3, 0, 0, 0, 0)	10 月最后一个周日 3:00

时间 tuple 中的 -1 表示"不指定"（wildcard），如 (-1, -1, -1, 8, 0, 0) 表示"每天 8:00"。

时钟源常量（ base 参数）

常量	值	说明
Clock.BASE_NONE	0	未指定
Clock.BASE_CRYSTAL	1	晶体振荡器
Clock.BASE_FREQUENCY_50	2	50 Hz
Clock.BASE_FREQUENCY_60	3	60 Hz
Clock.BASE_GPS	4	GPS 授时
Clock.BASE_RADIO	5	无线电授时

import dlms
# 手动设置（写完自动同步到硬件RTC）
clock.time = (2025, 8, 3, 14, 30, 0)     # 年,月,日,时,分,秒

# 客户端 GET → 自动从硬件RTC刷新后回复
# 客户端每次读到的都是准确的实时值
# Python 直接读 → 读的是C struct缓存，可能是旧值
print(clock.time)   # 如果半小时前设的，还显示半小时前

# 这个值不会自动变，硬件RTC不报存时区，需要手动填写
clock.time_zone = 480    # UTC+8（东八区），单位：分钟
clock.time_zone = -300   # UTC-5（美国东部）

# bit5 由 DST 判断逻辑自动设置/清除，不建议手动改这个位
clock.status = 0         # Clock OK
clock.status = 4         # doubtful_value = 时间不可信
clock.status = 1         # clock_ok（0和1都表示OK）

# 元组: (年, 月, 日, 时, 分, 秒, 毫秒, 星期), -1表示通配
# 第9个元素是 extra_info 标志位（可选）
# 默认值：begin=2月最后一天2:00, end=10月最后一天3:00
clock.begin = (-1, 3, -1, 2, 0, 0, 0, 0)    # DST开始 3月最后一个周日 2:00
clock.end   = (-1, 10, -1, 3, 0, 0, 0, 0)   # DST结束 10月最后一个周日 3:00

clock.deviation = 60       # DST 生效时偏移60分钟（默认）
clock.deviation = 30       # 有些地区只偏移30分钟

clock.enabled = True       # 启用夏时令自动调整
clock.enabled = False      # 禁用

#仅告诉客户端时钟源，不负责同步
clock.base = dlms.Clock.BASE_FREQUENCY_50  # 时钟源：50Hz电网频率
clock.base = dlms.Clock.BASE_FREQUENCY_60  # 时钟源：60Hz电网频率
clock.base = dlms.Clock.BASE_GPS           # 时钟源：GPS授时
clock.base = dlms.Clock.BASE_RADIO         # 时钟源：无线电授时

构造函数

Clock(logical_name: str, begin: tuple = ..., end: tuple = ..., time_zone: int = 0,
      deviation: int = 60, base: int = BASE_FREQUENCY_50, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，通常为 "0.0.1.0.0.255"
begin	tuple	见下方	DST 开始时间，默认 3 月最后一个周日 2:00
end	tuple	见下方	DST 结束时间，默认 10 月最后一个周日 3:00
time_zone	int	0	UTC 偏移量（分钟）， -480 =UTC−8（PST）， 480 =UTC+8
deviation	int	60	DST 偏移量（分钟），通常为 60
base	int	BASE_FREQUENCY_50	时钟源，可选 BASE_CRYSTAL / BASE_FREQUENCY_50 / BASE_FREQUENCY_60 / BASE_GPS / BASE_RADIO
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

方法（COSEM 动作）

方法	COSEM 方法	说明
adjust_to_quarter()	1	四舍五入到最近 15 分钟
adjust_to_minute()	3	四舍五入到整分钟
adjust_to_preset_time()	4	应用预存的目标时间
preset_adjusting_time(tuple)	5	存储目标时间供 adjust_to_preset_time() 使用
shift_time(seconds)	6	按秒数偏移（正数前进，负数后退）

C 层在每次读写 time 属性时自动与硬件 RTC 双向同步，无需 handler。事件钩子可用于审计日志等额外用途。

import dlms
# ============================================================
# 1. 初始化（含 DST 配置）
# ============================================================
clock = dlms.Clock(
    "0.0.1.0.0.255",
    time_zone=480,                       # UTC+8（东八区），单位：分钟
    deviation=60,                        # DST 生效时偏移 60 分钟
    begin=(-1, 3, -1, 2, 0, 0, 0, 0),   # DST start: 每年3月最后一个周日 2:00
    end=(-1, 10, -1, 3, 0, 0, 0, 0),    # DST end:   每年10月最后一个周日 3:00
    base=dlms.Clock.BASE_CRYSTAL,        # 告知客户端：内部晶振
    access={
        2: (dlms.AccessMode.READ, dlms.Authentication.LOW),
        1: (dlms.AccessMode.AUTHENTICATED_WRITE, dlms.Authentication.HIGH),
    }
)
# —— 中国无 DST，但保留代码作为参考 ——
clock.enabled = False

# 同步一次正确时间到硬件 RTC
import utime
rtc = utime.localtime()
clock.time = (rtc[0], rtc[1], rtc[2], rtc[3], rtc[4], rtc[5])

# 另一台设备作为 client，远程操作这台表，该例子仅供理解
client = dlms.Client(client_address=0x10, server_address=0x01, ...)
client.connect(...)
objs = client.get_objects()
clock = [o for o in objs if type(o).__name__ == "Clock"][0]

# 标准校时两步走：
client.method(clock, 5, (2026, 8, 3, 14, 30, 0))  # 先预设时间
client.method(clock, 4, None)                      # 再激活

# 或直接对齐刻钟/分钟/偏移：
client.method(clock, 1, None)   # adjust_to_quarter
client.method(clock, 3, None)   # adjust_to_minute
client.method(clock, 6, 3600)   # shift_time 快进 1 小时

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
time	tuple	√	当前日期时间 (year, month, day, hour, minute, sec) ，每读自动从 RTC 刷新
time_zone	int	√	UTC 偏移量（分钟）
status	int	√	时钟状态标志
begin	tuple	√	DST 开始时间
end	tuple	√	DST 结束时间
deviation	int	√	DST 偏移量（分钟）
enabled	bool	√	DST 是否启用
base	int	√	时钟源（见上方常量表）
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

访问控制

import dlms
# 构造函数中指定
clock = dlms.Clock("0.0.1.0.0.255",
    access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)})

# 或事后修改
clock.access_dict = {2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)}

典型用例

场景	示例
UTC+8 时区，无 DST	Clock("0.0.1.0.0.255", time_zone=480, deviation=0)
UTC−8 (PST)，美国 DST	Clock("0.0.1.0.0.255", time_zone=-480, deviation=60)
GPS 授时	Clock("0.0.1.0.0.255", base=Clock.BASE_GPS)

---

CompactData

模板驱动的紧凑编码对象。将多个捕获对象的属性值编码为一个紧凑二进制缓冲区， 省去每个值的类型标签 ，显著减少窄带链路（如 PLC、SMS）上的传输字节数，其 COSEM ID=62 。

template_description （属性 5）由 Gurux 自动维护，描述了类型模板，客户端据此解码缓冲区。 capture_objects 使用与 ProfileGeneric 相同的三元组格式。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	buffer	octet-string	紧凑编码后的原始字节（可手动写入预编码 payload）
3	capture_objects	ARRAY of STRUCT	定义捕获哪些对象的哪些属性，格式同 ProfileGeneric
4	template_id	uint8	模板标识符（0–255）
5	template_description	octet-string	类型模板描述（只读，自动维护）
6	capture_method	enum	捕获模式

捕获模式（ capture_method ）

常量	值	说明
CaptureMethod.IMPLICIT	0	每次读属性 2（buffer）时自动刷新编码
CaptureMethod.INVOKE	1	客户端调用方法 2 时才刷新编码

#例子仅用clock和register对象做演示
import dlms
clock = dlms.Clock("0.0.1.0.0.255", time_zone=480)
reg = dlms.Register("1.0.1.8.0.255", 0, scaler=-3)
cd = dlms.CompactData("0.0.96.60.0.255")

print(cd)                    #<CompactData ln='0.0.96.60.0.255', objs=2, buf=4, method=IMPLICIT>
cd.buffer = b'123456'
print(cd.buffer)             #b'123456'
cd.capture_objects = [(clock, 2, 0), (reg, 2, 0)]
print(cd.capture_objects)    #[(None, 2, 0), (None, 2, 0)]
cd.template_id=1
print(cd.template_id)        #1
cd.capture_method=dlms.CaptureMethod.IMPLICIT
print(cd.capture_method)     #IMPLICIT

构造函数

CompactData(logical_name: str, capture_objects: list = None, template_id: int = 0,
            capture_method: int = CaptureMethod.IMPLICIT, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.96.60.0.255"
capture_objects	list	None	捕获列表，每元素为 (obj, attr_idx, data_idx) 三元组
template_id	int	0	模板标识符（0–255）
capture_method	int	CaptureMethod.IMPLICIT	IMPLICIT =读时自动编码， INVOKE =客户端方法触发
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

import dlms
cd = dlms.CompactData(
    "0.0.96.60.0.255",
    capture_objects=[(clock, 2, 0), (reg, 2, 0)],
    template_id=1,
    capture_method=dlms.CaptureMethod.IMPLICIT,
    access={2: (dlms.AccessMode.READ_WRITE, dlms.Authentication.HIGH)}
)

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
buffer	bytes	√	紧凑编码后的原始字节（也可手动写入预编码 payload）
capture_objects	list	√	(obj, attr_idx, data_idx) 三元组列表
template_id	int	√	模板标识符（0–255）
template_description	bytes	×	类型模板描述（自动维护）
capture_method	int	√	捕获模式（ IMPLICIT / INVOKE ）
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

方法

方法	说明
capture(server)	手动触发捕获 （非请求上下文中）。将当前捕获对象的值编码到 buffer，并刷新 template_description 。传入一个运行中的 Server 实例。
idx(name)	返回 Python 属性名对应的 Blue Book 属性编号（继承自 CosemObject）

⚠️ 重要： on_before_read / on_before_action handler 禁止设置 event.handled = True 。handler 返回后 Gurux 必须继续执行紧凑编码。

IMPLICIT 模式 + on_before_read （推荐）

import dlms

cd = dlms.CompactData(
    "0.0.96.60.0.255",
    capture_objects=[(clock, 2, 0), (reg, 2, 0)],
    template_id=1,
    capture_method=dlms.CaptureMethod.IMPLICIT,
)

# 在编码前更新捕获对象的值（如从传感器读取）
def refresh(self, event):
    if event.index == self.idx('buffer'):
        reg.value = read_sensor()   # 在编码前刷新硬件读数
        # 不要 return False — 编码必须继续执行

cd.on_before_read = refresh

手动捕获：启动时预填 buffer

# 在 server.run() 之前手动触发一次捕获
reg.value = 12345678
cd.capture(server)  # 将当前值编码进 buffer 并更新 template_description

访问控制

# 构造函数中指定
cd = dlms.CompactData("0.0.96.60.0.255",
    access={2: (dlms.AccessMode.READ, dlms.Authentication.NONE)})

# 或事后修改
cd.access_dict = {2: (dlms.AccessMode.READ, dlms.Authentication.NONE)}

典型用例

场景	示例
窄带 PLC 上报电能和时间	CompactData("0.0.96.60.0.255", capture_objects=[(clock, 2, 0), (reg, 2, 0)], template_id=1)
SMS 推送紧凑编码	CompactData("0.0.96.60.0.255", capture_objects=[(reg_a, 2, 0), (reg_b, 2, 0)], capture_method=CaptureMethod.INVOKE)

---

DisconnectControl

可控供电开关继电器。用于远程断开/合上负载（如欠费断电、远程资产管理）。C 层默认行为只更新内存中的 output_state 和 control_state ；如需驱动真实硬件继电器，用 on_before_action 拦截，其 COSEM ID=70 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	output_state	bool	True =负载接通, False =负载断开
3	control_state	enum	0 =断开, 1 =接通, 2 =等待重连
4	control_mode	uint8	操作模式（0–6，见 Blue Book）

import dlms
disconnect_ctl = dlms.DisconnectControl("0.0.96.3.10.255")
print(disconnect_ctl.logical_name)     #输出0.0.96.3.10.255
disconnect_ctl.output_state = True     
print(disconnect_ctl.output_state)     #输出 True
disconnect_ctl.control_mode = 1 
print(disconnect_ctl.control_mode)     #输出 1

构造函数

DisconnectControl(logical_name: str, output_state: bool = False,
                  control_state: int = 0, control_mode: int = 0,
                  access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.96.3.10.255"
output_state	bool	False	初始继电器状态， True =接通, False =断开
control_state	int	0	控制状态码， 0 =断开, 1 =接通, 2 =等待重连
control_mode	int	0	操作模式（0–6），决定允许哪些手动/远程操作
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

控制状态码

值	名称	说明
0	DISCONNECTED	已断开
1	CONNECTED	已接通
2	READY_FOR_RECONNECTION	等待重连

import dlms
disconnect_ctl = dlms.DisconnectControl(
    "0.0.96.3.10.255",
    output_state=True,
    control_state=1,
    control_mode=1,
    access={
        2: (dlms.AccessMode.READ, dlms.Authentication.NONE),           
        4: (dlms.AccessMode.AUTHENTICATED_WRITE, dlms.Authentication.HIGH),
    }
    )

方法（COSEM 动作）

方法	COSEM 方法	说明
remote_disconnect()	1	远程断开负载，默认行为更新 output_state=False 、 control_state=0
remote_reconnect()	2	远程合上负载，默认行为更新 output_state=True 、 control_state=1

默认行为只更新内存中的 C 结构体。如需驱动真实硬件 GPIO，使用 on_before_action 拦截。

硬件联动： on_before_action

import dlms
from machine import Pin
# 假设 GPIO 引脚用于控制继电器
gpio_relay = Pin(Pin.GPIO1, Pin.OUT, Pin.PULL_DISABLE, 1)

def relay_handler(self, event):
    if event.index == 1:
        gpio_relay.write(0)   # remote_disconnect → 断开硬件
    elif event.index == 2:
        gpio_relay.write(1)   # remote_reconnect → 合上硬件
    return True               # 允许 C 层继续更新内存状态

disconnect_ctl = dlms.DisconnectControl(
    "0.0.96.3.10.255",
    output_state=True,
    control_state=1,
    control_mode=1,
    )
disconnect_ctl.on_before_action = relay_handler

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
output_state	bool	√	继电器状态， True =接通, False =断开
control_state	int	√	控制状态， 0 =断开, 1 =接通, 2 =等待重连
control_mode	int	√	操作模式（0–6）
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

访问控制

import dlms
# 构造函数中指定：正常遥测可读，远程通断需要 HIGH 认证
disconnect_ctl = dlms.DisconnectControl("0.0.96.3.10.255",
    output_state=True,
    control_state=1,
    control_mode=1,
    access={
        2: (dlms.AccessMode.READ, dlms.Authentication.NONE),           # output_state 可读
        4: (dlms.AccessMode.AUTHENTICATED_WRITE, dlms.Authentication.HIGH),  # control_mode 需 HIGH 认证
    })

# 或事后修改
disconnect_ctl.access_dict = {4: (dlms.AccessMode.AUTHENTICATED_WRITE, dlms.Authentication.HIGH)}

典型用例

场景	示例
欠费断电，默认接通	DisconnectControl("0.0.96.3.10.255", output_state=True, control_state=1, control_mode=1)
初始断开，等待远程合闸	DisconnectControl("0.0.96.3.10.255", output_state=False, control_state=0, control_mode=1)

---

ScriptTable

COSEM Class ID = 9，用于将"一组对 DLMS 对象的操作"打包成 脚本（Script） ，
供远程客户端通过 Execute 方法一键触发，实现断连/复电、电价切换、限载等自动化场景。

---

ScriptTable
├── Script #1 (id=1)          ← 由 add_script() 添加
│   ├── ScriptAction (Write)   → 写目标对象属性
│   └── ScriptAction (Execute) → 调用目标对象方法
├── Script #2 (id=2)
└── ...

构造函数

ScriptTable：

ScriptTable(logical_name: str, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.10.0.106.255"
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

import dlms
script_table = dlms.ScriptTable(
    "0.0.10.0.106.255",
    access={1:(dlms.AccessMode.AUTHENTICATED_WRITE,dlms.Authentication.HIGH),
    })

ScriptAction：

ScriptAction(type: int, target: object, method: int = None,
             attribute: int = None, parameter: int | bool = None)

参数	类型	默认值	说明
type	int	必传	动作类型，决定这条动作"干什么"： ScriptAction.Write (1) = 写目标对象的 属性 ； ScriptAction.Execute (2) = 调用目标对象的 方法 。见下方"两种动作类型"
target	object	必传	动作作用于的 目标 DLMS 对象实例 （ Register 、 DisconnectControl 、 Clock 等）。必须是已创建好的对象，C 层执行时会向其发 SET (Write) 或 ACTION (Execute) 请求。传非 DLMS 对象会报 TypeError
method	int	None	仅 type=Execute 时使用 （必填，不能为 0）。目标对象 方法的编号 ，如 DisconnectControl 中 1 =remote_disconnect（断开）、 2 =remote_reconnect（恢复）。具体编号含义取决于目标对象类
attribute	int	None	仅 type=Write 时使用 （必填，不能为 0）。目标对象 属性的编号 ，如 Register 中 2 =value（当前值）。执行时把 parameter 的值写入该属性
parameter	int | bool	None	动作携带的数据： Write 时 = 要写入属性的新值； Execute 时 = 传给方法调用的参数。可选，不传则动作不带数据。仅支持 int （按大小自动转为 int8/int16/int32）和 bool ，其它类型报 TypeError

ScriptAction 类型

类型	值	说明
ScriptAction.Write	1	把目标对象的指定属性设为给定值
ScriptAction.Execute	2	调用目标对象的指定方法

import dlms

tariff_reg  = dlms.Register("1.0.1.8.0.255", default_value=0)
tariff_reg1  = dlms.Register("1.0.1.8.0.255", default_value=0)
disconnect  = dlms.DisconnectControl("0.0.2.0.0.255")
disconnect1  = dlms.DisconnectControl("0.0.2.0.0.255")

script_table = dlms.ScriptTable("0.0.10.0.106.255")

action_a = dlms.ScriptAction(
    type=dlms.ScriptAction.Write,
    target=tariff_reg,
    attribute=2,
    parameter=2,          
)
action_b = dlms.ScriptAction(
    type=dlms.ScriptAction.Write,
    target=tariff_reg1,
    attribute=2,
    parameter=2,          
)

action_c = dlms.ScriptAction(
    type=dlms.ScriptAction.Execute,
    target=disconnect,
    method=2,
    parameter=0,
)

action_d = dlms.ScriptAction(
    type=dlms.ScriptAction.Execute,
    target=disconnect1,
    method=2,
    parameter=0,
)

script_table.add_script(id=3, actions=[action_a,action_c]) #关联组： ID:3对应action_a, action_c
script_table.add_script(id=2, actions=action_b)          #关联组： ID2对应action_b
script_table.add_script(id=4, actions=action_d)          #关联组： ID2对应action_b
print(script_table.get_scripts())                        # [3，2,4]
script_table.remove_script(3)                            #删除关联组：3
print(script_table.get_scripts())                        # [2,4]

#填写id为2是失败的，因为2不是设置为Execute，而4是成功的
script_table.execute_script(4)                           #True

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

方法	说明
add_script(id, actions)	添加脚本。 actions 可以是单个 ScriptAction 或 ScriptAction 列表
get_scripts()	返回配置的脚本 ID 列表
remove_script(id)	按 ID 删除脚本，成功返回 True
execute_script(id)	该接口需要在回调中进行执行，本地调用仅测试时使用
deinit()	释放 C 侧资源

on_before_action 在脚本执行前触发。返回 False 可阻止整个脚本执行；返回 True 后 C 层按序迭代所有动作。

import dlms
#该服务器例子仅供参考，服务器和客户端通信时，如何触发execute，不能正常运行
script_table = dlms.ScriptTable("0.0.10.0.106.255")
# ... 添加脚本 1、2、99 ...

def handle_script_execute(self, event):
    script_id = event.parameters
    print("[ScriptTable] Python handler: Execute script ID {}".format(script_id))

    # 场景 1：脚本 1 —— 做日志/校验后交给 C 执行
    if script_id == 1:
        print("[ScriptTable] Preparing to close disconnect control...")
        return True          # 放行，C 层按序执行动作

    # 场景 2：脚本 99 —— 纯 Python 实现，不走 C
    elif script_id == 99:
        print("[ScriptTable] Custom Python-only script")
        return False         # 阻止 C 执行

    # 场景 3：其它脚本 —— 默认放行
    return True

script_table.on_before_action = handle_script_execute

#客户端例子
import dlms
client = dlms.Client(
    client_address=0x41,
    server_address=1,
    authentication=dlms.Authentication.HIGH,
    password=b"your_password",
)
client.connect(serial)

# 获取 ScriptTable 对象
objs = client.get_objects()
st = next(o for o in objs if type(o).__name__ == "ScriptTable")

# 触发 script_id=1：发送 Action-Request，方法号=1，参数=script_id
result = client.method(st, 1, 1)   # (对象, 方法号=1固定, script_id)

---

ActivityCalendar

分层费率计划表。以"季节 → 周 → 日"的三层结构组织定时脚本执行（其 COSEM ID=20 ）：

季节 (Season)
 └── 周模板 (Week Profile)
      └── 日模板 (Day Profile)
           └── 定时动作列表: ((hour, min, sec), script_table, script_id)

日历采用 主备双版本 设计： active （活跃，对客户端可见）和 passive （待生效，在后台预先构建）。调用 activate_passive_calendar() 将 passive 版本提升为 active。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	calendar_name_active	visible-string	活跃日历名称
3–5	（season / week / day 模板，active）	complex	COSEM 复杂结构，Python 方法操作
6	calendar_name_passive	visible-string	待生效日历名称
7–9	（season / week / day 模板，passive）	complex	COSEM 复杂结构，Python 方法操作
10	activate_passive_calendar_time	octet-string	passive → active 的触发时间

构造函数

ActivityCalendar(logical_name: str, calendar_name_active: str = "",
                 calendar_name_passive: str = "", access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.13.0.0.255"
calendar_name_active	str	""	活跃日历名称
calendar_name_passive	str	""	待生效日历名称
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

通配符

常量	值	说明
dlms.ANY	-1	在时间字段中表示"任意"，如 (ANY, ANY, ANY, 8, 0, 0) 表示"每天 8:00"

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
calendar_name_active	str	√	活跃日历名称
calendar_name_passive	str	√	待生效日历名称
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

方法

import dlms
cal = dlms.ActivityCalendar("0.0.13.0.0.255",
calendar_name_passive="WinterTariff")
cal.add_day_profile(
    day_id=1,
    actions=[
        ((6, 0, 0), script_table, 1),    # 6:00 — peak starts
        ((22, 0, 0), script_table, 2),   # 22:00 — off-peak starts
    ],
    passive=True, 
)
cal.add_week_profile(
    "AllWeek",
    monday=1, tuesday=1, wednesday=1, thursday=1,
    friday=1, saturday=1, sunday=1,
    passive=True,
)
cal.add_season_profile("Winter", (1, 1, 0), "AllWeek", passive=True)
cal.activate_passive_calendar()

方法	说明
add_season_profile(name, start_time, week_name, passive=False)	添加季节。 start_time 为 (month, day, hour) 三元组
add_week_profile(name, monday, tuesday, ..., sunday, passive=False)	添加周模板，每天映射到一个 day_id
add_day_profile(day_id, actions, passive=False)	添加日模板。 actions 为 ((hour, min, sec), script_table, script_id) 元组列表
get_season_profiles(passive=False)	获取季节列表
get_week_profiles(passive=False)	获取周模板列表
get_day_profiles(passive=False)	获取日模板列表
clear_profiles(passive=False)	清空所有模板
copy_active_to_passive()	将 active 日历复制到 passive
activate_passive_calendar()	动作 1 ：将 passive 提升为 active
deinit()	释放 C 侧资源

所有构建方法默认修改 passive 日历（ passive=True ）。构建完成后调用 activate_passive_calendar() 使之生效。

访问控制

import dlms
cal = dlms.ActivityCalendar("0.0.13.0.0.255",
    calendar_name_active="SummerRates",
    calendar_name_passive="WinterRates",
    access={2: (AccessMode.READ, Authentication.NONE)})

# 或事后修改
cal.access_dict = {2: (AccessMode.READ_WRITE, Authentication.HIGH)}

---

SingleActionSchedule

单次定时动作调度器。在指定的日期/时间组合（支持通配符）触发一个 ScriptTable 中的脚本执行。C 层自动根据当前时间匹配 execution_times 列表中的表达式，匹配时调用对应的 (ScriptTable, script_id) ，其 COSEM ID=22 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	executed_script	STRUCT	要执行的 (ScriptTable, script_id) 对（⏩ COMPLEX）
3	execution_type	enum	执行类型，当前固定为 1 （Type 1：按日期/时间表达式匹配）
4	execution_times	ARRAY of STRUCT	执行时间表达式列表（⏩ COMPLEX）

构造函数

SingleActionSchedule(logical_name: str, executed_script: tuple = None,
                     execution_type: int = 1, execution_times: list = None,
                     access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.15.0.1.255"
executed_script	tuple	None	(script_table_instance, script_id) 二元组
execution_type	int	1	执行类型，固定为 1 （Type 1：日期/时间表达式）
execution_times	list	None	执行时间表达式列表，每个元素为 7 元 (year, month, day, hour, min, sec, dow)
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

执行时间表达式（7 元组）

(year, month, day, hour, minute, second, day_of_week)

位置	字段	范围	说明
0	year	任意整数 / ANY	年份
1	month	1–12 / ANY	月份
2	day	1–31 / ANY	日
3	hour	0–23 / ANY	小时
4	minute	0–59 / ANY	分钟
5	second	0–59 / ANY	秒
6	day_of_week	0–7 / ANY	0=全周, 1=周一, ..., 7=周日

任意位置使用 dlms.ANY （值为 -1 ）表示通配，如 (ANY, ANY, ANY, 2, 0, 0, ANY) = 每天 02:00:00。

from dlms import ANY
schedule = dlms.SingleActionSchedule(
    "0.0.15.0.1.255",
    executed_script=(script_table, 1),
    execution_type=1,
    execution_times=[
    (ANY, ANY, ANY, 2, 0, 0, ANY),   # every day at 02:00
    ],
)

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
executed_script	tuple	√	(script_table, script_id) 要执行的脚本
execution_type	int	√	执行类型，固定为 1
execution_times	list	√	执行时间表达式列表
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

方法

方法	说明
deinit()	释放 C 侧资源

访问控制

import dlms
schedule = dlms.SingleActionSchedule("0.0.15.0.1.255",
    executed_script=(script_table, 1),
    execution_times=[(dlms.ANY, dlms.ANY, dlms.ANY, 2, 0, 0, dlms.ANY)],
    access={2: (AccessMode.READ_WRITE, Authentication.HIGH)})

# 或事后修改
schedule.access_dict = {2: (AccessMode.READ_WRITE, Authentication.HIGH)}

典型用例

场景	示例
每天凌晨 2:00 执行抄表脚本	execution_times=[(ANY, ANY, ANY, 2, 0, 0, ANY)]
每月 1 号 0:00 执行结算脚本	execution_times=[(ANY, 1, 1, 0, 0, 0, ANY)]
每周一 8:00 上报周报	execution_times=[(ANY, ANY, ANY, 8, 0, 0, 1)]

---

RegisterMonitor

阈值监控器。持续监视某个 DLMS 对象的属性值，当值跨越预配置的阈值时自动触发 ScriptTable 中对应的脚本。服务端后台线程每秒轮询一次监视值；调用 server.monitor() 可立即触发一次检查，其 COSEM ID=21 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	thresholds	ARRAY of int	阈值列表（⏩ COMPLEX），从低到高排序
3	monitored_value	STRUCT	被监视的 (dlms_object, attribute_index) 对（⏩ COMPLEX）
4	actions	ARRAY of STRUCT	每个阈值对应一对 "up"/"down" 动作脚本（⏩ COMPLEX）

RegisterMonitor（类 21） 没有 COSEM 动作方法 ，不会触发 on_before_action / on_after_action 。

构造函数

RegisterMonitor(logical_name: str, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.16.1.0.255"
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

import dlms
rm = dlms.RegisterMonitor("0.0.16.1.0.255")
rm.thresholds = [5000, 25000] 
rm.monitored_value = (reg, reg.idx('value')) 
rm.actions = [
    {"up": (script_table, 1), "down": (script_table, 2)},
    {"up": (script_table, 1), "down": (script_table, 2)}, 
]

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
thresholds	list[int | float]	√	阈值列表，从低到高排序
monitored_value	tuple | None	√	(dlms_object, attribute_index) 被监视的目标
actions	list[dict]	√	每个阈值一条， "up" / "down" 键对应 (ScriptTable, selector) 或 None
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

方法

方法	说明
deinit()	释放 C 侧资源

工作方式

server.run() 启动后，后台线程每秒轮询一次：
  monitored_value → 当前值 = reg.value
  比较 thresholds 列表 → 找到所处的区间
  如果跨越了阈值 → 触发对应的 "up" 或 "down" 脚本

server.monitor()  → 立即手动触发一次检查（无需等待下一秒）

actions 格式

rm.actions = [
    {"up": (script_table, 1), "down": (script_table, 2)},   # 阈值 1 的动作
    {"up": (script_table, 3), "down": None},                 # 阈值 2 只有 up 动作
]
# "up"   → 值从下方跨越此阈值时触发
# "down" → 值从上方跨越此阈值时触发
# None   → 此方向不触发任何动作

访问控制

import dlms
rm = dlms.RegisterMonitor("0.0.16.1.0.255",
    access={2: (AccessMode.READ_WRITE, Authentication.HIGH)})

# 或事后修改
rm.access_dict = {2: (AccessMode.READ_WRITE, Authentication.HIGH)}

典型用例

场景	示例
低电量告警 + 过载断电	thresholds = [1000, 50000] , 低于 1000 告警 / 高于 50000 断电
欠费跳闸（单阈值）	thresholds = [0] , 余额向下跨 0 时断开
多级电价切换	thresholds = [1000, 5000, 10000] , 各阈值触发不同费率脚本

---

PushSetup

主动推送通知配置。将一组 COSEM 属性值按需编码为 DLMS DATA-NOTIFICATION PDU，由服务器主动发送到远程目标地址（如 HES 系统）。应用层负责实际传输，通常配合 MobileConnection.send() 在 relay 模式下使用，其 COSEM ID=40 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	objectList	ARRAY of STRUCT	推送的数据项列表（⏩ COMPLEX）
3	destination	visible-string	目标地址， "ip:port" 格式（⏩ COMPLEX）
4	communicationWindow	ARRAY of STRUCT	推送时间窗口列表（⏩ COMPLEX）
5	randomisationStartInterval	uint16	随机延迟区间（秒），防止多设备同时推送
6	retries	uint8	重试次数
7	retryDelay	uint32	重试间隔（秒）

构造函数

PushSetup(logical_name: str, objectList: list = None, destination: str = None,
          retries: int = 3, retryDelay: int = 60,
          randomisationStartInterval: int = 0,
          communicationWindow: list = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.25.9.0.255"
objectList	list	None	推送的数据项，每项为 (obj, attr_idx, data_idx) 三元组
destination	str	None	目标地址， "ip:port" 格式
retries	int	3	推送失败重试次数
retryDelay	int	60	重试间隔（秒）
randomisationStartInterval	int	0	随机延迟区间（秒），避免多设备集中推送
communicationWindow	list	None	推送允许时间窗口， [[start_tuple, end_tuple], ...]

import dlms
push = dlms.PushSetup(
    "0.0.25.9.0.255",
    objectList=[(reg, reg.idx('value'), 0),
        (clock, clock.idx('time'), 0)],
    destination="203.0.113.1:4060",
    retries=3,
    retryDelay=60, ) 
push.communicationWindow = [  #communicationWindow 格式
    [(-1, -1, -1, 6, 0, 0), (-1, -1, -1, 22, 0, 0)],  # 每天 06:00–22:00
]
# start/end 为 6 元组 (year, month, day, hour, min, sec)
# -1 表示 wildcard，如 (-1, -1, -1, 6, 0, 0) = 每天 06:00
def do_push(self, event):
    if event.index == 1:
        pdu = self.generate_pdu()
        mobile_conn.send(pdu)
    return True

push.on_before_action = do_push

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
objectList	list	√	(obj, attr_idx, data_idx) 三元组列表
destination	str	√	目标 "ip:port" 地址
retries	int	√	重试次数
retryDelay	int	√	重试间隔（秒）
randomisationStartInterval	int	√	随机延迟区间（秒）
communicationWindow	list	√	推送时间窗口 [[start, end], ...]
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
on_before_action	Callable	√	动作前钩子（继承自 CosemObject）
on_after_action	Callable	√	动作后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

方法

方法	说明
generate_pdu()	编码当前属性值为 DLMS DATA-NOTIFICATION PDU，返回 bytes
deinit()	释放 C 侧资源

生成推送流程

客户端调用 COSEM 方法 1（push）
        │
        ▼
on_before_action(self, event)  ← 你的 handler
        │
        ├── self.generate_pdu() →  bytes（DATA-NOTIFICATION PDU）
        │
        └── mobile_conn.send(pdu) → 通过 relay UDP 发出

generate_pdu() 做了 3 件事：

1. 读取 objectList 中所有对象的当前属性值
2. 按 DLMS 结构格式序列化
3. 包装为 DATA-NOTIFICATION 命令，返回 bytes

你的 handler 只负责传输： 调用 generate_pdu() 拿数据 → mobile_conn.send(pdu) 发出。

典型用例

场景	示例
定期上报电能和时间	objectList=[(reg, 2, 0), (clock, 2, 0)] , destination="203.0.113.1:4060"
仅在白天推送	communicationWindow=[[(-1,-1,-1,6,0,0), (-1,-1,-1,22,0,0)]]
告警主动推送	objectList=[(event_code, 2, 0)] , retries=5

---

GsmDiagnostic

蜂窝网络实时诊断对象。DLMS 客户端可远程读取设备连接状态。调用 update() 从 QuecPython net 模块刷新所有字段；配合 on_before_read 可在每次客户端读取前自动刷新，其 COSEM ID=47，OBIS 0.0.25.6.0.255 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	operator_name	visible-string	运营商名称，如 "A1 Srbija" ，不可用时为 None
3	status	enum	网络注册状态（⏩ VOLATILE）
4	circuit_switch_status	enum	电路交换连接状态（⏩ VOLATILE）
5	packet_switch_status	enum	数据技术类型（⏩ VOLATILE）
6	cell_info	STRUCT	服务小区详细信息（⏩ COMPLEX + VOLATILE）

构造函数

GsmDiagnostic(logical_name: str, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，通常为 "0.0.25.6.0.255"
access	dict	None	实例级权限，格式 {attr_index: (AccessMode, Authentication)}

注册状态常量（ status ）

值	常量	说明
0	NOT_REGISTERED	未注册到任何网络
1	HOME_NETWORK	已注册到归属网络
2	SEARCHING	正在搜索网络
3	DENIED	注册被拒绝
4	UNKNOWN	状态未知
5	ROAMING	漫游中

数据技术常量（ packet_switch_status ）

值	说明
0	INACTIVE
1	GPRS
2	EGPRS
3	UMTS
4	HSDPA
5	LTE

方法

方法	说明
update()	调用 QuecPython net 模块（ net.getCellInfo() / net.getSignal() / net.operatorName() / net.getState() ）刷新所有属性。网络不可用时仅打印警告，不抛异常

import dlms
gsm = dlms.GsmDiagnostic("0.0.25.6.0.255")

def refresh_gsm(self, event):
    self.update()

gsm.on_before_read = refresh_gsm

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
operator_name	str | None	√	运营商名称
status	int	√	网络注册状态（见状态常量表）
circuit_switch_status	int	√	电路交换状态
packet_switch_status	int	√	数据技术类型（见数据常量表）
cell_info	GsmCellInfo	×	服务小区详情对象（只读，由 update() 更新）
adjacent_cells	list[AdjacentCell]	×	邻小区列表（只读，由 update() 更新）
adjacent_cells_count	int	×	邻小区数量
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
access_dict	dict	√	实例级访问控制

GsmCellInfo — 服务小区详情

属性	类型	说明
cell_id	int	小区标识（CID）
location_id	int	位置区码（LAC）或跟踪区码（TAC for LTE）
signal_quality	int	接收信号强度（dBm，通常为负值）
ber	int	误码率等级（0–7，GSM 05.08）
mobile_country_code	int	移动国家码（MCC，如 220 表示塞尔维亚）
mobile_network_code	int	移动网络码（MNC）
channel_number	int	ARFCN / UARFCN / EARFCN

AdjacentCell — 邻小区

属性	类型	说明
cell_id	int	邻小区标识
signal_quality	int	接收信号强度（dBm）

GsmDiagnostic 上同时暴露旧式快捷访问属性（ cell_id 、 location_id 、 signal_quality 、 ber 、 mobile_country_code 、 mobile_network_code 、 channel_number ），它们直接映射到 cell_info 的同名字段。

访问控制

import dlms
gsm = dlms.GsmDiagnostic("0.0.25.6.0.255",
    access={2: (AccessMode.AUTHENTICATED_READ, Authentication.HIGH)})

# 或事后修改
gsm.access_dict = {2: (AccessMode.AUTHENTICATED_READ, Authentication.HIGH)}

典型用例

场景	示例
远程诊断信号强度	gsm.update(); print(gsm.cell_info.signal_quality)
判断是否在归属网络	if gsm.status == 1: print("Home")
扫描邻小区数量	print(gsm.adjacent_cells_count)

---

SecuritySetup

安全配置对象，管理 DLMS 端到端加密和认证所需的密钥、系统标题、证书及策略配置。用于 AssociationLogicalName 或 AssociationShortName 中以启用加密通信，其 COSEM ID=64 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	server_system_title	octet-string(8)	服务端系统标题（8 字节）
3	client_system_title	octet-string(8)	客户端系统标题（8 字节）
4	security_policy	enum	安全策略（见 SecurityPolicy 表）
5	security_suite	enum	安全套件版本（0/1/2）
6	certificates	ARRAY of STRUCT	证书列表（⏩ COMPLEX）
7	min_invocation_counter	uint32	最低调用计数器
8	gak	octet-string	全局认证密钥（16 或 32 字节）
9	guek	octet-string	全局单播加密密钥（Block Cipher Key，16 或 32 字节）
10	gbek	octet-string	全局广播加密密钥

构造函数

SecuritySetup(logical_name: str)

参数	类型	说明
logical_name	str	OBIS 代码字符串，如 "0.0.43.0.1.255"

SecurityPolicy 常量

常量	值	说明
SecurityPolicy.NOTHING	0	无保护
SecurityPolicy.AUTHENTICATED	1	仅认证
SecurityPolicy.ENCRYPTED	2	仅加密
SecurityPolicy.AUTHENTICATED_ENCRYPTED	3	认证+加密（HighGMac 用此值）

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
security_policy	int	√	安全策略（SecurityPolicy 枚举）
security_suite	int	√	安全套件版本（0/1/2）
min_invocation_counter	int	√	最低调用计数器
server_system_title	bytes	√	服务端系统标题（8 字节）
client_system_title	bytes	√	客户端系统标题（8 字节）
guek	bytes	√	全局单播加密密钥
gak	bytes	√	全局认证密钥
gbek	bytes	√	全局广播加密密钥
certificates	list	√	证书列表，每项为 dict
on_before_read	Callable	√	读前钩子（继承自 CosemObject）
on_after_read	Callable	√	读后钩子（继承自 CosemObject）
on_before_write	Callable	√	写前钩子（继承自 CosemObject）
on_after_write	Callable	√	写后钩子（继承自 CosemObject）
access_dict	dict		实例级访问控制

方法

方法	说明
deinit()	释放 C 侧资源

访问控制

import dlms
# 构造函数中指定
sec = dlms.SecuritySetup("0.0.43.0.1.255")

# 事后修改
sec.access_dict = {
    4: (AccessMode.AUTHENTICATED_WRITE, Authentication.HIGH),  # security_policy
    8: (AccessMode.AUTHENTICATED_WRITE, Authentication.HIGH),  # gak
    9: (AccessMode.AUTHENTICATED_WRITE, Authentication.HIGH),  # guek
}

网络传输对象

以下对象配置 DLMS 通信栈： IecHdlcSetup （HDLC 帧参数）、 LocalPortSetup （光口 Mode E 协商）、 GprsSetup （蜂窝 APN）、 IPv4Setup （IP 地址）、 TcpUdpSetup （TCP/UDP 端口）、 MacAddressSetup （MAC 地址）。

---

IecHdlcSetup

HDLC（IEC 62056-46）链路层配置。控制 UART 上的帧窗口大小、分片大小及超时。与 SerialConnection 配合使用，其 COSEM ID = 23，OBIS 0.0.22.0.0.255 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码
2	comm_speed	enum	通信速率（baud）
3	window_size_rx	uint8	接收窗口大小
4	window_size_tx	uint8	发送窗口大小
5	max_info_len_tx	uint16	发送最大信息长度
6	max_info_len_rx	uint16	接收最大信息长度
7	inactivity_timeout	uint16	非活动超时（秒）
8	device_address	uint16	设备地址（默认 0x10 ）

构造函数

IecHdlcSetup(logical_name: str, commSpeed: int = 9600,
             windowSizeRx: int = 1, windowSizeTx: int = 1,
             maxInfoLenTx: int = 128, maxInfoLenRx: int = 128,
             timeout: int = 120, deviceAddr: int = 0x10)

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码
commSpeed	int	√	波特率，如 9600
windowSizeRx	int	√	接收窗口大小
windowSizeTx	int	√	发送窗口大小
maxInfoLenTx	int	√	发送最大信息帧长
maxInfoLenRx	int	√	接收最大信息帧长
timeout	int	√	非活动超时（秒）
deviceAddr	int	√	HDLC 设备地址

典型用例

hdlc = dlms.IecHdlcSetup(
    "0.0.22.0.0.255",
    commSpeed=9600,
    windowSizeRx=1, windowSizeTx=1,
    maxInfoLenTx=128, maxInfoLenRx=128,
    deviceAddr=0x10,
)
conn = dlms.SerialConnection(uart_port=2, hdlc_setup=hdlc, flowcontrol=0)

---

LocalPortSetup

IEC 62056-21 Mode E 光口协议配置。实现 300 bps（7E1）Mode E 协商后自动切换到高速模式，其 COSEM ID = 19 。

Blue Book 属性

编号	名称	说明
1	logical_name	OBIS 代码
2	default_mode	默认光口协议模式（DLMS_OPTICAL_PROTOCOL_MODE）
3	default_baud	默认波特率（典型 300 bps）
4	proposed_baud	协商后切换的目标波特率（9600/19200）
5	response_time	响应超时（ms）
6	device_address	设备地址
7	password_1	P1 密码（最低安全级）
8	password_2	P2 密码（中等安全级）
9	password_5	P5 密码（最高安全级）

构造函数

LocalPortSetup(logical_name: str, default_mode: int = 0,
               default_baud: int = 300, proposed_baud: int = 9600,
               response_time: int = 1000, device_address: bytes = None,
               password_1: bytes = None, password_2: bytes = None,
               password_5: bytes = None)

典型用例

local_port = dlms.LocalPortSetup(
    "0.0.19.0.0.255",
    default_baud=300,
    proposed_baud=9600,
    device_address=b"MTR001",
    password_1=b"00000000",
)
hdlc = dlms.IecHdlcSetup("0.0.22.0.0.255", commSpeed=9600, deviceAddr=0x10)
conn = dlms.OpticalConnection(uart_port=1, local_port_setup=local_port, hdlc_setup=hdlc)

---

GprsSetup

蜂窝/GPRS 网络接入点配置，其 COSEM ID = 45 。

Blue Book 属性

编号	名称	类型	说明
2	apn	visible-string	接入点名称
3	pin_code	uint16	SIM PIN 码（0=无 PIN）

构造函数

GprsSetup(logical_name: str, apn: str = "", pin_code: int = 0)

Python 属性一览

属性	类型	可写	说明
apn	str	√	APN 名称
pin_code	int	√	SIM PIN 码

典型用例

gprs = dlms.GprsSetup("0.0.25.0.0.255", apn="internet", pin_code=0)
gprs.apn = "m2m.carrier.net"

---

IPv4Setup

IPv4 地址配置（静态或 DHCP），其 COSEM ID = 42 。

Blue Book 属性

编号	名称	说明
2	data_link_reference	引用 GprsSetup/MacAddressSetup
3	ip_address	IP 地址
4	multicast_ip_address	多播地址（ "0.0.0.0" =DHCP）
5	subnet_mask	子网掩码（⏩ COMPLEX）
6	gateway_ip_address	默认网关
7	use_dhcp	是否启用 DHCP
8	primary_dns_address	主 DNS
9	secondary_dns_address	备 DNS

构造函数

IPv4Setup(logical_name: str, datalink_reference: object = None,
          ip_address: str = "0.0.0.0", subnet_mask: str = "255.255.255.0",
          gateway_ip_address: str = "0.0.0.0", use_dhcp: bool = True,
          primary_dns_address: str = "0.0.0.0",
          secondary_dns_address: str = "0.0.0.0")

典型用例

ipv4 = dlms.IPv4Setup(
    "0.0.25.1.0.255",
    datalink_reference=gprs,
    ip_address="0.0.0.0",
    use_dhcp=True,
    primary_dns_address="8.8.8.8",
    secondary_dns_address="8.8.4.4",
)

---

TcpUdpSetup

TCP/UDP 端口配置（DLMS 默认端口 4059），其 COSEM ID = 41 。

Blue Book 属性

编号	名称	说明
2	tcp_udp_port	TCP/UDP 端口号
3	ip_reference	引用 IPv4Setup
4	maximum_simultaneous_connections	最大并发连接数
5	inactivity_timeout	非活动超时（秒，0=禁用）

构造函数

TcpUdpSetup(logical_name: str, port: int = 4059,
            ip_reference: object = None, max_simultaneous_connections: int = 1,
            inactivity_timeout: int = 0)

Python 属性一览

属性	类型	可写	说明
port	int	√	端口号
ip_reference	IPv4Setup | None	√	IP 配置引用
max_simultaneous_connections	int	√	最大并发连接
inactivity_timeout	int	√	非活动超时（秒）

典型用例

tcp_udp = dlms.TcpUdpSetup(
    "0.0.25.2.0.255",
    port=4059,
    ip_reference=ipv4,
    max_simultaneous_connections=1,
    inactivity_timeout=120,
)

---

MacAddressSetup

以太网/蜂窝 MAC 地址配置，其 COSEM ID = 43 。

Blue Book 属性

编号	名称	说明
2	mac_address	6 字节 MAC 地址

构造函数

MacAddressSetup(logical_name: str, mac_address: bytes = None)

典型用例

mac = dlms.MacAddressSetup("0.0.25.0.0.255", mac_address=b'\x00\x11\x22\x33\x44\x55')

---

M-Bus 对象

以下是 M-Bus（Meter-Bus，EN 13757）相关的所有 COSEM 接口对象。 重点： dlms 模块只提供 COSEM 属性结构供 DLMS 客户端读取配置和测量值， 不实现 M-Bus 传输本身 。应用层负责与 M-Bus 设备通信（如通过 GenericConnection 包装连接到 M-Bus 收发器的 UART），并将读取到的数据填充到这些 COSEM 对象中。

---

MbusSlavePortSetup

当 QuecPython 设备 作为 M-Bus 从站/仪表 时使用。暴露设备的 M-Bus 端口参数给 DLMS 客户端（主站/集中器）读取，其 COSEM ID = 25 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码
2	default_baud	enum	默认 M-Bus 波特率
3	available_baud	enum	当前可用波特率
4	address_state	enum	地址分配状态
5	bus_address	uint8	M-Bus 主站地址（0–255，有效从站范围 1–250）

构造函数

MbusSlavePortSetup(logical_name: str, access: dict = None)

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码
default_baud	int	√	默认波特率（300/600/1200/2400/4800/9600/19200/38400/57600/115200）
available_baud	int	√	当前可用波特率
address_state	int	√	AddressState.NONE (0) 或 AddressState.ASSIGNED (1)
bus_address	int	√	从站主站地址（1–250）

⚠️ 没有 COSEM 动作方法， on_before_action / on_after_action 永不触发。

典型用例

slave_port = dlms.MbusSlavePortSetup("0.0.24.9.0.255")
slave_port.default_baud   = 9600
slave_port.available_baud = 9600
slave_port.address_state  = dlms.AddressState.ASSIGNED
slave_port.bus_address    = 1

# 读前自动刷新地址状态：
def on_read(self, event):
    slave_port.address_state = dlms.AddressState.ASSIGNED if get_mbus_assigned() else dlms.AddressState.NONE

slave_port.on_before_read = on_read

---

MbusMasterPortSetup

当设备 作为 M-Bus 主站/集中器 时使用，仅暴露通信速率给 DLMS 客户端，其 COSEM ID = 74 。

Blue Book 属性（仅一个）

属性	类型	说明
comm_speed	int	通信速率（300/600/…/115200）

构造函数

MbusMasterPortSetup(logical_name: str, access: dict = None)

⚠️ 没有 COSEM 动作方法。

典型用例

mbus_master = dlms.MbusMasterPortSetup("0.0.24.3.0.255")
mbus_master.comm_speed = 9600

---

MbusPortSetup

描述主站的某个 M-Bus 通信端口详情：主站地址、监听窗口、标识字段等，其 COSEM ID = 76 。

Blue Book 属性

编号	名称	说明
1	logical_name	OBIS 代码
2	profile_selection	关联的 ProfileGeneric OBIS（⏩ COMPLEX）
3	port_communication_status	端口状态
4	data_header_type	数据头类型
5	primary_address	主站地址（0–255）
6	identification_number	标识号（只读）
7	manufacturer_id	厂家 ID（2 字节，只读）
8	mbus_version	协议版本（只读）
9	device_type	设备类型（只读）
10	max_pdu_size	最大 PDU 大小
11	listening_window	监听窗口列表 [[start_tuple, end_tuple], ...] （⏩ COMPLEX）

构造函数

MbusPortSetup(logical_name: str, access: dict = None)

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码
profile_selection	str	√	关联的 Profile OBIS
port_communication_status	int	√	端口状态
data_header_type	int	√	数据头类型
primary_address	int	√	主站地址（0–255）
identification_number	int	×	标识号
manufacturer_id	int	×	厂家 ID
mbus_version	int	×	协议版本
device_type	int	×	设备类型
max_pdu_size	int	√	最大 PDU 大小
listening_window	list	√	[[start, end], ...] ，时间使用 dlms.ANY 通配

⚠️ 没有 COSEM 动作方法。

典型用例

from dlms import ANY

port = dlms.MbusPortSetup("0.0.24.7.0.255")
port.primary_address = 1
port.listening_window = [
    [(ANY, ANY, ANY, 8, 0, 0), (ANY, ANY, ANY, 18, 0, 0)]  # 每天 8:00–18:00
]

---

MbusClient

代表总线上的一台从站仪表。通过 mbus_port 链接到 MbusPortSetup ，存储标识信息和状态告警字节。从 M-Bus 传输层读到报文后应在下一次 DLMS 客户端轮询前更新此对象属性，其 COSEM ID = 72 。

Blue Book 属性

编号	名称	说明
1	logical_name	OBIS 代码
2	mbus_port	引用的 MbusPortSetup
3	capture_definition	捕获定义（⏩ COMPLEX）
4	capture_period	捕获间隔（秒）
5	primary_address	主站地址
6	identification_number	标识号（只读）
7	manufacturer_id	厂家 ID（只读）
8	mbus_version	协议版本（只读）
9	device_type	设备类型（只读）
10	access_number	访问计数（只读，⏩ VOLATILE）
11	status	仪表状态字节（只读，⏩ VOLATILE）
12	alarm	仪表告警字节（只读，⏩ VOLATILE）
13	configuration	配置字（2 字节）
14	encryption_key_status	加密密钥状态

构造函数

MbusClient(logical_name: str, mbus_port: object = None, access: dict = None)

参数	说明
logical_name	OBIS 代码
mbus_port	MbusPortSetup 对象引用

方法（全部通过 on_before_action 派发）

序号	方法名	说明
1	slave_install	安装从站
2	slave_deinstall	卸载从站
3	capture	捕获数据（触发测量值更新）
4	reset_alarm	清零告警
5	synchronize_clock	同步时钟
6	data_send	发送数据
7	set_encryption_key	设置加密密钥
8	transfer_key	传输密钥

所有方法需要 on_before_action handler 实现实际的 M-Bus 物理通信。

典型用例

port = dlms.MbusPortSetup("0.0.24.7.0.255")
client = dlms.MbusClient("0.0.24.1.0.255", mbus_port=port)
client.capture_period = 900   # 15 分钟
client.device_type = 3        # 燃气表

# 捕获方法由 on_before_action handler 实现
def on_capture(self, event):
    if event.index == 3:  # capture
        # 此处应发送 M-Bus 报文->读取数据->更新 client 属性
        update_client_data(self)

client.on_before_action = on_capture

---

MbusDiagnostic

M-Bus 通道链路质量和帧计数器监控，其 COSEM ID = 77 。

Blue Book 属性

编号	名称	说明
1	logical_name	OBIS 代码
2	received_signal_strength	接收信号强度 dBµV（⏩ VOLATILE）
3	channel_id	通道标识（0–255）
4	link_status	链路状态（⏩ VOLATILE）
5	broadcast_frames	广播帧列表（⏩ COMPLEX）
6	transmissions	发送帧总数（⏩ VOLATILE）
7	received_frames	成功接收帧数（⏩ VOLATILE）
8	failed_received_frames	错误帧数（⏩ VOLATILE）
9	capture_time	最近捕获的属性及时间戳（⏩ COMPLEX）

构造函数

MbusDiagnostic(logical_name: str, access: dict = None)

方法

方法	COSEM 方法	说明
reset()	1	清零所有计数器。默认更新内存中的 C 值；如需硬件级清零，用 on_before_action

典型用例

diag = dlms.MbusDiagnostic("0.0.24.8.0.255")
diag.received_signal_strength = 120
diag.link_status = 1   # 链路正常

# 硬件级计数器复位
def on_reset(self, event):
    if event.index == 1:
        reset_hardware_counters()

diag.on_before_action = on_reset

---

G3-PLC 对象（ITU-T G.9903）

以下三个对象提供 G3-PLC 窄带电力线通信的 COSEM 属性结构。 重点： 与 M-Bus 对象一样，这些目前为纯数据持有者—— dlms 模块暴露 COSEM 属性结构供 DLMS 客户端读取配置和统计数据， 不实现 G3-PLC 传输本身 。应用层负责通过 GenericConnection 挂接 G3-PLC 调制解调器进行物理层通信，并将从传输层获取的数据填充到这些 COSEM 对象中。

注意： 当前版本中 G3-PLC 动作方法没有 C 层派发，所有动作必须通过 on_before_action handler 实现。

---

G3PlcMacCounters

单设备 MAC 层包统计计数器。所有计数器均为无符号 32 位整数。 方法 1（reset）： 清零所有计数器；必须通过 on_before_action handler 实现（无 C 层派发），其 COSEM ID = 90 。

Blue Book 属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码
2	tx_data_packet_count	uint32	发送数据包总数
3	rx_data_packet_count	uint32	接收数据包总数
4	tx_cmd_packet_count	uint32	发送命令包总数
5	rx_cmd_packet_count	uint32	接收命令包总数
6	csma_fail_count	uint32	CSMA 失败次数
7	csma_no_ack_count	uint32	CSMA 未收到 ACK 次数
8	bad_crc_count	uint32	坏 CRC 帧数
9	tx_data_broadcast_count	uint32	发送广播数据包数
10	rx_data_broadcast_count	uint32	接收广播数据包数

构造函数

G3PlcMacCounters(logical_name: str, access: dict = None)

Python 属性一览

属性	类型	可写
logical_name	bytes	×
tx_data_packet_count	int	√
rx_data_packet_count	int	√
tx_cmd_packet_count	int	√
rx_cmd_packet_count	int	√
csma_fail_count	int	√
csma_no_ack_count	int	√
bad_crc_count	int	√
tx_data_broadcast_count	int	√
rx_data_broadcast_count	int	√

COMS 方法：方法 1 = reset

典型用例

import dlms

counters = dlms.G3PlcMacCounters("0.0.29.1.0.255")

# 从 G3-PLC 调制解调器更新计数器
counters.tx_data_packet_count = 100
counters.rx_data_packet_count = 95

# 方法 1（reset）必须通过 on_before_action 实现
def reset_counters(self, event):
    if event.index == 1:
        self.tx_data_packet_count = 0
        self.rx_data_packet_count = 0
        self.tx_cmd_packet_count = 0
        self.rx_cmd_packet_count = 0
        self.csma_fail_count = 0
        self.csma_no_ack_count = 0
        self.bad_crc_count = 0
        self.tx_data_broadcast_count = 0
        self.rx_data_broadcast_count = 0
        # 此处可请求 G3-PLC 调制解调器执行硬件级计数器复位
    return True

counters.on_before_action = reset_counters

---

G3PlcMacSetup

G3-PLC MAC 层完整配置：包括短地址、CSMA 参数、邻区表、tone mask 等约 25 个参数。 方法 1（get_neighbour_table）： 必须通过 on_before_action handler 实现，其 COSEM ID = 91 。

Blue Book 核心属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码
2	short_address	uint16	本节点 MAC 短地址
3	rc_coord	uint8	到协调器的路由成本
4	pan_id	uint16	PAN 标识符
5	key_table	ARRAY	密钥表 [{"id": int, "key": bytes(16)}, ...] （⏩ COMPLEX）
6	frame_counter	uint32	帧计数器
7	tone_mask	bytes	活跃子载波 packed bit-array
8	tmr_ttl	uint8	TMR 生命周期
9	max_frame_retries	uint8	最大帧重传次数
10	neighbour_table_entry_ttl	uint8	邻区条目生命周期（秒）
11	neighbour_table	ARRAY	邻区表 [{short_address, lqi, valid_time, ...}, ...] （⏩ COMPLEX）
12–26	（CSMA / beacon / 衰减参数）	—	约 15 个微调参数，详见 Blue Book

构造函数

G3PlcMacSetup(logical_name: str, access: dict = None)

Python 属性一览（核心属性）

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码
short_address	int	√	本节点 MAC 短地址
rc_coord	int	√	到协调器路由成本
pan_id	int	√	PAN 标识符
key_table	list[dict]	√	[{"id": int, "key": bytes(16)}]
frame_counter	int	√	帧计数器
tone_mask	bytes	√	子载波 packed 位数组
neighbour_table	list[dict]	√	[{short_address, lqi, valid_time, ...}]
max_frame_retries	int	√	最大帧重传
min_be / max_be	int	√	最小/最大退避指数
max_csma_backoffs	int	√	最大 CSMA 退避次数

邻区表应在客户端读取前从 G3-PLC 调制解调器刷新。

典型用例

import dlms

mac = dlms.G3PlcMacSetup("0.0.29.0.0.255")
mac.short_address = 0x1234
mac.pan_id = 0x5678
mac.tone_mask = b'\xFF\xFF...'  # 子载波 tone mask

# 客户端读前从调制解调器刷新邻区表
def refresh_mac(self, event):
    if event.index == self.idx('neighbour_table'):
        # 从 G3-PLC 传输层获取最新邻区表
        self.neighbour_table = get_neighbour_table_from_modem()
    return True

mac.on_before_read = refresh_mac

# 方法 1（get_neighbour_table）通过 on_before_action 实现
def get_neighbour_table_action(self, event):
    if event.index == 1:
        refresh_mac(self, event)  # 复用刷新逻辑
    return True

mac.on_before_action = get_neighbour_table_action

---

G3Plc6LoWPAN

6LoWPAN/LOADng 路由适配层配置：包括路由表、黑名单、广播日志、上下文信息表、组表及相关阈值参数。 类 92 没有 COSEM 动作方法； 路由表数据应在 on_before_read 中按需从传输层刷新,其 COSEM ID = 92 。

Blue Book 核心属性

编号	名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码
2	max_hops	uint8	最大 LOADng 路由跳数
3	weak_lqi_value	uint8	"弱链路"的 LQI 阈值
4	security_level	uint8	适配帧最低安全等级
5	prefix_table	bytes	PAN 前缀列表（⏩ COMPLEX）
6	routing_configuration	ARRAY	LOADng 路由参数（⏩ COMPLEX）
7	broadcast_log_table_entry_ttl	uint16	广播日志 TTL（分钟）
8	routing_table	ARRAY	LOADng 路由表 [{destination, next_hop, cost, ...}] （⏩ COMPLEX）
9	context_information_table	ARRAY	6LoWPAN 上下文信息（⏩ COMPLEX）
10	blacklist_table	ARRAY	黑名单邻区（⏩ COMPLEX）
11	broadcast_log_table	ARRAY	广播日志（⏩ COMPLEX）
12	group_table	ARRAY	本设备注册的组地址（⏩ COMPLEX）
13–23	（join / path / LQI / 路由开启标志）	—	其他 LOADng 配置参数

构造函数

G3Plc6LoWPAN(logical_name: str, access: dict = None)

⚠️ 类 92 没有 COSEM 动作方法， on_before_action / on_after_action 永不触发。

典型用例

import dlms

lowpan = dlms.G3Plc6LoWPAN("0.0.29.2.0.255")
lowpan.max_hops = 8
lowpan.max_join_wait_time = 60
lowpan.metric_type = 1  # LOADng 路由度量类型

# 客户端读前从 G3-PLC 传输层刷新路由表
def refresh_lowpan(self, event):
    if event.index == self.idx('routing_table'):
        self.routing_table = get_routing_table_from_modem()
    elif event.index == self.idx('blacklist_table'):
        self.blacklist_table = get_blacklist_from_modem()
    elif event.index == self.idx('broadcast_log_table'):
        self.broadcast_log_table = get_broadcast_log_from_modem()
    return True

lowpan.on_before_read = refresh_lowpan

---

预付费子系统（Prepayment）

预付费子系统由四个协作类组成：

类 ID	类名	OBIS	角色
111	Account	0.0.19.0.0.255	顶层控制器，串联 Credit / Charge / TokenGateway
112	Credit	0.0.19.10.0.255	信用额度（余额）管理
113	Charge	0.0.19.20.0.255	基于消费量的费用计算
115	TokenGateway	0.0.19.40.0.255	Token 入口、验证与执行

重要 ：这四个类 均无 C 端动作调度 。所有方法（method）处理必须通过 on_before_action 回调在 Python 侧实现。

---

数据流概述

客户端/键盘 → TokenGateway.enter(token)
                         ↓
              on_before_action 验证 token
                         ↓
              TokenGateway.token / time / status 更新
                         ↓
              Account 根据 token_gateway_configurations
              按比例分配额度到对应 Credit
                         ↓
              Credit.current_credit_amount 增加
              Credit.status 更新
                         ↓
              Account 重算 available_credit /
              current_credit_status
                         ↓
              Charge.collect() → 根据 tariff 扣除 Credit 余额

---

TokenGateway

Token 网关，管理预付费 Token 的录入、验证和执行。OBIS 0.0.19.40.0.255 ，其 COSEM ID = 115 。

Blue Book 属性

编号	Python 名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	token	bytes	最后一次接受的 token 原始数据
3	time	tuple(6)	token 处理时间 (年,月,日,时,分,秒)
4	descriptions	list[str]	token 描述列表（每条描述是一个字符串）
5	delivery_method	TokenDelivery	Token 投递方式（REMOTE / LOCAL / MANUAL）
6	status	TokenStatusCode	Token 处理状态码
7	data_value	bytes	Bit 数组形式的附加数据

方法（均需通过 on_before_action 实现）

编号	名称	说明
1	enter	录入 token
2	verify	验证 token
3	execute	执行 token

构造函数

TokenGateway(logical_name: str, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.19.40.0.255"
access	dict	None	实例级权限，格式 {auth: {attr_index: AccessMode}}

基本用法

import dlms

gateway = dlms.TokenGateway("0.0.19.40.0.255")

# 配置投递方式
gateway.delivery_method = dlms.TokenDelivery.REMOTE

# 录入 token（需通过 on_before_action 实现验证逻辑）
# gateway.on_before_action = my_token_handler

# 读取 token 处理结果
token_data = gateway.token       # bytes
proc_time = gateway.time         # (year, month, day, hour, min, sec)
proc_status = gateway.status     # TokenStatusCode 枚举值

资源清理

gateway.deinit()  # 释放 C 堆上的 token 和 descriptions 内存

``on_before_action — 方法处理

所有三个方法均需通过 on_before_action 实现：

import dlms

def on_token_action(self, event):
    """TokenGateway 方法处理"""
    if event.index == 1:  # enter
        raw_token = self.token
        # ...验证逻辑...
        if valid:
            self.status = dlms.TokenStatusCode.VALIDATION_OK
            self.time = (2025, 6, 15, 10, 30, 0)
        else:
            self.status = dlms.TokenStatusCode.AUTHENTICATION_FAILURE
    elif event.index == 2:  # verify
        # ...验证逻辑...
        pass
    elif event.index == 3:  # execute
        # ...执行逻辑，更新对应 Credit...
        pass
    return True

gateway = dlms.TokenGateway("0.0.19.40.0.255")
gateway.on_before_action = on_token_action

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
token	bytes	√	最后一次接受的 token 数据
time	tuple(6)	√	token 处理时间戳
descriptions	list[str]	√	token 描述字符串列表
delivery_method	TokenDelivery	√	投递方式枚举
status	TokenStatusCode	√	处理状态枚举
data_value	bytes	√	附加 bit 数据
on_before_read	Callable	√	读前钩子（继承 CosemObject）
on_after_read	Callable	√	读后钩子
on_before_write	Callable	√	写前钩子
on_after_write	Callable	√	写后钩子
on_before_action	Callable	√	动作前钩子（ 所有方法实现入口 ）
on_after_action	Callable	√	动作后钩子
access_dict	dict	√	实例级访问控制

---

Credit

信用额度管理。持有单个信用余额及其配置。OBIS 0.0.19.10.0.255 ，其 COSEM ID = 112 。

Blue Book 属性

编号	Python 名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	current_credit_amount	int32	当前信用额度余额
3	type	CreditType	信用类型（Token / Emergency / TimeBased / ConsumptionBased）
4	priority	uint8	优先级（值越低越优先消耗）
5	warning_threshold	int32	警告阈值
6	limit	int32	限制值（可为负数，表示债务空间）
7	credit_configuration	CreditConfiguration	配置位掩码
8	status	uint8	状态（对应 CreditStatus 值）
9	preset_credit_amount	int32	预设信用额度
10	credit_available_threshold	int32	信用可用阈值
11	period	tuple(6)	周期时间 (年,月,日,时,分,秒)

方法（均需通过 on_before_action 实现）

编号	名称	说明
1	update_amount	更新信用额度（指定增/减量）
2	set_amount_to_value	将信用额度设为指定值
3	invoke_credit	调用/激活信用

构造函数

Credit(logical_name: str, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.19.10.0.255"
access	dict	None	实例级权限

基本用法

import dlms

credit = dlms.Credit("0.0.19.10.0.255")
credit.current_credit_amount = 5000       # 设置余额
credit.type = dlms.CreditType.TOKEN        # 信用类型
credit.priority = 1                        # 高优先级
credit.warning_threshold = 500             # 低于此值触发警告
credit.limit = -200                        # 允许 -200 的债务
credit.credit_configuration = (
    dlms.CreditConfiguration.VISUAL |
    dlms.CreditConfiguration.CONFIRMATION
)
credit.period = (2025, 1, 1, 0, 0, 0)     # 周期起始时间

on_before_action — 方法处理

def on_credit_action(self, event):
    """Credit 方法处理"""
    if event.index == 1:  # update_amount
        # event 中携带参数：增/减量
        # 在 Python 侧实现：self.current_credit_amount += delta
        pass
    elif event.index == 2:  # set_amount_to_value
        # self.current_credit_amount = target_value
        pass
    elif event.index == 3:  # invoke_credit
        # 激活信用（设置 status = CreditStatus.INVOKED）
        pass
    return True

credit.on_before_action = on_credit_action

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
current_credit_amount	int	√	当前余额
type	CreditType	√	信用类型枚举
priority	int	√	消耗优先级（值越低越优先）
warning_threshold	int	√	警告阈值
limit	int	√	限制（可为负值）
credit_configuration	CreditConfiguration	√	配置位掩码
status	int	√	当前状态（对应 CreditStatus）
preset_credit_amount	int	√	预设信用额度
credit_available_threshold	int	√	信用可用阈值
period	tuple(6)	√	周期时间
on_before_action	Callable	√	动作前钩子

---

Charge

费用计算。使用 tariff 表基于消费量计算费用。OBIS 0.0.19.20.0.255 ， COSEM ID = 113 。

Blue Book 属性

编号	Python 名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	total_amount_paid	int32	已支付总金额
3	charge_type	ChargeType	收费类型
4	priority	uint8	优先级
5	unit_charge_active	dict	活动 tariff 表（复杂结构，见下文）
6	unit_charge_passive	dict	备用 tariff 表
7	unit_charge_activation_time	tuple(6)	tariff 激活时间
8	period	uint32	计费周期（秒）
9	charge_configuration	ChargeConfiguration	收费配置位掩码
10	last_collection_time	tuple(6)	上次回收时间
11	last_collection_amount	int32	上次回收金额
12	total_amount_remaining	int32	剩余总金额
13	proportion	uint16	比例因子

方法（均需通过 on_before_action 实现）

编号	名称	说明
1	update_unit_charge	更新 unit_charge（将 passive→active 或设置新值）
2	activate	激活备用 tariff（passive → active）
3	collect	执行费用回收（消耗 Credit 余额）
4	update_last_collection_time	更新上次回收时间
5	update_total_amount_remaining	更新剩余总金额
6	set_total_amount_paid	设置已支付总金额

构造函数

Charge(logical_name: str, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.19.20.0.255"
access	dict	None	实例级权限

``unit_charge 字典结构

unit_charge_active 和 unit_charge_passive 使用相同的字典格式：

unit_charge = {
    "charge_per_unit_scaling": {
        "commodity_scale": int,   # 商品缩放
        "price_scale": int,       # 价格缩放
    },
    "commodity": {
        "target": dlms_object,    # 引用一个 DLMS 对象（如 Register）
        "attribute_index": int,   # 目标属性索引
    },
    "charge_tables": [
        {
            "index": bytes,       # 费率索引
            "charge_per_unit": int  # 每单位费用
        },
        # ...更多费率条目
    ]
}

基本用法

import dlms

charge = dlms.Charge("0.0.19.20.0.255")
charge.charge_type = dlms.ChargeType.CONSUMPTION_BASED_COLLECTION
charge.priority = 2
charge.period = 3600  # 每小时计费周期

# 设置活动 tariff
charge.unit_charge_active = {
    "charge_per_unit_scaling": {
        "commodity_scale": 0,
        "price_scale": -3,     # 3 位小数
    },
    "commodity": {
        "target": register_obj,  # 引用一个 Register 对象
        "attribute_index": 2,    # 读取 attribute 2（value）
    },
    "charge_tables": [
        {"index": b"\x01", "charge_per_unit": 150},   # 费率 1: 1.50 元/单位
        {"index": b"\x02", "charge_per_unit": 120},   # 费率 2: 1.20 元/单位
        {"index": b"\x03", "charge_per_unit": 80},    # 费率 3: 0.80 元/单位
    ]
}

on_before_action — 方法处理

def on_charge_action(self, event):
    """Charge 方法处理"""
    if event.index == 1:  # update_unit_charge
        # 将 unit_charge_passive 复制到 unit_charge_active
        self.unit_charge_active = self.unit_charge_passive
    elif event.index == 2:  # activate
        # 激活备用 tariff
        self.unit_charge_active = self.unit_charge_passive
        self.unit_charge_activation_time = (2025, 6, 15, 8, 0, 0)
    elif event.index == 3:  # collect
        # 从关联 Credit 回收费用
        # ...实现费用计算和扣减逻辑...
        self.last_collection_time = (2025, 6, 15, 10, 0, 0)
        self.last_collection_amount = 120
    elif event.index == 4:  # update_last_collection_time
        pass
    elif event.index == 5:  # update_total_amount_remaining
        pass
    elif event.index == 6:  # set_total_amount_paid
        pass
    return True

charge.on_before_action = on_charge_action

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
total_amount_paid	int	√	已支付总金额
charge_type	ChargeType	√	收费类型枚举
priority	int	√	收费优先级
unit_charge_active	dict	√	活动 tariff 表（复杂 dict）
unit_charge_passive	dict	√	备用 tariff 表
unit_charge_activation_time	tuple(6)	√	tariff 激活时间
period	int	√	计费周期（秒）
charge_configuration	ChargeConfiguration	√	收费配置位掩码
last_collection_time	tuple(6)	√	上次回收时间
last_collection_amount	int	√	上次回收金额
total_amount_remaining	int	√	剩余总金额
proportion	int	√	比例因子
on_before_action	Callable	√	动作前钩子

---

Account

顶层预付费控制器。关联多个 Credit（余额源）、Charge（扣费逻辑）和 TokenGateway（Token入口）。OBIS 0.0.19.0.0.255 , COSEM ID = 111 。

Blue Book 属性

编号	Python 名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	payment_mode	AccountPaymentMode	支付模式（CREDIT / PREPAYMENT）
3	account_status	AccountStatus	账户状态
4	current_credit_in_use	uint8	当前使用的信用索引
5	current_credit_status	AccountCreditStatus	当前信用状态（位掩码）
6	available_credit	int32	可用信用总额
7	amount_to_clear	int32	需清偿金额
8	clearance_threshold	int32	清偿阈值
9	aggregated_debt	int32	累计债务
10	credit_references	list[str]	关联的 Credit 对象的 OBIS 列表
11	charge_references	list[str]	关联的 Charge 对象的 OBIS 列表
12	credit_charge_configurations	list[dict]	Credit↔Charge 映射配置
13	token_gateway_configurations	list[dict]	TokenGateway↔Credit 映射配置
14	account_activation_time	tuple(6)	账户激活时间
15	account_closure_time	tuple(6)	账户关闭时间
16	currency	dict	货币信息 {name, scale, unit}
17	low_credit_threshold	int32	低信用阈值
18	next_credit_available_threshold	int32	下一信用可用阈值
19	max_provision	uint16	最大预授权
20	max_provision_period	int32	最大预授权周期

方法

支持 18 个动作方法 ，编号 1–18。应用须在 on_before_action 中实现全部 18 个方法。

编号	说明
1–18	由应用自行定义（如激活、充值、冻结、结算等）

构造函数

Account(logical_name: str, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.19.0.0.255"
access	dict	None	实例级权限

基本用法

import dlms

account = dlms.Account("0.0.19.0.0.255")
account.payment_mode = dlms.AccountPaymentMode.PREPAYMENT
account.account_status = dlms.AccountStatus.ACTIVE

# 关联 Credit 和 Charge 对象（通过 OBIS 字符串引用）
account.credit_references = [
    "0.0.19.10.0.255",   # 主信用
    "0.0.19.11.0.255",   # 应急信用
]
account.charge_references = [
    "0.0.19.20.0.255",   # 电费
]

# 配置 credit 到 charge 的映射
account.credit_charge_configurations = [
    {
        "credit_reference": "0.0.19.10.0.255",
        "charge_reference": "0.0.19.20.0.255",
        "collection_configuration": dlms.CreditCollectionConfiguration.NONE,
    }
]

# 配置 token gateway 到 credit 的比例分配
account.token_gateway_configurations = [
    {
        "credit_reference": "0.0.19.10.0.255",
        "token_proportion": 100,   # 100% 分配给主信用
    }
]

# 设置货币信息
account.currency = {
    "name": b"CNY",
    "scale": -2,     # 2 位小数（分）
    "unit": dlms.Currency.MONETARY,
}

account.low_credit_threshold = 1000
account.next_credit_available_threshold = 2000
account.max_provision = 50000
account.max_provision_period = 2592000  # 30 天（秒）

复杂属性详解

credit_charge_configurations — 列表，每个元素是一个 dict：

{
    "credit_reference": str,            # Credit 对象的 OBIS
    "charge_reference": str,            # Charge 对象的 OBIS
    "collection_configuration": int,    # CreditCollectionConfiguration 位掩码
}

token_gateway_configurations — 列表，每个元素是一个 dict：

{
    "credit_reference": str,    # Credit 对象的 OBIS
    "token_proportion": int,    # 分配给该 Credit 的比例（0–100）
}

currency — 字典：

{
    "name": bytes,    # 货币名称（如 b"CNY" / b"USD"）
    "scale": int,     # 小数位数（负值表示相反方向）
    "unit": int,      # Currency 枚举值
}

on_before_action — 方法处理

def on_account_action(self, event):
    """Account 方法处理 — 需实现全部 18 个方法"""
    if event.index == 1:
        # 激活账户
        self.account_status = dlms.AccountStatus.ACTIVE
    elif event.index == 2:
        # 关闭账户
        self.account_status = dlms.AccountStatus.CLOSED
        self.account_closure_time = (2025, 12, 31, 23, 59, 59)
    elif event.index == 3:
        # 充值 — 根据 token_gateway_configurations 按比例分配
        pass
    # ... 方法 4–17 ...
    elif event.index == 18:
        # 最终结算
        pass
    return True

account.on_before_action = on_account_action

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
payment_mode	AccountPaymentMode	√	支付模式
account_status	AccountStatus	√	账户状态
current_credit_in_use	int	√	当前使用信用索引
current_credit_status	AccountCreditStatus	√	当前信用状态位掩码
available_credit	int	√	可用信用总额
amount_to_clear	int	√	需清偿金额
clearance_threshold	int	√	清偿阈值
aggregated_debt	int	√	累计债务
credit_references	list[str]	√	Credit OBIS 引用列表
charge_references	list[str]	√	Charge OBIS 引用列表
credit_charge_configurations	list[dict]	√	Credit→Charge 映射列表
token_gateway_configurations	list[dict]	√	TokenGateway→Credit 映射列表
account_activation_time	tuple(6)	√	激活时间
account_closure_time	tuple(6)	√	关闭时间
currency	dict	√	货币信息 {name, scale, unit}
low_credit_threshold	int	√	低信用阈值
next_credit_available_threshold	int	√	下一信用可用阈值
max_provision	int	√	最大预授权
max_provision_period	int	√	最大预授权周期
on_before_action	Callable	√	动作前钩子

---

枚举参考

所有枚举均通过 dlms.ClassName.VALUE 访问。

---

TokenStatusCode

Token 处理状态码。

Python 名称	值	说明
TokenStatusCode.FORMAT_OK	0	格式正确
TokenStatusCode.AUTHENTICATION_OK	1	认证通过
TokenStatusCode.VALIDATION_OK	2	验证通过
TokenStatusCode.TOKEN_EXECUTION_OK	3	执行成功
TokenStatusCode.TOKEN_FORMAT_FAILURE	4	格式错误
TokenStatusCode.AUTHENTICATION_FAILURE	5	认证失败
TokenStatusCode.VALIDATION_RESULT_FAILURE	6	验证结果失败
TokenStatusCode.TOKEN_EXECUTION_RESULT_FAILURE	7	执行结果失败
TokenStatusCode.TOKEN_RECEIVED	8	Token 已接收（待处理）

import dlms
gateway.status = dlms.TokenStatusCode.VALIDATION_OK

---

TokenDelivery

Token 投递方式。

Python 名称	值	说明
TokenDelivery.REMOTE	0	远程投递（如通信网络）
TokenDelivery.LOCAL	1	本地投递（如光口/串口）
TokenDelivery.MANUAL	2	手动输入（如键盘）

import dlms
gateway.delivery_method = dlms.TokenDelivery.REMOTE

---

CreditType

信用类型。

Python 名称	值	说明
CreditType.TOKEN	0	Token 充值信用
CreditType.RESERVED	1	保留
CreditType.EMERGENCY	2	应急信用
CreditType.TIME_BASED	3	基于时间的信用
CreditType.CONSUMPTION_BASED	4	基于消费量的信用

import dlms
credit.type = dlms.CreditType.TOKEN

---

CreditStatus

信用状态。

Python 名称	值	说明
CreditStatus.ENABLED	0	已启用
CreditStatus.SELECTABLE	1	可选
CreditStatus.INVOKED	2	已调用
CreditStatus.IN_USE	3	使用中
CreditStatus.CONSUMED	4	已耗尽

import dlms
credit.status = dlms.CreditStatus.ENABLED

---

CreditConfiguration （位掩码）

信用配置标志，可使用 | 组合。

Python 名称	值	说明
CreditConfiguration.NONE	0	无
CreditConfiguration.VISUAL	1	可视显示
CreditConfiguration.CONFIRMATION	2	需确认
CreditConfiguration.PAID_BACK	4	已偿还
CreditConfiguration.RESETTABLE	8	可重置
CreditConfiguration.TOKENS	16	使用 Token

import dlms
credit.credit_configuration = (
    dlms.CreditConfiguration.VISUAL |
    dlms.CreditConfiguration.CONFIRMATION
)

---

CreditCollectionConfiguration （位掩码）

信用回收配置标志。

Python 名称	值	说明
CreditCollectionConfiguration.NONE	0	无
CreditCollectionConfiguration.DISCONNECTED	1	断开时回收
CreditCollectionConfiguration.LOAD_LIMITING	2	限荷时回收
CreditCollectionConfiguration.FRIENDLY_CREDIT	4	友好信用回收

import dlms
ccc = dlms.CreditCollectionConfiguration.DISCONNECTED

---

ChargeType

收费类型。

Python 名称	值	说明
ChargeType.CONSUMPTION_BASED_COLLECTION	0	基于消费量收费
ChargeType.TIME_BASED_COLLECTION	1	基于时间收费
ChargeType.PAYMENT_EVENT_BASED_COLLECTION	2	基于支付事件收费

import dlms
charge.charge_type = dlms.ChargeType.CONSUMPTION_BASED_COLLECTION

---

ChargeConfiguration （位掩码）

收费配置标志。

Python 名称	值	说明
ChargeConfiguration.NONE	0	无
ChargeConfiguration.PERCENTAGE_BASED_COLLECTION	1	基于百分比回收
ChargeConfiguration.CONTINUOUS_COLLECTION	2	连续回收

import dlms
charge.charge_configuration = dlms.ChargeConfiguration.CONTINUOUS_COLLECTION

---

AccountStatus

账户状态。

Python 名称	值	说明
AccountStatus.NEW_INACTIVE_ACCOUNT	0	新建未激活
AccountStatus.ACTIVE	1	已激活
AccountStatus.CLOSED	2	已关闭

import dlms
account.account_status = dlms.AccountStatus.ACTIVE

---

AccountPaymentMode

支付模式。

Python 名称	值	说明
AccountPaymentMode.CREDIT	0	后付费（信用模式）
AccountPaymentMode.PREPAYMENT	1	预付费

import dlms
account.payment_mode = dlms.AccountPaymentMode.PREPAYMENT

---

AccountCreditStatus （位掩码）

账户信用状态位掩码。

Python 名称	值	说明
AccountCreditStatus.NONE	0	无特殊状态
AccountCreditStatus.IN_CREDIT	1	有可用信用
AccountCreditStatus.LOW_CREDIT	2	低信用
AccountCreditStatus.NEXT_CREDIT_ENABLED	4	下一信用已启用
AccountCreditStatus.NEXT_CREDIT_SELECTABLE	8	下一信用可选
AccountCreditStatus.CREDIT_REFERENCE_LIST	16	信用引用列表
AccountCreditStatus.SELECTABLE_CREDIT_IN_USE	32	可选信用正在使用
AccountCreditStatus.OUT_OF_CREDIT	64	信用耗尽
AccountCreditStatus.RESERVED	128	保留

import dlms
account.current_credit_status = dlms.AccountCreditStatus.IN_CREDIT

---

Currency

货币单位类型。

Python 名称	值	说明
Currency.TIME	0	时间单位
Currency.CONSUMPTION	1	消费量单位
Currency.MONETARY	2	货币单位

import dlms
currency_unit = dlms.Currency.MONETARY

---

---

其他辅助类

---

StatusMapping

状态字映射。表示一个状态字及其比特位到引用表条目的映射关系，其 COSEM ID = 63 。

Blue Book 属性

编号	Python 名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	status_word	tuple(2)	(dlms_type, value) — 类型标签选择编码格式， value 为 int 或 bytes
3	mapping_table	tuple(2)	(ref_table_id, mapping) — mapping 为单个起始条目整数或每比特条目索引列表

注意 ：客户端 SET 请求会拒绝属性 2 和 3 的写入。需直接从 Python 更新值。

status_word 支持的类型标签

标签值	含义
4	BIT_STRING（位串）
6	UINT32（32 位无符号整数）
9	OCTET_STRING（八位位组串）
10	STRING（字符串）
12	STRING_UTF8（UTF-8 字符串）
17	UINT8（8 位无符号整数）
18	UINT16（16 位无符号整数 — 默认 ）
21	UINT64（64 位无符号整数）

mapping_table 格式

• 长-无符号选择 （ mapping 为 int ）：表示引用表中的起始条目索引
• 数组选择 （ mapping 为 list[int] ）：显式指定每个比特位对应的条目索引列表

构造函数

StatusMapping(logical_name: str, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串
access	dict	None	实例级权限

基本用法

import dlms

sm = dlms.StatusMapping("0.0.127.0.0.255")

# 设置状态字：UINT16 类型，值 = 0x0003（bit 0 和 bit 1 置位）
sm.status_word = (18, 3)

# 设置映射表：引用表 ID = 1，每比特对应条目索引
sm.mapping_table = (1, [10, 11, 12, 13])

# 或使用长-无符号选择（起始条目索引）
sm.mapping_table = (1, 10)

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
status_word	tuple(2)	√	(dlms_type, value) — 状态字及其编码
mapping_table	tuple(2)	√	(ref_table_id, mapping) — 映射描述
on_before_read	Callable	√	读前钩子（继承 CosemObject）
on_after_read	Callable	√	读后钩子
on_before_write	Callable	√	写前钩子
on_after_write	Callable	√	写后钩子
on_before_action	Callable	√	动作前钩子
on_after_action	Callable	√	动作后钩子
access_dict	dict	√	实例级访问控制

---

Arbitrator

仲裁器。解决不同参与方之间并发请求的冲突，使用加权权限判定，其 COSEM ID = 68 。

Blue Book 属性

编号	Python 名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	actions	list[tuple]	(ScriptTable_or_None, selector) 列表
3	permissions_table	list[bytes]	权限位集，每参与方一行
4	weightings_table	list[list[int]]	权重表，每参与方×每动作的 uint16 权重
5	most_recent_requests_table	list[bytes]	最近请求位集，每参与方一行
6	last_outcome	uint8	上次仲裁结果（0–255）

方法（需通过 on_before_action 实现）

编号	名称	说明
1	push	提交请求 bytes（由应用实现仲裁逻辑）

构造函数

Arbitrator(logical_name: str, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串
access	dict	None	实例级权限

基本用法

import dlms

arb = dlms.Arbitrator("0.0.127.1.0.255")

# 设置动作列表：每个元素为 (ScriptTable 对象或 None, selector)
script1 = dlms.ScriptTable("0.0.10.1.0.255")
arb.actions = [
    (script1, 1),
    (None, 0),       # 无脚本的动作占位
]

# 权限表：bytes 位集，每参与方一行（bitset）
arb.permissions_table = [
    b"\x03",   # 参与方 0：bit 0,1 有权限
    b"\x01",   # 参与方 1：bit 0 有权限
    b"\x00",   # 参与方 2：无权限
]

# 权重表：每行 = 该参与方对各动作的权重（uint16）
arb.weightings_table = [
    [100, 50],    # 参与方 0：动作 0 权重 100，动作 1 权重 50
    [200, 30],    # 参与方 1：动作 0 权重 200，动作 1 权重 30
    [0, 0],       # 参与方 2：无权重
]

# 最近请求表：bytes 位集
arb.most_recent_requests_table = [
    b"\x01",   # 参与方 0 请求了动作 0
    b"\x02",   # 参与方 1 请求了动作 1
    b"\x00",   # 参与方 2 无请求
]

# 上次仲裁结果
arb.last_outcome = 0

on_before_action — 方法处理

def on_arbitrator_action(self, event):
    """Arbitrator 方法处理 — push"""
    if event.index == 1:  # push
        # 从 event 中获取请求数据
        # ...实现仲裁逻辑...
        # self.last_outcome = chosen_action_index
        pass
    return True

arb = dlms.Arbitrator("0.0.127.1.0.255")
arb.on_before_action = on_arbitrator_action

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
actions	list[tuple]	√	(ScriptTable_or_None, selector) 列表
permissions_table	list[bytes]	√	每参与方一行的权限位集
weightings_table	list[list[int]]	√	每参与方×每动作的 uint16 权重
most_recent_requests_table	list[bytes]	√	每参与方一行的最近请求位集
last_outcome	int	√	上次仲裁结果
on_before_action	Callable	√	动作前钩子

---

DataProtection

数据保护。为包裹的一组对象提供 COSEM 级密码保护，其 COSEM ID = 30 。

重要 ：Gurux 中对类 30 无 C 端 GET 或 ACTION 调度（ cosem_getDataProtection() 返回 DLMS_ERROR_CODE_NOT_IMPLEMENTED ）。应用必须通过 on_before_read 和 on_before_action 回调实现所有读取和动作处理。

Blue Book 属性

编号	Python 名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	protection_buffer	bytes	保护缓冲区（加密/认证后的数据）
3	required_protection	RequiredProtection	所需保护类型位掩码
4	protection_object_list	list[dict]	受保护对象列表（Python 侧存储）
5	protection_parameters_get	list[dict]	GET 操作的保护参数（Python 侧存储）
6	protection_parameters_set	list[dict]	SET 操作的保护参数（Python 侧存储）

方法

方法	说明
protect(plaintext_bytes) → bytes	使用当前会话密码（AES-GCM）加密数据，自动更新 protection_buffer
unprotect(ciphertext_bytes) → bytes	解密之前由 protect() 产生的密文包
deinit()	释放 C 堆内存

protect() / unprotect() 仅在启用 HIGH GMAC 加密会话时可用。无密码密钥时将抛出 ValueError 。

protection_object_list 字典结构

{
    "target": dlms_object,       # 受保护的 DLMS 对象
    "attribute_index": int,      # 受保护的属性索引
    "data_index": int,           # 数据索引
}

protection_parameters_get/set 字典结构

{
    "type": int,                 # ProtectionType 值
    "options": {
        "id":          bytes,    # 标识
        "originator":  bytes,    # 发起方标识
        "recipient":   bytes,    # 接收方标识
        "information": bytes,    # 附加信息
        "info": {
            "type": int,         # DataProtectionKeyType 值
            # 当 type=IDENTIFIED (0):
            "identified_options": int,  # IdentifiedKeyType 选项
            # 当 type=WRAPPED (1):
            "wrapped_key": {
                "id":  int,      # 密钥标识
                "key": bytes     # 包裹密钥
            },
            # 当 type=AGREED (2):
            "agreed_key": {
                "parameters":    bytes,  # 协商参数
                "ciphered_data": bytes   # 加密数据
            }
        }
    }
}

构造函数

DataProtection(logical_name: str, access: dict = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串
access	dict	None	实例级权限

基本用法

import dlms

dp = dlms.DataProtection("0.0.127.2.0.255")

# 所需保护类型（位掩码，可组合）
dp.required_protection = (
    dlms.RequiredProtection.AUTHENTICATED_REQUEST |
    dlms.RequiredProtection.ENCRYPTED_REQUEST
)

# 受保护对象列表
dp.protection_object_list = [
    {
        "target": register_obj,
        "attribute_index": 2,
        "data_index": 0,
    }
]

# GET 保护参数
dp.protection_parameters_get = [
    {
        "type": dlms.ProtectionType.AUTHENTICATION_ENCRYPTION,
        "options": {
            "id": b"\x01",
            "originator": b"server",
            "recipient": b"client",
            "information": b"",
            "info": {
                "type": dlms.DataProtectionKeyType.IDENTIFIED,
                "identified_options": dlms.IdentifiedKeyType.UNICAST_ENCRYPTION,
            }
        }
    }
]

# 加密数据（需已建立加密会话）
# ciphertext = dp.protect(b"plaintext data")
# plaintext  = dp.unprotect(ciphertext)

# 资源清理
dp.deinit()

on_before_read / on_before_action — 处理

由于 C 端无 GET/ACTION 调度，复杂属性的读取和方法执行需通过回调实现：

def on_dp_before_read(self, event):
    """DataProtection 读前钩子 — 实现属性 4/5/6 的读取"""
    if event.index == 4:
        # 构造 protection_object_list
        event.data = self.protection_object_list
        return True
    elif event.index == 5:
        event.data = self.protection_parameters_get
        return True
    elif event.index == 6:
        event.data = self.protection_parameters_set
        return True
    return True

dp.on_before_read = on_dp_before_read

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
protection_buffer	bytes	√	保护缓冲区数据
required_protection	int	√	所需保护类型位掩码
protection_object_list	list[dict]	√	受保护对象列表
protection_parameters_get	list[dict]	√	GET 保护参数
protection_parameters_set	list[dict]	√	SET 保护参数
on_before_read	Callable	√	读前钩子（ 需实现复杂属性读取 ）
on_before_action	Callable	√	动作前钩子
on_after_read	Callable	√	读后钩子
on_after_write	Callable	√	写后钩子
on_before_write	Callable	√	写前钩子
on_after_action	Callable	√	动作后钩子
access_dict	dict	√	实例级访问控制

---

枚举参考（续）

RequiredProtection （位掩码）

所需保护类型，可使用 | 组合。

Python 名称	值	说明
RequiredProtection.NONE	0	无需保护
RequiredProtection.AUTHENTICATED_REQUEST	1	请求需认证
RequiredProtection.ENCRYPTED_REQUEST	2	请求需加密
RequiredProtection.DIGITALLY_SIGNED_REQUEST	4	请求需数字签名
RequiredProtection.AUTHENTICATED_RESPONSE	16	响应需认证
RequiredProtection.ENCRYPTED_RESPONSE	32	响应需加密
RequiredProtection.DIGITALLY_SIGNED_RESPONSE	64	响应需数字签名

import dlms
dp.required_protection = (
    dlms.RequiredProtection.AUTHENTICATED_REQUEST |
    dlms.RequiredProtection.ENCRYPTED_REQUEST
)

---

ProtectionType

保护类型。

Python 名称	值	说明
ProtectionType.AUTHENTICATION	0	仅认证
ProtectionType.ENCRYPTION	1	仅加密
ProtectionType.AUTHENTICATION_ENCRYPTION	2	认证+加密

import dlms
pt = dlms.ProtectionType.AUTHENTICATION_ENCRYPTION

---

DataProtectionKeyType

数据保护密钥类型。

Python 名称	值	说明
DataProtectionKeyType.IDENTIFIED	0	已标识密钥
DataProtectionKeyType.WRAPPED	1	包裹密钥
DataProtectionKeyType.AGREED	2	协商密钥

import dlms
kt = dlms.DataProtectionKeyType.IDENTIFIED

---

IdentifiedKeyType

已标识密钥类型。

Python 名称	值	说明
IdentifiedKeyType.UNICAST_ENCRYPTION	0	单播加密
IdentifiedKeyType.BROADCAST_ENCRYPTION	1	广播加密

import dlms
ikt = dlms.IdentifiedKeyType.UNICAST_ENCRYPTION

---

WrappedKeyType

包裹密钥类型。

Python 名称	值	说明
WrappedKeyType.MASTER_KEY	0	主密钥

import dlms
wkt = dlms.WrappedKeyType.MASTER_KEY

---

SapAssignment

SAP（Service Access Point，服务访问点）分配对象。将 SAP 编号映射到逻辑设备名称（LDN），允许 DLMS 客户端发现集中器上的逻辑设备。OBIS 0.0.41.0.0.255 ， COSEM ID = 17 。

Blue Book 属性

编号	Python 名称	类型	说明
1	logical_name	octet-string(6)	OBIS 代码，只读
2	sap_assignment_list	list[tuple]	[(sap_id, device_name), ...] ， device_name 为 bytes

方法

方法	签名	说明
connect_logical_device	(sap_id, device_name)	添加或更新单个 SAP 条目。若 sap_id 已存在则更新设备名称，否则新增
deinit	()	释放 C 堆内存

构造函数

SapAssignment(logical_name: str, sap_assignment_list: list = None)

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码字符串，如 "0.0.41.0.0.255"
sap_assignment_list	list	None	初始 SAP 分配列表，格式 [(sap_id, device_name), ...]

基本用法

import dlms

# 创建 SAP 分配对象
sap = dlms.SapAssignment(
    "0.0.41.0.0.255",
    sap_assignment_list=[
        (1, b"GRX0000000012345"),
    ]
)

server.add_object(sap)

# 读取 SAP 分配列表
assignments = sap.sap_assignment_list
for sap_id, device_name in assignments:
    print("SAP {} → {}".format(sap_id, device_name))

# 动态添加或更新 SAP 条目
sap.connect_logical_device(2, b"GRX0000000067890")

# 资源清理
sap.deinit()

Python 属性一览

属性	类型	可写	说明
logical_name	bytes	×	OBIS 代码（6 字节）
sap_assignment_list	list[tuple]	√	[(sap_id(int), device_name(bytes)), ...]
on_before_read	Callable	√	读前钩子（继承 CosemObject）
on_after_read	Callable	√	读后钩子
on_before_write	Callable	√	写前钩子
on_after_write	Callable	√	写后钩子
on_before_action	Callable	√	动作前钩子
on_after_action	Callable	√	动作后钩子
access_dict	dict	√	实例级访问控制

---

事件处理（Event Handling）

每个 COSEM 对象暴露六个回调属性，在客户端请求处理的不同阶段被调用。回调是普通的 Python 函数，直接在对象上赋值即可。

---

DLMSEvent — 事件上下文

DLMSEvent 是服务器为每个客户端请求创建的事件对象，传递给所有回调。

属性	类型	说明
index	int	属性编号（GET/SET）或方法编号（ACTION），1-based
selector	int	选择器类型： 0 =无, 1 =范围, 2 =条目
is_action	bool	True 为 ACTION（方法调用）， False 为 GET/SET
selector_params	dict / None	选择器参数（见下表）
parameters	int / None	ACTION 请求的方法参数

selector_params

selector	含义	字典内容
0	无选择器	None
1	范围选择器	{"from_time": int, "to_time": int}
2	条目选择器	{"from_entry": int, "to_entry": int}

辅助方法

obj.idx("attr_name")      # 属性名 → 索引，如 obj.idx('buffer') → 2
obj.attr_name(index)      # 索引 → 属性名，如 obj.attr_name(2) → 'buffer'

---

六大回调

所有回调签名为 handler(self, event) 。默认均为 None （禁用），赋值为可调用对象即启用。

回调	触发时机	用途
on_before_read	GET 响应序列化前	刷新值、返回自定义数据、拒绝访问
on_after_read	GET 响应发送后	日志、审计
on_before_write	SET 请求应用前	验证、拒绝写入
on_after_write	SET 请求应用后	同步硬件、记录变更
on_before_action	ACTION 分发前	执行方法逻辑
on_after_action	ACTION 分发后	日志、审计

on_before_read 返回值

返回值	效果
None / True	默认处理，序列化当前属性值
False	拒绝请求，客户端收到 Access Violation
list	替代默认值，将返回的列表编码后发送给客户端（用于 ProfileGeneric）

---

常见模式

读前刷新传感器值

import dlms

energy_reg = dlms.Register("1.0.1.8.0.255", scaler=-3)

def refresh_value(self, event):
    if event.index == self.idx('value'):
        self.value = read_energy_sensor()  # 应用自定义
    return True

energy_reg.on_before_read = refresh_value

写前校验

def validate_write(self, event):
    if event.index == 2:  # value
        # event 中可获取待写入值
        pass  # return False 可拒绝写入
    return True

data.on_before_write = validate_write

驱动硬件（ACTION）

def relay_handler(self, event):
    if event.index == 1:      # remote_disconnect
        gpio_relay.value(0)
    elif event.index == 2:    # remote_reconnect
        gpio_relay.value(1)
    return True               # 允许内存状态同步

disconnect_ctl.on_before_action = relay_handler

ProfileGeneric 从外部存储提供数据

def provide_buffer(self, event):
    if event.index != self.idx('buffer'):
        return True
    if event.selector == 1:
        t_from = event.selector_params['from_time']
        t_to   = event.selector_params['to_time']
        return load_flash_rows(t_from, t_to)  # 返回 list
    return load_flash_rows(None, None)

profile.on_before_read = provide_buffer

写后审计日志

def audit_write(self, event):
    attr = self.attr_name(event.index)
    now  = utime.localtime()
    msg  = "WRITE {}.{} at {}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(
        self.logical_name, attr,
        now[0], now[1], now[2], now[3], now[4], now[5])
    append_audit_log(msg)

clock.on_after_write = audit_write

---

线程安全要点

• 每个连接在独立的后台线程中运行，回调在所属连接的线程中执行
• SerialConnection / OpticalConnection / MobileConnection 在 C 系统线程中运行
• GenericConnection 在调用 connect() 的 Python 线程中运行
• Helios RTOS 无 GIL，共享可变对象需用 _thread.allocate_lock() 保护
• 不要在回调中调用 server.stop() ，会导致死锁

安全模型（Security）

DLMS/COSEM 定义了分层安全模型：访问控制按属性/方法实施，认证决定客户端的信任级别，加密保护传输中的 PDU 内容。

---

访问控制（Access Dictionaries）

每个 COSEM 对象构造函数接受可选的 access 参数。字典以 Authentication 级别为键，内层字典以属性索引（1-based）映射到 AccessMode 常量。

import dlms

reg = dlms.Register("1.0.1.8.0.255", access={
    dlms.Authentication.NONE: {2: dlms.AccessMode.READ},
    dlms.Authentication.HIGH: {3: dlms.AccessMode.AUTHENTICATED_WRITE},
})

AccessMode 常量

常量	值	含义
AccessMode.NONE	0	无访问权限（客户端不可见）
AccessMode.READ	1	允许 GET，拒绝 SET
AccessMode.WRITE	2	允许 SET，拒绝 GET
AccessMode.READ_WRITE	3	允许 GET 和 SET
AccessMode.AUTHENTICATED_READ	4	仅认证客户端可 GET
AccessMode.AUTHENTICATED_WRITE	5	仅认证客户端可 SET
AccessMode.AUTHENTICATED_READ_WRITE	6	GET 和 SET 均需认证

Authentication 常量

常量	值	说明
Authentication.NONE	0	公开，无需密码
Authentication.LOW	1	明文密码认证
Authentication.HIGH	2	挑战-响应（HLS）
Authentication.HIGH_MD5	3	挑战-响应 + MD5
Authentication.HIGH_SHA1	4	挑战-响应 + SHA-1
Authentication.HIGH_GMAC	5	AES-GCM 认证+可选加密
Authentication.HIGH_SHA256	6	挑战-响应 + SHA-256
Authentication.HIGH_ECDSA	7	ECDSA 认证

类级默认值

dlms.set_default_access() 可为整个类设置默认权限，避免在每个对象上重复编写。优先级： 实例 access_dict > 构造函数 access > set_default_access > 默认（无权限） 。

# 所有 Register 对象默认公开可读
dlms.set_default_access(dlms.Register, {
    dlms.Authentication.NONE: {2: dlms.AccessMode.READ},
})

# 个别对象可覆盖
sensitive_reg = dlms.Register("1.0.1.8.1.255")
sensitive_reg.access_dict = {
    dlms.Authentication.HIGH: {2: dlms.AccessMode.AUTHENTICATED_READ_WRITE},
}

---

认证机制

AssociationLogicalName.auth_mechanism 选择该关联所需的认证级别。

值	说明
'None'	公开关联，无密码。适用于初始化阶段的逻辑设备，配合 access_dict 限制敏感属性
'Low'	明文密码认证。客户端在 AA-open 握手中提交明文密码，服务器与 secret 比对。仅适合光口等物理安全接口
'High'	挑战-响应（HLS）。双方交换随机挑战，用共享密钥哈希后返回摘要。需要 SecuritySetup 且 security_policy = SecurityPolicy.NOTHING
'HighGMac'	AES-GCM 认证（可选加密通信）。需要配置 SecuritySetup 的系统标题、GUEK 和 GAK。所有 APDU 受完整性保护

---

SecuritySetup 配置

dlms.SecuritySetup 是接口类 64（OBIS 0.0.43.0.X.255 ），每个安全上下文创建一个对象。

关键属性

属性	类型	说明
security_policy	int	SecurityPolicy 值（见下表）
security_suite	int	密码套件： 0 =AES-128 GCM, 1 =ECDH/AES, 2 =ECDH/AES 变体
server_system_title	bytes	服务器 8 字节系统标题（标识符 3 字节 + 序列号 5 字节）
client_system_title	bytes	客户端 8 字节系统标题
guek	bytes	Global Unicast Encryption Key（16 或 32 字节）
gak	bytes	Global Authentication Key（16 或 32 字节）
min_invocation_counter	int	最小调用计数器，防重放攻击

SecurityPolicy 值

常量	值	含义
SecurityPolicy.NOTHING	0	无密码层保护（用于 High HLS）
SecurityPolicy.AUTHENTICATED	1	认证所有 APDU
SecurityPolicy.ENCRYPTED	2	加密所有 APDU
SecurityPolicy.AUTHENTICATED_ENCRYPTED	3	认证+加密所有 APDU

最小 HighGMac 配置

import dlms

sec = dlms.SecuritySetup("0.0.43.0.2.255")
sec.security_policy = dlms.SecurityPolicy.AUTHENTICATED_ENCRYPTED
sec.security_suite = 0
sec.server_system_title = b'GRX12345'
sec.client_system_title = b'GRX00001'
sec.guek = b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0A\x0B\x0C\x0D\x0E\x0F'
sec.gak  = b'\xD0\xD1\xD2\xD3\xD4\xD5\xD6\xD7\xD8\xD9\xDA\xDB\xDC\xDD\xDE\xDF'

调用计数器持久化

GCM 调用计数器随每个加密帧递增。重启后必须恢复，否则客户端会拒绝重放帧。推荐方案：使用 nocopy=True 的 Data 对象引用 SecuritySetup 属性 6。

import dlms

# nocopy=True 表示 .value 持有对 (sec, 6) 的实时引用
inv_ctr = dlms.Data("0.0.43.1.2.255", nocopy=True)
inv_ctr.value = (sec, 6)  # 指向 SecuritySetup 属性 6（调用计数器）
inv_ctr.access_dict = {dlms.Authentication.HIGH_GMAC: {2: dlms.AccessMode.READ}}

启动时，反序列化后将 sec.min_invocation_counter 设为恢复值加上安全余量（如 +100）。

---

密码处理

• AssociationLogicalName.secret 和 AssociationShortName.secret 接受 bytes 值
• 生产环境建议从安全存储读取密码和密钥，而非硬编码在固件中
• 切勿在日志或调试输出中打印 secret 或密钥值
• 每台设备的密钥应从根密钥和设备序列号派生，而非全设备相同

KEK（Key Encryption Key）

当管理客户端可能使用 Global-Key-Transfer 服务时，需通过 dlms.set_kek() 设置 16 字节 KEK。KEK 必须保存在安全存储中。

import dlms
import SecureData

KEK_INDEX = 1
buf = bytearray(16)
length = SecureData.Read(KEK_INDEX, buf, 16)
if length != 16:
    raise RuntimeError("KEK not provisioned (SecureData index {})".format(KEK_INDEX))
dlms.set_kek(bytes(buf))

首次写入 KEK：

import SecureData

kek = b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0A\x0B\x0C\x0D\x0E\x0F'
SecureData.Store(KEK_INDEX, kek, 16)

---

AssociationLogicalName 配置

接口类 15（OBIS 0.0.40.0.X.255 ），每个客户端关联创建一个对象。

关键属性

属性	类型	说明
auth_mechanism	str	'None' / 'Low' / 'High' / 'HighGMac'
secret	bytes / None	Low 或 High 认证的共享密码； None 用于 None 和 HighGMac
objects	list	该关联可访问的 COSEM 对象列表
clientSAP	int	客户端 SAP 号（管理用 SAP 1，公开用 SAP 16）
security_setup	SecuritySetup / None	High 和 HighGMac 必需；None 和 Low 为 None
context	DLMSContext / None	可选，控制协商的 PDU 大小和一致性块

DLMSContext

属性	说明
conformance	Conformance 常量位掩码（默认协商）
maxReceivePduSize	最大接收 PDU 大小（字节；0=使用连接默认）
maxSendPduSize	最大发送 PDU 大小（字节；0=使用连接默认）
dlmsVersionNumber	DLMS 版本（6=DLMS/COSEM Ed. 7+）

双关联示例

import dlms

# --- 公开只读关联 ---
assoc_pub = dlms.AssociationLogicalName("0.0.40.0.1.255")
assoc_pub.auth_mechanism = 'None'
assoc_pub.clientSAP = 16
assoc_pub.objects = [clock, energy_reg, profile]

# --- 管理关联（HighGMac）---
assoc_mgmt = dlms.AssociationLogicalName("0.0.40.0.2.255")
assoc_mgmt.auth_mechanism = 'HighGMac'
assoc_mgmt.clientSAP = 1
assoc_mgmt.objects = [clock, energy_reg, profile, inv_ctr, sec]
assoc_mgmt.security_setup = sec
assoc_mgmt.context = dlms.DLMSContext(dlmsVersionNumber=6)

# 两个关联对象都需通过 server.add_object() 注册
server.add_object(assoc_pub)
server.add_object(assoc_mgmt)

---

Short Name 关联（AssociationShortName）

接口类 12（OBIS 0.0.40.0.0.255 ）。SN 关联使用 16 位短名引用属性，减少帧开销。仅支持 Low 认证。

import dlms

assoc_sn = dlms.AssociationShortName("0.0.40.0.0.255")
assoc_sn.secret = b"PASSword"
assoc_sn.objects = [clock, energy_reg]

server.add_object(assoc_sn)

conn = dlms.MobileConnection(
    tcp_udp_setup=tcp_udp,
    gprs_setup=gprs,
    gsm_diag=gsm,
    recv_buffer=recv_buf,
    use_logical_name=False,  # SN 模式
)
server.add_connection(conn)

SN 和 LN 关联可在同一服务器上共存，互不冲突。

连接（Connection）

连接表示 DLMS 客户端与服务器通信的物理或逻辑传输通道。每种连接类型封装不同的底层介质（UART、光口、蜂窝 UDP 或自定义 Python I/O），均通过 server.add_connection(conn) 添加到服务器。调用 server.start() 后，服务器为每个连接启动一个独立线程。

---

IecHdlcSetup

接口类 23（OBIS 0.0.22.0.0.255 ）。它既是 COSEM 对象，也是传递给 SerialConnection 、 OpticalConnection 、 GenericConnection 的 HDLC 参数块。

import dlms

hdlc = dlms.IecHdlcSetup(
    "0.0.22.0.0.255",
    commSpeed=9600,
    windowSizeRx=1,
    windowSizeTx=1,
    maxInfoLenTx=128,
    maxInfoLenRx=128,
    timeout=120,
    deviceAddr=0x10,
)
server.add_object(hdlc)

构造函数参数

参数	类型	默认值	说明
logical_name	str	必传	OBIS 代码
commSpeed	int	9600	UART 波特率
windowSizeRx	int	1	HDLC 接收窗口
windowSizeTx	int	1	HDLC 发送窗口
maxInfoLenTx	int	128	最大发送 HDLC 信息字段长度（字节）
maxInfoLenRx	int	128	最大接收 HDLC 信息字段长度（字节）
timeout	int	120	不活动超时（秒）
deviceAddr	int	0x10	HDLC 服务器地址

---

SerialConnection

串口连接。内部在 C 线程中管理 UART。适用于直连场景，无需自定义传输逻辑。

import dlms

conn = dlms.SerialConnection(uart_port=2, hdlc_setup=hdlc, flowcontrol=0)
conn.on_connected    = lambda: print("serial client connected")
conn.on_disconnected = lambda: print("serial client disconnected")
server.add_connection(conn)

构造函数参数

参数	类型	默认值	说明
uart_port	int	必传	UART 端口号
hdlc_setup	IecHdlcSetup	必传	HDLC 参数块
flowcontrol	int	0	流控制： 0 =无, 1 =RTS/CTS
use_logical_name	bool	True	LN 关联； False 为 SN 关联

---

OpticalConnection

光口连接。实现 IEC 62056-21 Mode E 协商协议。

协商流程：

1. 服务器在 300 bps (7E1) 等待客户端签到序列 /?...\r\n
2. 服务器以 300 bps 回复标识字符串 /<FLAG><BAUD><MODE>\r\n
3. 等待客户端 ACK 字节 ( 0x06 ) 确认波特率
4. 双方切换至协商后的波特率 (8N1)，继续 HDLC 帧通信

需要以下两个支持对象：

LocalPortSetup

接口类 19（OBIS 0.0.20.0.0.255 ），配置光口参数。

属性	类型	说明
default_mode	int	光口协议模式（ 0 =Mode E）
default_baud	int	初始波特率（Mode E 必须为 300 ）
proposed_baud	int	协商后的目标波特率（通常 9600 或 19200 ）
response_time	int	签到与响应间的等待时间（毫秒，默认 1000 ）
device_address	bytes / None	设备标识（最多 6 字节）
password_1	bytes / None	P1 级别（只读）光口密码
password_2	bytes / None	P2 级别（读写）光口密码
password_5	bytes / None	P5 级别（完全访问）光口密码

import dlms

local_port = dlms.LocalPortSetup(
    "0.0.20.0.0.255",
    default_mode=0,
    default_baud=300,
    proposed_baud=9600,
    response_time=200,
    device_address=b"MTR001",
    password_1=b"00000000",
    password_2=b"11111111",
    password_5=b"AAAAAAAA",
)

hdlc = dlms.IecHdlcSetup("0.0.22.0.0.255", commSpeed=9600, deviceAddr=0x10)

conn = dlms.OpticalConnection(
    uart_port=1,
    local_port_setup=local_port,
    hdlc_setup=hdlc,
)
server.add_connection(conn)

构造函数签名

OpticalConnection(uart_port, local_port_setup, hdlc_setup, *, use_logical_name=True)

---

MobileConnection

蜂窝网络连接。在后台线程中运行 C 管理 UDP 套接字。支持两种模式：

模式	条件	说明
直连模式	不提供 relay_tcp_setup	监听 tcp_udp_setup 定义的 UDP 端口，与知道设备公网 IP 的客户端直接通信
中继模式	提供 relay_tcp_setup	向 UDP 中继服务器注册，中继按 HDLC 地址路由；支持 connection.send(data) 主动推送

需要的支持对象

对象	作用
TcpUdpSetup	本地 UDP 端口（默认 4059 ），可关联 IPv4Setup 报告 IP
GprsSetup	APN 配置
GsmDiagnostic	蜂窝信号质量属性
recv_buffer	预分配的 bytearray （推荐 4096 字节）
relay_tcp_setup (可选)	中继模式的第二组 TcpUdpSetup

import dlms

tcp_udp  = dlms.TcpUdpSetup("0.0.25.0.0.255", port=4059)
relay    = dlms.TcpUdpSetup("0.0.25.0.0.254", port=4060)
relay.ip_reference = ipv4_setup  # ipv4_setup.ipAddress = 中继服务器 IP

gprs     = dlms.GprsSetup("0.0.2.0.0.255")
gsm      = dlms.GsmDiagnostic("0.0.25.6.0.255")
recv_buf = bytearray(4096)

mobile = dlms.MobileConnection(
    tcp_udp_setup=tcp_udp,
    gprs_setup=gprs,
    gsm_diag=gsm,
    recv_buffer=recv_buf,
    relay_tcp_setup=relay,  # 直连模式可省略
)
mobile.on_connected    = lambda: print("mobile connected")
mobile.on_disconnected = lambda: print("mobile disconnected")
server.add_connection(mobile)

构造函数签名

MobileConnection(tcp_udp_setup, gprs_setup, gsm_diag, recv_buffer,
                 relay_tcp_setup=None, *, use_logical_name=True)

---

GenericConnection

通用连接。适用于自定义传输场景（如特殊成帧的 UART、SPI、MQTT、G3-PLC 等）。

特性：

• 不创建 C 线程，由 Python 驱动 I/O 循环
• 通过 interface_type 选择成帧类型
• 调用 conn.process_msg(data) 处理收到的数据，返回响应字节或 None

成帧类型

常量	适用场景	所需参数
InterfaceType.HDLC	字节流传输：UART、SPI、RS485	hdlc_setup
InterfaceType.WRAPPER	数据包传输：UDP、MQTT、G3-PLC	tcp_udp_setup
InterfaceType.HDLC_WITH_MODE_E	光口手动 Mode E	hdlc_setup + local_port_setup

import dlms
import _thread
import utime
from machine import UART

hdlc = dlms.IecHdlcSetup("0.0.22.0.0.255", commSpeed=9600, deviceAddr=0x10)

conn = dlms.GenericConnection(
    interface_type=dlms.InterfaceType.HDLC,
    frame_size=1024,
    pdu_size=512,
    hdlc_setup=hdlc,
)
server.add_connection(conn)

def uart_loop():
    uart = UART(2, hdlc.commSpeed, 8, 0, 1, 0)
    conn.connect()
    buf = bytearray(1024)
    while True:
        n = uart.readinto(buf)
        if n:
            resp = conn.process_msg(buf[:n])
            if resp:
                uart.write(resp)
        utime.sleep_ms(10)

_thread.start_new_thread(uart_loop, ())

构造函数签名

GenericConnection(interface_type, frame_size=1024, pdu_size=512,
                  *, hdlc_setup=None, tcp_udp_setup=None,
                  local_port_setup=None, use_logical_name=True)

---

连接生命周期回调

所有连接类型共享以下回调：

回调	触发时机
on_connected	物理链路建立（HDLC 链接 up、UDP 套接字绑定、或 GenericConnection.connect() 调用）
on_disconnected	链路断开（超时、传输错误、或 disconnect() 调用）
on_send	原始字节即将发送（C 驱动连接上为监控钩子； GenericConnection 客户端模式需接入发送）
on_receive	收到原始字节（功能同 on_send ，方向相反）

• C 驱动连接（ SerialConnection / OpticalConnection / MobileConnection ）的回调在 C 监听线程中触发
• GenericConnection 的回调在调用 connect() 的 Python 线程中触发
• 保持回调简短，避免阻塞操作

服务器配置（Server）

本章介绍如何将 COSEM 对象、安全关联对象、连接和 Server 实例组装成一个运行中的 DLMS 服务器。

---

启动顺序

服务器必须按以下顺序组装。顺序错乱可能导致启动或运行时错误。

1. 创建所有 COSEM 对象 — Register、Clock、ProfileGeneric 等，彼此间顺序无关
2. 创建 SecuritySetup 和 AssociationLogicalName/AssociationShortName — 关联对象需在引用的 COSEM 对象之后创建（因为 objects 列表持有 Python 引用）
3. 调用 dlms.set_serial_number(serial) 和 dlms.set_flag_id(flag_id) — 这两个调用 必须 在构造 Server 之前执行，它们决定 HDLC 服务器地址和 GCM 安全帧的系统标题前缀
4. 创建 Server(serial_number, flag_id) — Server 持有序列号和标志 ID 的权威副本
5. 调用 server.add_object(obj) 添加所有 COSEM 对象 — 包括 AssociationLogicalName、SecuritySetup、IecHdlcSetup、TcpUdpSetup 等配置对象。未添加到服务器的对象对客户端不可见
6. 调用 server.set_event_code(data_obj, event_log_obj) — 必须在所有对象添加之后、 server.run() 之前调用
7. 创建连接并调用 server.add_connection(conn) — 每个连接实例调用一次
8. 调用 dlms.set_kek(kek_bytes) — 如果需要 HighGMac 密钥更新功能
9. 调用 server.run() — 非阻塞，立即返回。为每个连接启动一个后台线程和一个监控线程
10. 进入应用主循环 — 主线程自由运行：采集传感器读数、更新 COSEM 对象值、管理文件系统等

最小骨架

import dlms

SERIAL  = 12345678
FLAG_ID = "GRX"

# Step 1 — COSEM 对象
clock       = dlms.Clock("0.0.1.0.0.255", time_zone=0)
energy_reg  = dlms.Register("1.0.1.8.0.255", 0, scaler=-3)
event_code  = dlms.Data("0.0.96.11.0.255")
event_code.value = 0

# Step 2 — 安全与关联
assoc_pub = dlms.AssociationLogicalName("0.0.40.0.1.255")
assoc_pub.auth_mechanism = 'None'
assoc_pub.clientSAP      = 16
assoc_pub.objects        = [clock, energy_reg]

# Steps 3 & 4 — 序列号 / 标志 ID，然后 Server
dlms.set_serial_number(SERIAL)
dlms.set_flag_id(FLAG_ID)
server = dlms.Server(serial_number=SERIAL, flag_id=FLAG_ID)

# Step 5 — 添加所有对象
for obj in [clock, energy_reg, event_code, assoc_pub]:
    server.add_object(obj)

# Step 6 — 事件代码（run 之前）
server.set_event_code(event_code)

# Step 7 — 连接
hdlc = dlms.IecHdlcSetup("0.0.22.0.0.255", commSpeed=9600)
server.add_object(hdlc)
conn = dlms.SerialConnection(uart_port=2, hdlc_setup=hdlc)
server.add_connection(conn)

# Step 9 — 启动
server.run()

# Step 10 — 应用循环
while True:
    energy_reg.value = read_energy_sensor()
    import utime
    utime.sleep(60)

---

将对象组织到模块中

按子系统分组（能源、安全、网络、预付费等），便于条件编译。若某个板型不支持预付费，只需省略 setup_prepayment_objects() 调用，相关对象永远不会进入 add_object() 。启动序列的其他部分无需改动。

---

多连接

服务器最多支持 8 个 同时连接，每个在独立线程中运行。SN 和 LN 关联可在同一服务器上共存：

• 一个 SerialConnection 使用 use_logical_name=True 服务 LN 客户端
• 另一个 SerialConnection 使用 use_logical_name=False （不同 UART 端口）服务 SN 客户端
• 一个 MobileConnection 可与两者同时运行

每个连接独占其 UART。两个 SerialConnection （或 GenericConnection ）实例 不能共享 同一 UART 端口号。

---

事件代码和事件日志

server.set_event_code() 将 C 层的内部事件代码跟踪连接到两个 Python 对象，使 C 层检测到的事件代码变化自动反映到对象中，无需 Python 侧轮询。

import dlms

event_code = dlms.Data("0.0.96.11.0.255")
event_code.value = 0

event_log = dlms.ProfileGeneric(
    "0.0.99.98.0.255",
    capture_objects=[(clock, 2, 0), (event_code, 2, 0)],
    profile_entries=200,
)

server.set_event_code(event_code, event_log)

启动后，写入 event_code.value 且值发生变化时，服务器自动向 event_log 捕获一行：

event_code.value = 255  # "power fail" — 自动捕获到 event_log

server.set_event_code() 必须 在 server.run() 之前调用，之后调用无效。

---

RegisterMonitor 与后台监控

server.run() 启动后，服务器创建一个后台监控线程，每秒唤醒一次检查所有 RegisterMonitor 阈值。阈值被触发时执行对应的 ScriptTable 动作。

若应用更新了受监控寄存器的值并希望立即检查阈值（无需等待最多 1 秒），可调用 server.monitor() ：

energy_reg.value = read_energy_sensor()
server.monitor()  # 立即检查；若阈值被触发则执行 ScriptTable 动作

• server.monitor() 返回已检查的连接数
• 若监控返回错误则抛出 RuntimeError
• 可从主应用线程安全调用，与后台监控线程并发执行

客户端（Client）

dlms.Client 提供同步接口，用于读取属性、写入属性和调用远程 DLMS/COSEM 服务器的方法。支持与服务器端相同的所有连接类型：串口、光口和蜂窝。所有客户端方法在通信失败时抛出 RuntimeError ，生产代码应将调用包裹在 try/except 中。

---

客户端地址和服务端地址

DLMS 会话由一对地址标识： 客户端地址 和 服务端地址 。

客户端地址 是 DLMS SAP，必须与目标服务器 AssociationLogicalName 上配置的 clientSAP 一致。常用约定值：

值	对应关联
16 ( 0x10 )	公开（无认证）关联
18 ( 0x12 )	High（挑战-响应）认证关联
1	HighGMac（AES-GCM）关联

任何一致的值均可；上述仅匹配常见 DLMS 测试工具的默认值。

服务端地址 由设备序列号派生。HDLC 连接（串口和光口）的计算公式为 serial % 10000 + 1000 。辅助函数 dlms.hdlc_server_address(serial) 执行此计算：

import dlms

addr = dlms.hdlc_server_address(12345)  # 序列号 12345 → 服务端地址 3345

蜂窝中继连接同样适用此公式：中继服务器使用帧中嵌入的 HDLC 服务端地址将数据包路由到正确的设备。

---

构造客户端

Client 构造函数参数：

参数	类型	默认值	说明
client_address	int	16	DLMS SAP
server_address	int	1	HDLC 服务端地址
authentication	int	Authentication.NONE	认证级别常量
password	str / None	None	Low 或 High 认证的密码；HighGMac 不需要
use_logical_name	bool	True	True =LN 引用, False =SN 引用
system_title	bytes / None	None	客户端 8 字节系统标题（HighGMac 必需）
authentication_key	bytes / None	None	GAK（HighGMac 必需）
block_cipher_key	bytes / None	None	GUEK（HighGMac 必需）
security	int	0	SecurityPolicy 值（HighGMac 必需）

公开（无认证）客户端

import dlms

client = dlms.Client(
    client_address=16,
    server_address=dlms.hdlc_server_address(12345),
)

HighGMac 客户端

import dlms

client = dlms.Client(
    client_address=1,
    server_address=dlms.hdlc_server_address(12345),
    authentication=dlms.Authentication.HIGH_GMAC,
    system_title=b'GRX00001',
    authentication_key=b'\xD0\xD1\xD2\xD3\xD4\xD5\xD6\xD7\xD8\xD9\xDA\xDB\xDC\xDD\xDE\xDF',
    block_cipher_key=b'\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0A\x0B\x0C\x0D\x0E\x0F',
    security=dlms.SecurityPolicy.AUTHENTICATED_ENCRYPTED,
)

---

连接与关联

client.connect(connection) 在一个调用中完成物理链路建立和 DLMS AARQ（关联请求）握手。任一步骤失败均抛出 RuntimeError 。

import dlms

hdlc   = dlms.IecHdlcSetup("0.0.22.0.0.255", commSpeed=9600)
conn   = dlms.SerialConnection(uart_port=2, hdlc_setup=hdlc)
client = dlms.Client(
    client_address=16,
    server_address=dlms.hdlc_server_address(12345),
)

client.connect(conn)
# ... 执行读写操作 ...
client.disconnect()

会话状态属性

属性	说明
client.connected	物理传输打开后为 True
client.associated	AARQ/AARE 握手成功后为 True ，DLMS 会话激活
client.currentAssociation	当前活动会话的 AssociationLogicalName 对象；未连接时为 None ，可用于检查协商的一致性块和 PDU 大小

client.disconnect() 发送 RLRQ（释放请求），等待 RLRE，然后关闭物理传输。

---

读写操作

所有读写方法均为同步：阻塞直到服务器响应或抛出 RuntimeError 。

单属性读取

import dlms

clock = dlms.Clock("0.0.1.0.0.255")
t = client.read(clock, clock.idx('time'))  # 读取属性 2 (time)
print("Time: {}-{:02d}-{:02d} {:02d}:{:02d}:{:02d}".format(*t))

使用 obj.idx('attr_name') 代替裸整数，使意图清晰易懂。

批量读取

传递 None 作为属性索引，在一次请求中读取所有持久属性：

reg = dlms.Register("1.0.1.8.0.255", 0, scaler=-3)
client.read(reg, None)  # 读取 value 和 scaler
print("Energy: {} Wh".format(reg.value * (10 ** reg.scaler)))

单属性写入

client.write(clock, clock.idx('time_zone'), 60)  # 设置 UTC+1

批量写入

传递 None 将所有当前存储在对象上的属性写回服务器：

clock.time_zone = 120
client.write(clock, None)

方法调用

client.method(disconnect_ctl, 1)         # 调用 remote_disconnect（方法 1）
client.method(clock, 6, 30)              # shift_time(30 秒)

多属性批量读取

read_multiple 发送单个 Get-Request-With-List，按输入顺序返回值列表：

objs = [(clock, clock.idx('time')), (reg, reg.idx('value'))]
vals = client.read_multiple(objs)
print("Time: {}, Energy: {}".format(vals[0], vals[1]))

---

读取负荷曲线

client.read_profile(profile, start_index, count) 从 ProfileGeneric 对象读取缓冲行。返回值是行列表，每行本身是一个列表，元素与 capture_objects 声明匹配。

import dlms

profile = dlms.ProfileGeneric(
    "1.0.99.1.0.255",
    capture_objects=[(clock, 2, 0), (reg, 2, 0)],
)

# 读取第 1–10 行
rows = client.read_profile(profile, start_index=1, count=10)
for row in rows:
    timestamp, energy = row[0], row[1]
    print("{}: {} Wh".format(timestamp, energy))

# 读取所有可用行
all_rows = client.read_profile(profile, start_index=1, count=0)

---

关联视图发现

当目标服务器未知或其对象列表可能变化时（如调试或互操作性测试），客户端可查询服务器的关联视图，而非本地构造对象。

get_object_info()

获取关联视图，返回 (class_id, version, logical_name_str) 元组列表：

info = client.get_object_info()
for class_id, version, ln in info:
    print("class={:3d}  ver={}  ln={}".format(class_id, version, ln))

get_objects()

执行关联视图查询，返回活的、带类型的 DLMS 对象列表。 dlms 模块中未实现的类 ID 会被静默跳过。

objects = client.get_objects()
print("Association contains {} objects".format(len(objects)))

# 找到第一个 Clock 并填充
for obj in objects:
    if isinstance(obj, dlms.Clock):
        client.read(obj)  # 批量读取所有持久属性
        print("Clock {}  time={}".format(obj.logical_name, obj.time))
        break

get_objects() 执行一次网络往返，应在每个会话中调用一次并复用结果。

---

蜂窝中继客户端

当服务器在 CGNAT 后使用 relay_tcp_setup 时，客户端也通过中继连接。中继根据帧中嵌入的 HDLC 服务端地址识别目标设备，因此客户端侧无需特殊配置，只需使用正确的服务端地址：

import dlms

relay_tcp = dlms.TcpUdpSetup("0.0.25.0.0.254", port=4060)
relay_tcp.ip_reference = ipv4_setup  # ipv4_setup.ipAddress = 中继服务器 IP

mobile = dlms.MobileConnection(
    tcp_udp_setup=dlms.TcpUdpSetup("0.0.25.0.0.255", port=4059),
    gprs_setup=gprs,
    gsm_diag=gsm,
    recv_buffer=bytearray(4096),
    relay_tcp_setup=relay_tcp,
)

client = dlms.Client(
    client_address=16,
    server_address=dlms.hdlc_server_address(12345),
)
client.connect(mobile)
value = client.read(reg, reg.idx('value'))
client.disconnect()

中继使用公式 serial % 10000 + 1000 将帧映射到已注册的设备。两个序列号在取模 10000 后结果相同的设备会发生冲突；分配序列号时应避免此情况。

持久化（Persistence）

嵌入式设备上的 DLMS 服务器必须在断电重启后保持对象状态。 dlms 模块提供 BinarySerializer 作为主要持久化机制，同时还有 JsonSerializer 和可自定义的 Serializer 基类。

---

BinarySerializer

每个 DLMS 对象保存为独立二进制文件，文件名来自 OBIS 代码（如 1.1.33.25.0.255.bin ）。目录必须预先创建。

import dlms

ser = dlms.BinarySerializer("/usr/dlms")

核心方法

方法	说明
ser.save_all(server)	序列化服务器上所有已注册的对象
ser.load_all(server)	恢复所有对象； .bin 文件不存在时静默跳过（首次启动安全）
ser.save(obj)	保存单个对象
ser.load(obj)	恢复单个对象
ser.get_size(server)	返回已保存文件占用的总字节数（不修改文件）

典型启动流程

import dlms

# 创建对象
reg   = dlms.Register("1.0.1.8.0.255", 0, scaler=-3)
clock = dlms.Clock("0.0.1.0.0.255")

server = dlms.Server(serial_number=12345, flag_id="GRX")
server.add_object(reg)
server.add_object(clock)

# 恢复持久化状态（首次启动无效果）
ser = dlms.BinarySerializer("/usr/dlms")
ser.load_all(server)

# 启动服务器
dlms.set_serial_number(12345)
dlms.set_flag_id("GRX")
server.run()

---

序列化规则

BinarySerializer 根据属性标志决定序列化内容。可以用 obj.attrs() 查看实际会被保存的属性：

reg = dlms.Register("1.0.1.8.0.255", 0)
for index, name in reg.attrs():
    print("attr {}: {}".format(index, name))

自动排除的标志：

标志	说明
AttributeFlag.VOLATILE	持续变化的属性（如 Clock.time 、 GsmDiagnostic.status ），自动排除
AttributeFlag.COMPLEX	结构化属性（如 ProfileGeneric.buffer ），自动排除，需自定持久化
只读属性	如 Unit/Scaler，同样自动排除

---

忽略特定属性

ser.ignore() 可从序列化中排除特定属性。可以按类排除（所有实例）或按实例排除。

import dlms

ser = dlms.BinarySerializer("/usr/dlms")

# 对所有 ProfileGeneric 对象跳过 buffer 属性（属性 2）
ser.ignore(dlms.ProfileGeneric, 2)

# 对特定 Register 跳过 scaler/unit（属性 3）
ser.ignore(my_special_reg, 3)

---

ProfileGeneric 缓冲区持久化

ProfileGeneric.buffer 因标记为 COMPLEX 被自动排除。推荐模式：新行捕获时追加到文件，而非一次性序列化整个缓冲区。

import dlms
import utime

BUFFER_PATH = "/usr/dlms/1.0.99.1.0.255.buf"

def _encode_row(ts, energy):
    return "{},{}\n".format(ts, energy).encode()

def on_capture(profile, event):
    if event.index == 2:  # capture
        with open(BUFFER_PATH, "ab") as f:
            f.write(_encode_row(utime.time(), energy_reg.value))
    return True

启动时按行计数恢复 entries_in_use ：

def _count_entries(path):
    count = 0
    try:
        with open(path, "rb") as f:
            while f.readline():
                count += 1
    except OSError:
        pass  # 文件尚不存在
    return count

注意 ：切勿将大缓冲区文件整个加载到 Python 列表中计数或遍历。应逐行读取，否则可能超出堆内存。

---

JsonSerializer

JsonSerializer 定义在 dlms_serializer.py 中，使用 JSON 格式保存，便于调试查看。

from dlms_serializer import JsonSerializer

ser = JsonSerializer("/usr/dlms")
ser.save_all(server)

限制： 二进制属性（安全密钥、OCTET STRING 等）无法在标准 JSON 中无损表示。生产环境使用 BinarySerializer ， JsonSerializer 仅限开发和测试。

---

自定义序列化器

可继承 Serializer 基类实现自定义后端（EEPROM、SD 卡等）。

from dlms_serializer import Serializer

class EepromSerializer(Serializer):
    def save_all(self, server):
        for obj in server.object_registry:
            self.save(obj)

    def load_all(self, server):
        for obj in server.object_registry:
            self.load(obj)

    def save(self, obj):
        for index, name in obj.attrs():
            if self._is_ignored(obj, index):
                continue
            value = getattr(obj, name)
            self._dev.write(obj.logical_name, index, value)

    def load(self, obj):
        for index, name in obj.attrs():
            if self._is_ignored(obj, index):
                continue
            value = self._dev.read(obj.logical_name, index)
            if value is not None:
                setattr(obj, name, value)

    def _is_ignored(self, obj, index):
        for target, attr in self._ignored:
            if attr == index and (target is type(obj) or target is obj):
                return True
        return False

---

文件系统路径与存储

路径	说明
/usr/	内部闪存，始终可用，适用于配置和小型对象状态
/bak/	备份分区，由 FOTA/OTA 工具链管理， 不可写入
/ext/	SPI NOR 闪存（需硬件支持并通过 QPyCOM 启用）
/sd/	SD 卡，运行时挂载

目录需在首次保存前创建：

import uos

def _ensure_dirs():
    for path in ("/usr/dlms",):
        try:
            uos.mkdir(path)
        except OSError:
            pass  # 已存在

_ensure_dirs()

挂载 SD 卡（SPI，EC600N/EC800N 系列）

import uos

cdev = uos.VfsFat(1, 0, 4, 1)  # SPI port 1, mode 0, 13 MHz, CS=GPIO1
uos.mount(cdev, '/sd')

with open('/sd/test.txt', 'w+') as f:
    f.write('hello')
uos.listdir('/sd')

挂载 SD 卡（SDIO，EC600U/EC200U/EC200A 系列）

from uos import VfsSd
import uos

udev = VfsSd("sd_fs")
uos.mount(udev, '/sd')
udev.set_det(udev.GPIO10, 0)  # 可选：配置卡检测引脚

with open('/sd/dlms/lp.jsonl', 'a') as f:
    f.write('[1714000000,12345]\n')

容量警告： /usr/ 是内部闪存文件系统，与所有应用文件和固件脚本共享。保存大量 COSEM 对象（尤其是 ProfileGeneric 日志）可能耗尽空间。建议使用外部存储。

DLMS UDP 路由器（Router）

用途与 CGNAT 穿透

蜂窝模块由运营商 CGNAT 分配私有 IP 地址，头端系统无法向设备的 IP 发起入站连接。DLMS UDP 路由器作为一个公网 IP 的会合点解决此问题：

• 设备向路由器发起 出站 UDP 连接并注册自身
• 头端系统连接到路由器的独立端口
• 路由器检查帧中嵌入的 HDLC 目标地址，在头端和设备间转发数据报

---

架构和地址映射

路由器监听两个独立的 UDP 套接字：

套接字	默认端口	连接方	用途
Board socket	4059	DLMS 服务器设备	设备注册、设备→客户端回复
Client socket	4060	头端/DLMS 客户端	客户端→设备请求

设备注册： 设备发送 BOARD:<serial>\n 作为第一个 UDP 数据报到端口 4059。例如设备序列号 METER-12345 发送：

BOARD:METER-12345\n

路由器从序列号中提取数字部分（去掉非数字字符），然后计算 HDLC 目标地址：

address = (numeric_serial % 10000) + 1000

METER-12345 的数字部分为 12345 ，因此 (12345 % 10000) + 1000 = 3345 。有效地址范围 1000–10999。

路由客户端数据报： 数据报到 4060 端口时，路由器解析帧头中的 HDLC 目标地址，查找已注册的匹配设备，转发数据报到该设备的 UDP 端点。

地址冲突： 两个序列号模 10000 结果相同的设备（如 12345 和 22345 ）会被分配相同 HDLC 地址。路由器会记录警告并将流量转发给最近注册的设备。确保所有部署的序列号后四位唯一。

---

运行路由器

路由器是独立的 Python 3.8+ 包，位于 dlms_udp_router/ 。

# 本地运行
cd dlms_udp_router
python -m pip install .
dlms-udp-router --host 0.0.0.0 --board-port 4059 --client-port 4060 --log-level INFO

# Docker
cd dlms_udp_router/docker
docker compose up --build

# systemd（Linux 服务器）
sudo cp dlms_udp_router/etc/dlms_udp_router.service /etc/systemd/system/
sudo systemctl daemon-reload
sudo systemctl enable --now dlms_udp_router.service

环境变量配置

变量	默认值	说明
ROUTER_HOST	0.0.0.0	绑定地址
ROUTER_BOARD_PORT	4059	设备注册 UDP 端口
ROUTER_CLIENT_PORT	4060	DLMS 客户端 UDP 端口
ROUTER_LOG_LEVEL	INFO	日志级别
ROUTER_SYSLOG_HOST	(无)	远程 syslog 主机(UDP)
ROUTER_SYSLOG_PORT	(无)	远程 syslog 端口

---

设备侧配置

设备通过 MobileConnection 的 relay_tcp_setup 参数向路由器注册。 relay_tcp_setup 指向路由器的 board 端口（4059）。

import dlms

SERIAL    = 12345678
RELAY_IP  = "203.0.113.10"  # 路由器服务器公网 IP
RELAY_PORT = 4059            # 设备注册端口

tcp_udp = dlms.TcpUdpSetup("0.0.25.0.0.255", port=4059)

relay_setup = dlms.TcpUdpSetup("0.0.25.0.0.254", port=RELAY_PORT)
relay_setup.ipReference = ipv4_obj  # ipv4_obj.ipAddress = RELAY_IP

gprs     = dlms.GprsSetup("0.0.2.0.0.255")
gsm      = dlms.GsmDiagnostic("0.0.25.6.0.255")
recv_buf = bytearray(4096)

mobile = dlms.MobileConnection(
    tcp_udp_setup=tcp_udp,
    gprs_setup=gprs,
    gsm_diag=gsm,
    recv_buffer=recv_buf,
    relay_tcp_setup=relay_setup,
)
mobile.on_connected    = lambda: print("relay: connected")
mobile.on_disconnected = lambda: print("relay: disconnected")
server.add_connection(mobile)

server.run() 启动后，运行时在每次（重）连接时自动发送 BOARD:<serial>\n 注册数据报。

主动推送

中继模式下，设备可主动推送帧到当前连接的客户端：

mobile.send(raw_dlms_frame_bytes)

用于 PushSetup 通知。若中继连接未激活则抛出 RuntimeError 。

---

客户端侧使用

路由器对标准 DLMS/HDLC 流量透明。客户端发送正常 HDLC 请求帧到路由器的 client 端口（4060）；路由器读取帧中的 HDLC 目标地址并转发给匹配的设备。

计算目标 HDLC 地址

import dlms

BOARD_SERIAL = 12345678
server_address = dlms.hdlc_server_address(BOARD_SERIAL % 10000)

QuecPython 客户端通过中继连接

import dlms

BOARD_SERIAL = 12345678
RELAY_IP     = "203.0.113.10"
CLIENT_PORT  = 4060

tcp_udp = dlms.TcpUdpSetup("0.0.25.0.0.255", port=4060)
relay   = dlms.TcpUdpSetup("0.0.25.0.0.254", port=CLIENT_PORT)
relay.ipReference = ipv4_obj  # ipv4_obj.ipAddress = RELAY_IP

gprs     = dlms.GprsSetup("0.0.2.0.0.255")
gsm      = dlms.GsmDiagnostic("0.0.25.6.0.255")
recv_buf = bytearray(4096)

mobile = dlms.MobileConnection(
    tcp_udp_setup=tcp_udp,
    gprs_setup=gprs,
    gsm_diag=gsm,
    recv_buffer=recv_buf,
    relay_tcp_setup=relay,
)

client = dlms.Client(
    client_address=16,
    server_address=dlms.hdlc_server_address(BOARD_SERIAL % 10000),
    conn=mobile,
)

PC/头端客户端

任何标准 DLMS 客户端库（如 Gurux DLMS.Net、gurux-dlms-python）均可与中继配合使用。将传输指向路由器的公网 IP 和 client 端口（4060）。出站帧中的 HDLC 目标地址必须等于 (BOARD_SERIAL % 10000) + 1000 。

防火墙注意： 确保路由器服务器上的 UDP 端口 4059 和 4060 已开放，入站（来自设备和客户端）和出站（回复）流量均需允许。

最佳实践（Best Practices）

---

内存管理

内存是 QuecPython 开发板上最受限的资源，通常可用堆仅 200–400 KB。

预分配接收缓冲区。 MobileConnection 需要构造时传入 recv_buffer bytearray。一次性分配，切勿重复创建：

recv_buf = bytearray(4096)  # 在模块级别分配，任何线程启动之前
mobile = dlms.MobileConnection(
    tcp_udp_setup=tcp_udp,
    gprs_setup=gprs,
    gsm_diag=gsm,
    recv_buffer=recv_buf,
)

避免在回调或循环内分配 bytearray 对象，重复分配会使堆碎片化，长期运行可能导致 MemoryError 。

对持有 C 端定时器的对象调用 deinit() 。 以下类分配了后台资源，在服务器关闭或对象不再需要时必须显式释放： ProfileGeneric 、 PushSetup 、 ActivityCalendar 、 ScriptTable 、 SingleActionSchedule 。不调用 deinit() 会泄漏定时器并阻止干净重启。

使用 BinarySerializer 持久化状态 ，而非内存结构。大型字典、历史读数列表或完整 ProfileGeneric 缓冲区应保存在文件系统上，而非 Python 堆中。按需读取请求所需的数据（参见第 9.4 节的逐行模式）。

切勿将完整捕获缓冲区加载到 Python 列表。 即使一个适中的 1440 条负荷曲线，一次全部反序列化也可能超出可用堆内存。在 on_before_read 中逐行从文件流式读取。

---

安全加固

部署前更改所有默认密钥。 运行时会初始化密码学密钥为已知测试值。使用这些默认值的设备可被知晓这些值的任何人读取或控制。必须更改的密钥：

• GMAC_GUEK — Global Unicast Encryption Key（加密数据）
• GMAC_GAK — Global Authentication Key（认证帧）
• GMAC_KEK — Key Encryption Key（保护密钥更新包裹）

在 SecuritySetup 对象上设置：

security_setup.global_unicast_encryption_key = bytes.fromhex("YOUR_32_HEX_CHARS_GUEK")
security_setup.global_authentication_key     = bytes.fromhex("YOUR_32_HEX_CHARS_GAK")

KEK 通过 dlms.set_kek() 在启动时一次性传递给运行时。

将 KEK 存储在受保护存储中。 KEK 是最敏感的密钥，因为它在密钥更新过程中包裹所有其他密钥。它 不能 以字符串字面量出现在 config.py 中。使用 Quectel 的保护闪存分区或硬件安全元件。至少，将其存储在一个 USB 大容量存储模式不可访问的分区文件中。

对承载真实计量数据的连接使用 HighGMac 认证。 AuthenticationLevel.HIGH_GMAC 提供双向密码学认证和每帧重放保护。较低的认证级别（NONE、LOW）仅保留给对象列表经过严格限制的只读诊断关联。

将未认证访问限制为最小对象集。 任何 auth_mechanism = 'None' 的 AssociationLogicalName 在无凭证情况下公开可读。其访问字典最多应暴露少量标识对象（如 clock、固件版本）。 切勿 在未认证关联中包含可写对象或安全相关对象。

---

线程安全

SerialConnection 和 MobileConnection 在 C 管理的后台线程中运行 I/O 循环。COSEM 对象上注册的事件处理程序从这些线程调用。主 Python 线程同时运行应用循环。这会产生共享状态风险。

保持事件处理程序简短且非阻塞。 阻塞（sleep、等待锁、慢速文件系统写入）的事件处理程序会延迟连接线程，可能导致客户端超时。若需在事件响应中执行重要工作，设置标志并在主循环中处理：

_capture_pending = [False]

def on_before_action_capture(profile, event):
    if event.index == 2:
        _capture_pending[0] = True
    return True

# 在主应用循环中：
while True:
    if _capture_pending[0]:
        _capture_pending[0] = False
        # ... 在此处理慢速工作 ...
    utime.sleep_ms(100)

当处理程序和主循环都写入同一对象时，用锁保护共享状态。 MicroPython 的 _thread.allocate_lock() 提供简单互斥锁：

import _thread

_lock = _thread.allocate_lock()
_shared_value = [0]

def on_before_write(obj, event):
    with _lock:
        _shared_value[0] = event.value
    return True

# 在主循环中：
with _lock:
    v = _shared_value[0]

在紧循环中使用 utime.sleep_ms(0) 出让 CPU。 MicroPython 调度器对 Python 线程是协作式的。主循环迭代超过几毫秒不让出可能导致连接线程饥饿。在任何紧轮询循环中插入 utime.sleep_ms(0) （或一个小的正值）。

---

连接可靠性

通过 on_disconnected 响应断连。 每个连接类都暴露 on_disconnected 回调。注册一个至少记录事件的处理程序，并可选地重启受影响的连接或触发 modem 重置：

def _on_disconnected():
    print("connection lost; scheduling reconnect")
    _reconnect_pending[0] = True

mobile.on_disconnected = _on_disconnected

在 IecHdlcSetup 上设置 inactivity_timeout 以检测静默断连。 未发送 DISC 帧即消失的客户端会使服务器无限等待。设置非零的 inactivity_timeout （秒）使运行时在指定时间内无流量后关闭并重新打开连接：

hdlc = dlms.IecHdlcSetup(
    "0.0.22.0.0.255",
    commSpeed=9600,
    deviceAddr=0x10,
    inactivity_timeout=120,  # 静默 2 分钟后关闭
)

监控 GsmDiagnostic.status 以检查蜂窝链路。 在主循环中定期调用 gsm_diag.update() 并检查 gsm_diag.status 。值 1 （HOME_NETWORK）或 5 （ROAMING）表示注册激活；其他值表示设备不可达：

import utime

STATUS_REGISTERED = (1, 5)  # HOME_NETWORK, ROAMING

while True:
    gsm_diag.update()
    if gsm_diag.status not in STATUS_REGISTERED:
        print("network lost; status={}".format(gsm_diag.status))
        # 等待重新注册后再尝试重连
    utime.sleep_ms(60000)  # 每 60 秒检查一次

---

无硬件测试

DLMS 服务器逻辑的端到端测试通常需要物理开发板和 DLMS 客户端探头。 GenericConnection 提供了替代方案：由于其传输完全由 Python 驱动，可以在单个 Python 进程中构建回环，无需任何硬件即可执行请求-响应周期。

模式是创建一个带 GenericConnection 的服务器和一个客户端，将 on_send 和 on_receive 钩子反向连接到 process_msg ：

import dlms

# 最小服务器
hdlc = dlms.IecHdlcSetup("0.0.22.0.0.255")
conn = dlms.GenericConnection(
    interface_type=dlms.InterfaceType.HDLC,
    hdlc_setup=hdlc,
)
server = dlms.Server(serial=99999, use_logical_name=True)
data_obj = dlms.Data("1.0.1.8.0.255")
data_obj.value = 42
server.add_object(data_obj)
server.add_connection(conn)
conn.connect()
server.run()

# 将帧直接发送到 process_msg 的客户端
client = dlms.Client(
    client_address=16,
    server_address=dlms.hdlc_server_address(99999),
    conn=conn,
)

def loopback_send(data):
    resp = conn.process_msg(data)
    if resp:
        conn.process_msg(resp)  # 将响应反馈回去

client.on_send = loopback_send
client.connect()
value = client.read(data_obj, 2)  # 读取属性 2
print(value)                       # 42
client.disconnect()

这种技术对于验证属性访问控制规则、确认 on_before_read / on_before_write 处理程序返回正确值，以及检查序列化往返都非常有用——全部在 PC 侧 Python 环境中完成。

客户端方法 （ connect 、 read 、 write 、 action ）是同步的，失败时抛出 RuntimeError 。在生产代码中，始终将它们包裹在 try/except 块中，以便优雅处理断连或意外响应，而非崩溃调用线程。

---

代码组织

大型部署受益于将应用拆分为集中的模块，而非将所有代码放在 main.py 中。 server_example/objects/ 目录展示了一种有效的布局：

config.py
    所有部署特定常量：OBIS 地址、数字序列号、APN、密码学密钥、端口号。
    这是不同开发板或客户的固件构建之间唯一不同的地方。

objects/data_objects.py
    Data、Register 和 ExtendedRegister 实例及其初始值。

objects/profile_objects.py
    ProfileGeneric 实例、捕获对象列表、缓冲区持久化辅助函数。

objects/security_objects.py
    SecuritySetup、AssociationLogicalName、AssociationShortName、访问字典。

objects/connection_objects.py
    IecHdlcSetup、TcpUdpSetup、GprsSetup、MobileConnection、SerialConnection。

main.py
    导入以上所有模块，调用 server.add_object() 和 server.add_connection()，
    调用 server.run()，然后进入应用循环。

启动时创建存储目录。 切勿假设 /usr/dlms/ 或 /sd/dlms/ 已存在。在实例化任何序列化器之前的启动阶段调用一次辅助函数，可防止首次写入时出现 OSError ：

import uos

DIRS = ["/usr/dlms", "/usr/dlms/objects"]

def _ensure_dirs():
    for path in DIRS:
        try:
            uos.mkdir(path)
        except OSError:
            pass  # 已存在

_ensure_dirs()

保持 config.py 不含逻辑。 配置常量应为简单赋值。若某个值需要计算（如从序列号派生 HDLC 地址），计算一次——若开销小则在导入时计算，否则在 main.py 调用的 init() 函数中计算。避免 config.py 中出现依赖运行时状态的条件逻辑；该逻辑属于 main.py 或相关模块。

术语表（Glossary）

---

APDU（Application Protocol Data Unit） — DLMS 栈中最顶层的 PDU。APDU 携带 COSEM 服务请求或响应，并传递给传输层进行成帧。

---

Association（关联） — DLMS 客户端和服务器之间建立的逻辑会话。每个关联具有协商的认证级别、密码套件和一致性位集。在 dlms 模块中，关联由 AssociationLogicalName （LN 引用）和 AssociationShortName （SN 引用）对象表示。

---

Authentication（认证） — 验证连接客户端身份的过程。 dlms 模块支持的级别从 NONE （开放访问）到 HIGH_GMAC （AES-GCM 双向认证）。参见第 5 章。

---

BinarySerializer — dlms 序列化辅助工具，将 COSEM 对象属性状态保存到设备文件系统上的二进制文件并恢复。标记为 AttributeFlag.COMPLEX （如 ProfileGeneric.buffer ）的属性被排除。参见第 9 章。

---

CGNAT（Carrier-Grade NAT） — 移动运营商运行的网络地址转换层，为蜂窝设备分配私有 IP 地址，使设备无法接受直接入站 TCP 连接。DLMS UDP 路由器提供了绕过此限制的中继方案。参见第 10 章。

---

COSEM（Companion Specification for Energy Metering） — IEC 62056 标准的数据模型部分，定义了接口类目录及其属性。 dlms 模块中的每个 Python 类对应一个 COSEM 接口类。

---

Default behaviour（默认行为） — 当无事件处理程序拦截请求时，服务器 C 层自动执行的处理。对于 GET 请求意味着序列化当前属性值；对于 ACTION 请求意味着分发内置方法实现。事件处理程序可以观察或替换默认行为。参见第 4 章。

---

DLMS（Device Language Message Specification） — IEC 62056 标准的协议部分，定义 COSEM 对象如何序列化为 APDU 以及 APDU 如何传输。"DLMS" 通常非正式地指代 DLMS/COSEM 组合标准。

---

DLMSEvent — 传递给每个事件处理程序的上下文对象。携带当前请求的属性或方法索引、选择器类型、选择器参数和动作标志。参见第 4.1 节。

---

GAK（Global Authentication Key） — 16 或 32 字节的 AES 密钥，用于在 HighGMac 安全模式下计算 DLMS 帧的 GMAC 认证标签。服务器和客户端必须相同。

---

GenericConnection — dlms 连接类，其传输 I/O 循环由 Python 编写。调用者打开物理通道，将接收到的字节传递给 process_msg() ，并将返回的字节写回通道。用于 UART、SPI、MQTT、G3-PLC 以及无需硬件的单元测试。参见第 6.4 节和第 11.5 节。

---

GUEK（Global Unicast Encryption Key） — 16 或 32 字节的 AES 密钥，用于在 HighGMac 安全模式下加密 DLMS APDU。有时写作 GMAC_GUEK。服务器和客户端必须相同。

---

HDLC（High-Level Data Link Control） — DLMS 在串行传输（RS-232、RS-485、光口）上使用的数据链路成帧层。提供寻址、成帧和流控制。HDLC 服务器地址由 dlms.hdlc_server_address() 从设备序列号派生。

---

HighGMac — dlms 模块中最高的认证级别（ Authentication.HIGH_GMAC ）。使用 AES-GCM 提供每帧认证和可选加密。需要配置了 guek 和 gak 密钥的 SecuritySetup 对象。参见第 5.2 节。

---

IEC 62056 — DLMS/COSEM 的国际标准系列。蓝皮书（IEC 62056-62）定义 COSEM 接口类；绿皮书（IEC 62056-5-3）定义安全；黄皮书涵盖一致性。

---

Interface class（接口类） — COSEM 中对对象类型的术语。每个接口类具有数字标识符（类 ID）并定义一组固定的属性和方法。例如， Register 是接口类 3， Clock 是接口类 8。

---

Invocation counter（调用计数器） — 每个 GMAC 保护帧中包含的单调递增的 32 位整数。服务器拒绝计数器不大于上次接受值的帧，防止重放攻击。计数器必须在断电重启间持久化。参见第 5.3 节。

---

KEK（Key Encryption Key） — 用于在密钥更新过程中包裹（加密）GUEK 和 GAK 的主密钥。在启动服务器前通过 dlms.set_kek() 设置。必须存储在设备的受保护存储中。参见第 5.4 节。

---

LN referencing（逻辑名引用） — COSEM 属性通过 OBIS 代码和属性索引标识的寻址模式。由 AssociationLogicalName 使用。这是 dlms 模块中的默认模式。

---

MobileConnection — dlms 连接类，管理 C 驱动的蜂窝 UDP 套接字。支持直连模式（来自任何客户端的入站 UDP）和中继模式（向 DLMS UDP 路由器的出站注册）。需要预分配的 recv_buffer 。参见第 6.3 节和第 10.4 节。

---

OBIS code（对象标识系统） — 六组数字标识符 A.B.C.D.E.F ，唯一标识设备上的 COSEM 对象实例。每个 dlms 对象构造函数接受 OBIS 字符串作为其第一个参数。

---

PDU（Protocol Data Unit） — 特定协议层上的结构化数据单元。在 DLMS 应用层为 APDU；在 HDLC 层为 HDLC 帧。

---

ProfileGeneric — COSEM 接口类 7。存储历史时间序列数据（负荷曲线、事件日志）的标准对象。其 buffer 属性标记为 COMPLEX，必须与 BinarySerializer 分开持久化。参见第 3.6 节和第 9.4 节。

---

QuecPython — 嵌入 Quectel 蜂窝模组中的 MicroPython 1.13 运行时。 dlms 模块是该运行时的编译 C 扩展。

---

SAP（Service Access Point） — DLMS 系统中逻辑设备或客户端的数字标识符。标准管理客户端 SAP 为 16。SAP 到逻辑设备的映射保存在 SapAssignment 中。

---

SecuritySetup — COSEM 接口类 64。持有一个安全上下文的密码学密钥、安全策略和系统标题。每个使用 HighGMac 的关联需要一个 SecuritySetup 对象。参见第 5.3 节。

---

SN referencing（短名引用） — 使用 16 位短名代替 OBIS 代码的替代 COSEM 寻址模式。由 AssociationShortName 使用。在现代部署中较少见；LN 引用更受青睐。

---

System title（系统标题） — DLMS 实体的 8 字节标识符，用作 GCM 初始化向量的一部分。常规格式：3 字节 ASCII 标志标识符后跟 5 字节序列号。存储在 SecuritySetup.server_system_title 和 client_system_title 中。

---

WRAPPER — 用于在 UDP 或 TCP 上承载 DLMS APDU 的薄成帧层（IEC 62056-47）。与 HDLC 不同，它不添加寻址；路由由 IP 层处理。通过 GenericConnection 上的 InterfaceType.WRAPPER 选择。