1、痛点

一、公司产品类型比较多，上百款机型产品，类别分十几种；
二、搭建一台产品测试工装（水机中央空调主板），大概需要 2 天实际，比如物料领取，流程审批，测试工装组装；
三、APP 迭代开发时，优化或者新增对接 N 款产品功能，项目周期短，APP 端开发思维仅是对接云端数据；
四、设备产品涉及到接电安全，接线错误随时引发元器件烧坏，电路短路，触电风险；
五、公司产品使用自定义协议，不支持云平台提供模拟设备工具；
六、设备的传感器数据不能随意模拟，故障状态不能随意上报；
七、测试工装，搭建一两天，使用一两个小时；暂无设备测试实验室，测试工装物料会重新拆解，投入下个项目中；
八、前端开发者开发思维，往往停留在仅对接后端数据，缺乏对实际产品设备业务理解，紧靠测试工程师测试过程中，将设备搭建完毕后，再使用 APP 联动测试；

2、需求

一、支持模拟阿里云上任何类产品包括 ICA 标准数据格式、透传/自定义类（Modbus 协议）产品；
二、支持云端下发数据时，模拟设备正常应答数据；
三、支持通过配置表格式，修改上报的参数值，可以模拟到设备任何点位数据；
四、支持通过输入项目名称，返回设备三元组信息；
五、支持返回设备二维码，用于 APP 扫码绑定设备；

3、物联网原理分析

4、物联网理论知识

一、三元组

物联网三元组是物联网系统中常用的一种身份识别方式，通常包含设备 ID（Device ID）、设备密钥（Device Key）和设备连接字符串（Device Connection String），也被称为设备标识符。它们的作用在于：
1.身份识别：物联网设备需要向云端或其他设备发送请求或数据，这些数据需要进行身份验证，以确保只有经过身份验证的设备才可以访问数据或云服务。三元组提供了一种简单、可靠的方法来唯一识别每个设备。
2.数据加密：通过使用每个设备的唯一密钥，可以对传输的数据进行加密，并确保数据安全。
3.设备管理：三元组还可以用于对设备进行管理，包括对设备进行注册、配置、更新和监控等操作，以及对设备进行管控和维护。
综上，物联网三元组对于实现物联网系统的安全性和可靠性具有重。

二、MQTT 协议

MQTT（Message Queuing Telemetry Transport）是一种轻量级的发布/订阅消息协议。MQTT 协议被设计用于连接远程设备，使其能够以低带宽和不稳定的网络条件下互相通信。MQTT 协议使用 TCP/IP 协议进行传输，在物联网应用中使用广泛。
MQTT 协议的主要特点包括：

1. 轻量级：MQTT 协议使用紧凑的二进制协议，可以在低带宽的环境中快速传输数据。
2. 支持发布/订阅模式：MQTT 协议使用发布/订阅模式传输消息。客户端可以订阅一个或多个主题（topics），而服务端负责将消息推送到客户端。
3. 可靠性：MQTT 协议具有高可靠性。它支持可靠的消息分发机制，以确保消息传递的稳定性和完整性。
4. 简单：MQTT 协议使用简单的协议格式，易于实现和调试。
5. 安全：MQTT 协议支持 SSL/TLS 加密通信，可以保护数据传输的安全性。
6. 多种服务质量等级（QoS）：MQTT 协议支持多种不同等级的服务质量，以满足不同应用场景的需求。

总之，由于 MQTT 协议具有轻量级、可靠性高、安全、灵活的特点，被广泛应用于物联网设备和应用程序之间的通信。

三、通信数据格式

阿里云支持 ICA 标准数据格式（Alink JSON）和透传/自定义

ICA 标准数据格式，也称为 Alink JSON，是阿里云物联网平台的一种标准数据格式，用于物联网设备与云端之间的通信。其中，ICA 代表 “智能连接智能的物)(Intelligence of Connection An Intelligent Thing)”。
Alink JSON 使用 JSON 格式进行编码和解码，包含通信协议（version）、设备信息（productKey、deviceName 和 deviceSecret）、请求 ID（id）、方法名（method）、参数列表（params）以及响应结果（code 和 message）等字段。
Alink JSON 具有以下特点：

