Python协议插件开发实践

为适应物联网设备多样性及工业协议复杂性，我们选用Python语言开发协议插件。Python凭借其跨平台特性、丰富的协议处理库及动态扩展能力，能够高效适配多设备，构建灵活稳健的物联网系统。其简洁的语法和快速开发特性，进一步降低了协议适配的复杂度。以下是Python开发协议插件的数据流程解析：

# Python开发协议插件数据流程解析

详情

设备层: 各类物联网设备（如传感器、控制器）持续产生运行数据，包括设备状态、环境参数等业务关键信息。
接入层: 通过网络API获取原始数据，Java开发的协议插件负责解析不同设备的通信协议，将数据转换为统一格式。
边缘处理: 收集到的数据暂存在边缘节点（MiniDataCenter），进行初步清洗和标准化处理，确保数据质量。
安全传输: 边缘网关通过MQTT协议将处理后的数据实时推送至云平台，采用加密传输保障数据安全，支持断线重连机制。
平台展示: IoTCenter平台接收数据后，通过可视化界面实时展示设备状态、运行趋势等信息，支持历史数据查询。

# 开发步骤

# 一、开发环境配置

详情

以下是针对 Windows系统 的Python开发环境详细配置流程：

下载安装包：
- 访问Python官网 (opens new window)，下载最新Windows安装包（如 *Python-3.x.x.exe）。
- 注意：选择 Windows installer (64-bit) 版本（根据系统选择32/64位），Python版本不低于3.11。
安装步骤：
- 双击运行安装包，勾选 Add Python to PATH（必须勾选，否则需手动配置环境变量）。
- 点击 Install Now 完成安装。
验证安装：
- 按 Win + R，输入 cmd 打开命令提示符。
- 输入以下命令检查版本：
```
Python --version
pip --version
```
- 若提示“命令不存在”，请重新安装并勾选 Add Python to PATH或配置环境变量。
Windows系统手动配置Python环境变量（可选）

打开环境变量窗口： - 右键 此电脑 → 属性 → 高级系统设置 → 点击 环境变量。

编辑系统变量Path： - 在 系统变量 区域，找到并选中 Path，点击编辑。

添加Python路径： - 点击新建，输入Python安装路径（如 C:\Python39）。 - 再次点击新建，输入Scripts路径（如 C:\Python39\Scripts）。

注意

您可以使用VS Code、PyCharm、IDLE等开发工具完成下列编码。输出一个名为xxx.py的文件即可。如CEquip.py

# 二、随机数代码示例

详情

import random
import threading

from ganweisoft.Database.YcpTable import YcpTableRow
from ganweisoft.Database.YxpTable import YxpTableRow
from ganweisoft.Logging import Logging
from ganweisoft.interface.CEquipBase import CEquipBase
from ganweisoft.interface import CommunicationState


class CEquip(CEquipBase):
    def __init__(self):
        super().__init__()
        self.data_fetch_counter = 0
        self.lock = threading.Lock()

    def GetData(self, pEquip) -> CommunicationState:
        super().Sleep(1000)
        if super().RunSetParmFlag:
            return CommunicationState.setreturn

        comm_state = super().GetData(pEquip)
        if comm_state != CommunicationState.ok:
            return comm_state

        if not pEquip.GetEvent():
            return CommunicationState.fail
        return CommunicationState.ok

    def GetYC(self, row: YcpTableRow) -> bool:
        min_val = max(row.val_min, 0.0)
        max_val = min(row.val_max, 100.0)
        super().SetYCData(row, random.uniform(min_val, max_val))
        return True

    def GetYX(self, row: YxpTableRow) -> bool:
        yx_value = random.choice([True, False])
        super().SetYXData(row, yx_value)
        return True

    def SetParm(self, main_instruct: str, minor_instruct: str, value: str) -> bool:
        try:
            if main_instruct.lower() == "setycyxvalue":
                if minor_instruct is None or len(minor_instruct) < 3:
                    Logging.write_log_file(f"Invalid MinorInstruct format: {minor_instruct}")
                    return False

                prefix = minor_instruct[0].lower()
                index_str = minor_instruct[2:]
                if not index_str.isdigit():
                    Logging.write_log_file(f"Invalid YC/YX index: {index_str}")
                    return False

                ycyx_no = int(index_str)
                if ycyx_no <= 0:
                    Logging.write_log_file(f"YC/YX index must be > 0: {ycyx_no}")
                    return False

                if prefix == 'c':  # YC
                    if value is None or len(value) < 0:
                        Logging.write_log_file("Missing YC value")
                        return False
                    yc_value = float(value)
                    with self.lock:
                        yc_results = super().YCResults
                        yc_results[ycyx_no] = yc_value
                    return True
                elif prefix == 'x':  # YX
                    yx_value = int(value) > 0
                    with self.lock:
                        yx_results = super().YXResults
                        yx_results[ycyx_no] = yx_value
                    return True
            return False
        except ValueError as e:
            Logging.write_log_file(f"Number format error: {str(e)}")
            return False
        except Exception as e:
            Logging.write_log_file(f"SetParm error: {str(e)}")
            return False

if __name__ == '__main__':
    from ganweisoft.DataCenter import DataCenter
    DataCenter.start()

# 三、调试运行

详情

1、配置变量导入

可选：1、使用配置文件。下载方式见下文：IoTCenter 边缘网关部署 -> 应用管理 -> ./config文件夹下

此时的目录结构为：

目录结构：

.
│─CEquip.py
│
└─config
        config.properties

可选：2、使用环境变量，此时可使用set（windows）或者export（Linux）先初始化后，再调用运行命令

set InstanceId=20250304
set MqUsername=gateway
set MqPassword=********
set MqServer=127.0.0.1
set MqPort=1883

2、在xxx.py同级目录下执行命令（windows环境）

python xxx.py

alt text

# 四、打包

详情

1、打包为独立可执行文件

注意

Windows和Linux下打包出来的文件是不相同的，所以在不同操作系统环境中，需独立打包

# 安装pyinstaller
pip install pyinstaller

# 打包CEquip.py成CEquip执行文件
pyinstaller CEquip.py --onefile --hidden-import=gw-mini-datacenter

2、打包Docker镜像包

将第一步生成的文件打包成Docker镜像包

# using debian bookworm image
FROM debian:bookworm-slim

# Continue with application deployment
RUN mkdir /opt/ganwei
COPY IoTCenter.Python /opt/ganwei/IoTCenter.Python

WORKDIR /opt/ganwei/IoTCenter.Python/

CMD ["./CEquip"]