BBIT-Kai
108 lines
2.6 KiB
Python
108 lines
2.6 KiB
Python
from typing import Optional
|
||
|
||
|
||
class MqttTopic:
|
||
"""
|
||
封装 MQTT topic，根据规则:
|
||
project/domain/deviceType/deviceId/resource
|
||
"""
|
||
|
||
LEVELS = 5
|
||
|
||
def __init__(self, topic: str):
|
||
self.raw = str(topic)
|
||
parts = self.raw.split("/")
|
||
|
||
# 不足的层级用 None 补齐，避免属性缺失
|
||
parts += [None] * (self.LEVELS - len(parts))
|
||
|
||
self.project: Optional[str] = parts[0]
|
||
self.domain: Optional[str] = parts[1]
|
||
self.device_type: Optional[str] = parts[2]
|
||
self.device_id: Optional[str] = parts[3]
|
||
self.resource: Optional[str] = parts[4]
|
||
|
||
@classmethod
|
||
def from_parts(
|
||
cls,
|
||
project: Optional[str] = None,
|
||
domain: Optional[str] = None,
|
||
device_type: Optional[str] = None,
|
||
device_id: Optional[str] = None,
|
||
resource: Optional[str] = None,
|
||
) -> "MqttTopic":
|
||
"""
|
||
通过结构化参数构造 topic
|
||
None -> '+'
|
||
"""
|
||
|
||
def _v(v: Optional[str]) -> str:
|
||
return "+" if v is None else str(v)
|
||
|
||
topic = "/".join(
|
||
map(
|
||
_v,
|
||
[
|
||
project,
|
||
domain,
|
||
device_type,
|
||
device_id,
|
||
resource,
|
||
],
|
||
)
|
||
)
|
||
return cls(topic)
|
||
|
||
def to_topic(self) -> str:
|
||
"""
|
||
根据当前字段生成 topic（允许 '+')
|
||
"""
|
||
|
||
def _v(v: Optional[str]) -> str:
|
||
return "+" if v is None else v
|
||
|
||
return "/".join(
|
||
map(
|
||
_v,
|
||
[
|
||
self.project,
|
||
self.domain,
|
||
self.device_type,
|
||
self.device_id,
|
||
self.resource,
|
||
],
|
||
)
|
||
)
|
||
|
||
def build(self) -> str:
|
||
"""
|
||
生成严格 topic（不允许 None / '+'）
|
||
用于 publish 场景
|
||
"""
|
||
parts = [
|
||
self.project,
|
||
self.domain,
|
||
self.device_type,
|
||
self.device_id,
|
||
self.resource,
|
||
]
|
||
|
||
if any(p in (None, "+") for p in parts):
|
||
raise ValueError(
|
||
f"Cannot build strict topic, wildcard exists: {self.to_topic()}"
|
||
)
|
||
|
||
return "/".join(parts)
|
||
|
||
def is_wildcard(self) -> bool:
|
||
return "+" in self.to_topic() or "#" in self.to_topic()
|
||
|
||
def __repr__(self):
|
||
return f"<MqttTopic {self.to_topic()}>"
|
||
|
||
def is_status(self) -> bool:
|
||
return self.domain == "status"
|
||
|
||
def is_cmd(self) -> bool:
|
||
return self.domain == "cmd"
|