1. 轻量级：使用基于文本的 JSON 格式，易于传输和解析。
2. 支持多种数据类型：包括数字、字符串、布尔值、数组等多种数据类型，支持不同的数据结构描述，如对象、数组、嵌套等。
3. 支持透传：通过设置 method 字段值为"thing.event.property.post"，Alkink JSON 可以支持将原始二进制数据透传到云端，满足较高的性能要求。
4. 支持批量操作：通过将多个请求合并成一个 JSON 数组，可以在一次通信中传递多个请求，提高数据传输效率。
5. 支持故障排除：通过响应结果中的 code 和 message 字段，可以快速确定请求是否成功，并在出现问题时快速排除故障。

综上，ICA 标准数据格式（Alink JSON）是阿里云物联网平台的一种标准数据格式，具有轻量级、灵活、可扩展性强等特点，适用于不同类型的设备和应用场景，并简化了设备接入阿里云 IoT 平台的难度，大大提高了通信效率和稳定性。

四、开发逻辑

简单梳理一下开发逻辑

1、技术框架介绍

目前使用 Python+ 阿里 linkkit+JSON+Excel+Qrcode，其中 JSON 用于保存设备数据，Excel 用于设备信息储存，Qrcode 将某个信息内容生成二维码，用于绑定设备；
阿里 Python-linkkit：https://help.aliyun.com/document_detail/261162.html

阿里云官网已提供 Demo 文件和指导方法；实际开发看业务需求。

Excel 表格内容：

设备软件编号	host_name	product_key	device_name	device_secret	data_type（0-自定义协议，1-标准协议）
ST100	eu-central-1	xxxx	xxxxx	xxxx	0
ST200	cn-shanghai	xxxx	xxxxx	xxxx	1

设备名称：输入项目名称匹配三元组信息
host_name：阿里云域名地址，主要看设备部署，eu-central-1 德国，cn-shanghai 上海；
product_key，device_name，device_secret 表示设备三元组信息，可以通过在阿里云上创建设备得到；
data_type：0-自定义协议，1-标准协议；

2、实际操作

一、启动程序，输入设备软件编号，从 device 表格中获取设备信息

from excelLogic import ExcelLogic
from connectMqtt import ConnectMqtt
import sys
import qrcode
def start():
    while True:
        print("---start----")
        print("请输入你测试项目")
        try:
            product=input()
        except KeyboardInterrupt:
            sys.exit()
        ex = ExcelLogic(product)  #获取Excel表格信息
        deviceinfo  = ex.get_triple()
        print(deviceinfo)
        if deviceinfo:
            data = deviceinfo[2]
            image = qrcode.make(data=data)
            image.show()  #显示二维码信息
            image.save("deviceCode.png")
            con = ConnectMqtt(ex,product,deviceinfo[0],deviceinfo[1],deviceinfo[2],deviceinfo[3],deviceinfo[4])
            con.start_worker_loop()#启动工作线程
        else:
            print("暂无查找到此项目信息")


if __name__ == '__main__':
    start()

2、输入 device.xlsx 表格存在的项目软件编号（如 ST100），点击回车，可以通过 APP 进行扫码订阅此设备，再点击关闭此二维码；

3、自定义协议：

一、启动项目为首次运行，此时将数据模板文件拷贝到项目目录下，并且命名；

def json_file_data(self):
     '''
     json 文件数据获取
     1、当无JSON文件时，将模板文件复制为项目JSON文件
     2、当存在JSON文件时，直接读取信息
     '''
     if os.path.exists(self.device_json_link ):
         with open(self.device_json_link,'r',encoding='utf8')as fp:
             json_data = json.load(fp)
     else:
         shutil.copy("template/CN.json","project_catalog/"+self.Project_name+".json")
         with open(self.device_json_link,'r',encoding='utf8')as fp:
             json_data = json.load(fp)
     return json_data

二、上报自定义数据：举例子使用 Modbus 协议 RS485 通信，30 个双字节为一个数据包，包的大小看实际业务；
修改上报数据：比如需要将 1001 段数据中 1002 地址，改成为 3；
操作步骤：打开指定项目的.json 文件，在 1001 表格中，找到 1002,数值列中，输入 3，点击保持，此时再指令框中，输入 1001 点击回车

指令：1 或者 1001，上报 1001-1030 段的数据包
指令：2 或者 1031，上报 1031-1060 段的数据包
指令：3 或者 2001，上报 2001-2030 段的数据包
指令：4 或者 2031，上报 2031-2160 段的数据包
指令：return ,退出当前设备，返回到设备选择功能
指令：exit，退出程序

查看云端下发数据，可以通过云端或者 APP 端修改设备参数，如下所示，云端下发 1011 地址改为 “1”

指令上报方法：

