zk-view对接python数据源

---

通过该文件提供的方法，实现python和zk-view的数据源对接，使用python编程，做图像识别、算法等场景，可以通过该文件提供的方法，实现数据、图像和zk-view对接组态，实时显示，大屏可视化

下载地址：http://zkview.com 视频教程

数据源代码下载地址

---

文件介绍

websocket_server.py
python启动websocket的主服务

Data.py
所有实时数据的缓存列表

test.py
用于测试随机数发送的文件

---

测试使用

1. 启动 test.py 文件
2. 打开zk-view,添加数据源地址： ws://localhost:9601/zkSource
3. 打开添加的数据源，点击原始点管理进行数据查看和添加

---

使用步骤

1. 数据源注册

websocket_server.py 中 connection_handler方法中
对应修改 elif msgName == "ORIGINAL":中的代码

示例1：全部测点一次性注册

elif msgName == "ORIGINAL":
  # 1. （必须）发送全部测点进行数据源添加，通过"ORIGINALDATA"标志发送所有的实时测点数据
  await self.send_data(websocket, {"name": "ORIGINALDATA", "data": testData})
  # 2. （可选）通过"NAMEMAP"发送对用中文名称的对应字段数据
  await self.send_data(websocket, {"name": "NAMEMAP", "data": nameMap})

示例2：发送带有分区的数据进行下拉框选择注册

elif msgName == "ORIGINAL":
  # 模拟分为三个区域
  var area = [
    {"name": "1号区域", "data": "area1"},
    {"name": "2号区域", "data": "area2"},
    {"name": "3号区域", "data": "area3"},
  ]
  # 发送分区信息，分区信息使用 "ORIGINAL" 标识
  await self.send_data(websocket, {"name": "ORIGINAL", "data": testData})

  # 示例：获取设置的区域参数并，发送测点数据到当前用户
  if msgData == "area1":
    # 发送区域1的测点数据
    await self.send_data(websocket, {"name": "ORIGINALDATA", "data": testData})
    # （可选）通过"NAMEMAP"发送对用中文名称的对应字段数据
    await self.send_data(websocket, {"name": "NAMEMAP", "data": nameMap})
  elif msgData == "area2"
    # ...
  else
    # ...

---

2. 用户节点注册

websocket_server.py 中 connection_handler方法中
对应修改 elif msgName == "REGISTER":中的代码

注册测点登记接口，前端图纸启动时，会将导入数据源的测点信息进行发送,并在 “REGISTER” 中收到
该接口用户切勿修改，可进行逻辑处理
接收到的值为：图纸上对该数据源的所有注册点位信息
用户可根据注册的点位信息筛选发送的数据
注意： 每一个数组对应一个测点，拼接起来即为完整测点key，如收到注册点[“c”,“b”][“c”,“cc”,“ccb”]，即为注册两个测点，分别为 c.b 和 c.cc.ccb
可以根据每个用户注册的点，单独发送注册点数据，以节约服务器资源

示例：注册一级节点

elif msgName == "REGISTER":
  # 遍历收到的多级测点key列表，将1级key加入到注册列表中
  for keys in msgData:
      self.registerSet.add(keys[0])
  print("收到用户注册测点")
  # 最后在发送的时候，可以在 全局发送`broadcast_data` 或 单个用户发送`send_data`方法中
  # 调用self.registerSet 获取所有需要发送的测点，进行筛选后发送

3. 发送数据

示例代码：1秒发送一次数据

# 模拟每秒发送一个随机数给所有连接的用户
async def send_random_numbers(websocket_server):
    while True:
        nameMap['test1'] = '测试1'
        nameMap['test2'] = '测试2'
        nameMap['test3'] = '测试3'
        testData['test1'] = random.uniform(2, 4)
        testData['test2'] = random.uniform(2, 4)
        testData['test3'] = random.uniform(2, 4)
        # logging.info(f"生成随机数: {testData}")
        await websocket_server.broadcast_data({"name": "DATA", "data": testData})
        await asyncio.sleep(1)

4. websocket服务器启动方法

# 主函数
from websocket_server import WebSocketServer
async def main():
    # 实例化WebSocketServer
    websocket_server = WebSocketServer()

    # 启动WebSocket服务器，并执行发送随机数的任务
    await asyncio.gather(websocket_server.start(), send_random_numbers(websocket_server))

if __name__ == "__main__":
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logging.info("WebSocket服务器被用户停止")

5. 修改服务器的路由地址方法，默认 /zkSource

在 websocket_server.py 中 connection_handler方法头部，对应修改path

    # 处理WebSocket连接和收到的消息,使用中,数据源注册等收到消息的方法在该类中进行自定义调用
    async def connection_handler(self, websocket, path):
        # 检查 WebSocket 连接的路径是否为 "/zkSource"
        if path != "/zkSource":
            await websocket.close()
            return
    #...