def user_update(self,userdata):
    '''
    user_update上报数据方法,自定义协议
    '''
    rc, mid = self.lk.publish_topic(self.lk.to_full_topic("user/update"),bytearray.fromhex(userdata))
    if rc == 0:
        print("publish topic success:%r, mid:%r" % (rc, mid))
    else:
        print("publish topic fail:%d" % rc)

指令下发解析：先订阅 user/get Topic（若不订阅，收到新建设备，会接收不到下发数据）

def user_get(self):
    """
    订阅user_get,用于获取下发的设备指令
    """    
    rc, mid = self.lk.subscribe_topic(self.lk.to_full_topic("user/get"))
    if rc == 0:
        print("publish topic success:%r, mid:%r" % (rc, mid))
    else:
        print("publish topic fail:%d" % rc)

def on_topic_message(self,topic, payload, qos, userdata):
    #logging.debug("阿里云发布数据:", str(payload))

    print("接收到阿里云数据:", payload.hex())
    self.parsing_data(payload.hex())  #数据解析方法
    self.reply_data(payload.hex())    #数据应答方法

数据 CRC 校验方法：

def calc_crc(self,string):
    '''
    calc_crc 用于数据CRC校验
    '''
    data = bytearray.fromhex(string)
    crc = 0xFFFF
    for pos in data:
        crc ^= pos
        for i in range(8):
            if (crc & 1) != 0:
                crc >>= 1
                crc ^= 0xA001
            else:
                crc >>= 1
    hex_crc = hex(((crc & 0xff) << 8) + (crc >> 8)) # 返回十六进制
    crc_0 = crc & 0xff
    crc_1 = crc >> 8
    str_crc_0 = '{:02x}'.format(crc_0).upper()
    str_crc_1 = '{:02x}'.format(crc_1).upper()
    return str_crc_0.lower(), str_crc_1.lower() # 返回两部分十六进制字符

4、物模型协议（ICA 标准数据格式）：

若是新增项目，需要在 project_catalog 文件夹下，新建一个 json 文档，命名要和新增项目名称一致，
物模型编号，可以登录阿里云，在项目界面物模型 TSL 中查看，可以通过导出模型文件。

上报自定义数据
指令（内置指令）：1，上报{"0":1}数据，表示物模型编号"0",参数设置为 1，如 ST840 项目，物模型编号 “0” 代表开关机，因此参数下发时，云端或者 APP 端会表现为开机
指令：2（内置指令），上报{“0":0}数据，表示物模型编号"0",参数设置为 1，如 ST840 项目，物模型编号 “0” 代表开关机，因此参数下发时，云端或者 APP 端会表现为关机
指令：josn，将项目.josn 文件全部内容，上报到云端
指令：符合 Python 字典类型的参数，会自动将此参数上报到云端，比如输入{"1":1}或者{"2":0}等等，符合此规则参数，都上报到云端，否则会提示不符合字典类型错误
指令：return ,退出当前设备，返回到设备选择
指令：exit，退出程序
如输入：json

如输入：自定义内容

查看云端下发数据，可以通过云端或者 APP 端修改设备参数，如下所示，物模型编号 “7” 值改为 175，ST840 项目表示将设定温度改为 17.5℃

指令逻辑

def input_thing_data(self):
       '''
       物模型类型，输入数据内容
       '''
       while True:
           try:
               msg = input()
           except KeyboardInterrupt:
               sys.exit()
           else:
               if msg == "exit":
                   sys.exit()
               elif msg == "return":
                   self.lk.disconnect() #断开连接
                   time.sleep(1)
                   break
               elif msg == "1": #开机
                   prop_data = {
                       0 : 1
                   }
                   self.lk.thing_post_property(prop_data)
               elif msg == "2":  #关机
                   prop_data = {
                       0 : 0
                   }
                   self.lk.thing_post_property(prop_data)
               elif msg == "json":  #通过json文档输入
                   prop_data = self.ex.get_json_data()
                   self.lk.thing_post_property(prop_data)
               else:  #手动输入字典类型
                   self.thing_update(msg)

上报物模型数据

def thing_update(self,userdata):
    '''
        thing_update 上报物模型数据的方法
        '''
    redata  = re.match(r"\{\"(.*)\"\:(.*)\}",userdata)#正则校验数据类型
    #print(redata)
    if redata is None:
        print("输入的值不符合字典类型")
        return  False
    else:
        try:
            data = ast.literal_eval(userdata)
            rc, mid = self.lk.thing_post_property(data)
        except Exception as e:
            print(e)

下发物模型数据

def on_thing_prop_changed(self, params, userdata): 
    print("-------接受到云端下发的数据------JSON")
    print("下发点位号:"+str(params.keys())+" 参数值:"+str(params.values()))
    print("------------------------------------")

5、成果

一、在物联网 APP 测试时间上，从原本的搭建工装环境一两天，测试二三小时，到现在配置几分钟，测试二三小时，并且配置一次，永久使用，个人已经在项目中使用，测试项目周期缩短；
二、设备故障报警条件，比较苛刻和复杂，有一些特殊的故障，需要等待几分钟或者几个小时业务逻辑判断，并且使用设备调节和触发，难度比较大；使用此工具，仅知道设备故障点位，即可模拟上报任何的故障；
三、测试工装搭建时，必须涉及到用电安全性，开发和测试并未考电工证，电路短接或者电路接不规范；使用此工具，大量减少设备电线使用和有效避免人员的用电安全；
四、后期物联网 APP 自动化测试，接口测试提供校验数据支撑；
五、对公司设备直接通信业务更熟悉，Python 语言运用更熟练，APP 测试时，更快更精致定位 APP 或者后台问题（已经借用此工具，找到 APP 下发控制设备指令异常问题）；

6、自我自答

一、物联网产品测试，离不开真实设备联调测试，并且模拟设备模拟不出真实设备控制业务逻辑？
答：1、APP 首次对接设备时，建议使用真实设备对接测试，因此需要搭建测试环境；模拟设备本身是一把双刃剑，迭代版本时，对接功能比较少或者需要回归测试时，使用模拟设备是会大量提升能效；
2、避免其他人员首次对接设备时，使用模拟设备对接测试，将本地设备配置表去除，通过使用编写后端 API 接口，返回支持设备信息，比如使用 Python+Flask+MySQL 框架编写，举例代码以下；

@app.route('/device/getdeviceinfo/<name>')
def deviceinfo(name):
    # 3.编写sql语句
    sql1 = 'select * from devicelist where product ="'+name+'";'               # 获取q1表内数据   
    # 4.执行指定sql语句并执行SQL语句之后受影响的行数
    print(sql1)
    # res =db.session.execute(sql1).fetchone()
    affect_rows = cursor.execute(sql1)       # 行数 5行
    # 5.获取结果
    res = cursor.fetchone()  #获取第一条数据
    # res = cursor.fetchall()  #获取全部数据
    if res:
        print("res",res)
        return res
    else:
        return {'product':None}

二、阿里云平台有提供类似模拟软件和工具？
答：1、阿里云平台，仅提供物模型协议（ICA 标准数据格式）设备模拟器，但是透传/自定义格式设备不支持

2、使用过 MQTT 模拟器，对于透传/自定义格式设备，并不友好；前期想使用此软件做测试，但是由于自定义协议需要数据整理；再加上 APP 下发数据，无响应机制，因此走上自我开发道路；

三、模拟设备工具开发，是否对前端接口自动化测试和 UI 自动化测试有相关的帮助？
答：物联网，离不开设备本身，前端（APP，Web 端）数据来源，都是设备本身上报提供，在接口测试和 UI 自动化测试时，界面显示数据可通过模拟设备上报数据校准核对；举例子，模拟设备上报室内温度为 25.0℃，接口测试和 UI 测试时，就可以拿 25.0℃作为断言判定；而真实设备上报室内温度会随时间变化，因此不太适合。

四、模拟设备开发，需要具备哪些知识技能？
答：首先具备一定的开发技能和梳理出可实现业务开发逻辑，遇到问题时，分析其中原因，寻找解决方案；
其次对公司的产品业务逻辑比较清晰，协议，数据构成，设备上报数据逻辑和方式，云端的协议机制等；
最后保持一份好奇心和永不放弃的恒心，总会见彩虹；开发过程，就是在解决问题过程；

五、模拟设备工具开发，往后如何优化？
答：从个人测试角度，往两个方面优化；1、若考虑多人本地使用，使用 Python-QT 框架，将工具桌面化，让设置参数简单化和提升交互体验；2、封装成方法类，提供给 UI 自动化和接口自动化，用于设备数据校验；

六、此工具开发完成后，想提供给别人使用？
答：安装使用 pyinstaller 指令，将程序打包成 exe 可执行文件，比如此项目；

最后，欢迎各位技术大佬，进行技术经验交流。