叶帆物联平台通信

概述

本章演示 YFAir8000 在多网络场景下通过 YFLink 协议上报设备遥测数据。支持 4G、以太网、WiFi 三种网络自动检测与切换，任一连通外网即可上云。

技术栈

运行环境：LuatOS（合宙 Air8000W 模组）
通信协议：MQTT + YFLink
认证方式：HMAC-SHA1（项目/产品/设备四元组）
网络支持：4G > 以太网 > WiFi 自动切换

系统架构

                    ┌─────────────────┐
                    │   main.lua       │
                    │  (主程序 + 网络管理)│
                    └────────┬────────┘
                             │
              ┌──────────────┼──────────────┐
              ▼              ▼              ▼
        ┌──────────┐  ┌──────────┐  ┌──────────┐
        │ 4G 网络   │  │ 以太网    │  │ WiFi STA │
        │ LWIP_GP  │  │CH390H    │  │ LWIP_STA │
        └────┬─────┘  └────┬─────┘  └────┬─────┘
             │              │              │
             └──────────────┼──────────────┘
                            │ 自动选最优
                            ▼
                    ┌─────────────────┐
                    │   yflink.lua     │
                    │  (YFLink MQTT)   │
                    └────────┬────────┘
                             │ MQTT
                             ▼
                    ┌─────────────────┐
                    │ 叶帆物联网平台    │
                    │ iot.yfios.net   │
                    └─────────────────┘

程序流程

Main()
  │
  ├─ 1. init_all_networks()
  │      ├─ init_4g()           → DNS 配置
  │      ├─ init_ethernet()     → SPI + CH390H 驱动
  │      ├─ init_wifi()         → wlan.connect()
  │      └─ 等待 60s → select_best_network()
  │
  ├─ 2. subscribe("IP_READY"/"IP_LOSE") → 网络动态监控
  │
  ├─ 3. connect_mqtt()
  │      ├─ YFLink.init()       → 四元组配置
  │      └─ YFLink.connect()    → HMAC-SHA1 认证 + 订阅主题
  │
  ├─ 4. upload_data()           → 首次上传
  │
  └─ 5. 主循环（30s）
         ├─ MQTT 断线 → 重连
         └─ MQTT 在线 → upload_data()

项目结构

07 - 数据上传叶帆物联平台/
├── main.lua        # 主程序（多网络管理 + 数据上传）
└── yflink.lua      # YFLink 协议库（MQTT 连接 + 属性上报）

核心代码详解

1. 三网自动切换

优先级：4G > 以太网 > WiFi，检测到任意网络获取 IP 后自动选择：

local network_priority = { socket.LWIP_GP, socket.LWIP_ETH, socket.LWIP_STA }

-- 选择最佳网络
local function select_best_network()
    for _, adapter in ipairs(network_priority) do
        if net_status[adapter] and net_status[adapter].ready then
            return adapter
        end
    end
    return nil
end

-- 订阅网络事件，动态切换
sys.subscribe("IP_READY", function(ip, adapter)
    net_status[adapter].ready = true
    check_and_switch_network()
end)

sys.subscribe("IP_LOSE", function(adapter)
    net_status[adapter].ready = false
    check_and_switch_network()
end)

2. 以太网 CH390H 初始化

spi.setup(1, nil, 0, 0, 8, 25600000)  -- SPI1
netdrv.setup(socket.LWIP_ETH, netdrv.CH390, {
    spi = 1, cs = 12, irq = 21
})
netdrv.dhcp(socket.LWIP_ETH, true)

3. YFLink MQTT 认证

-- ClientID:  项目ID-产品ID-设备ID
-- UserName:  项目ID&产品ID&设备ID
-- Password:  HMAC-SHA1(ClientID+UserName, DeviceKey) → 小写十六进制
local hmac = crypto.hmac_sha1(content, key)
local password = string.lower(hmac)

mqtt_client:auth(client_id, user_name, password)
mqtt_client:keepalive(60)
mqtt_client:autoreconn(true, 3000)  -- 自动重连

4. 属性上传格式

YFLink.post_property(properties, callback)

-- 实际发送的 JSON：
{
    "id": 1,
    "ver": "1.0.0",
    "timestamp": 1717000000000,
    "params": {
        "WrtmICCID": "89860...",
        "WrtmRSSI": 25,
        "YFLinkConnectNum": 5,
        "YFLinkUplinkFlow": 5120,
        "NetworkType": "4G"
    }
}

5. 网络切换 → MQTT 重连

-- 切换网络时发布事件
local function switch_network(new_adapter)
    socket.dft(new_adapter)  -- 设置默认网卡
    sys.publish("NETWORK_SWITCHED")
end

-- 独立任务监听切换事件
sys.taskInit(function()
    while true do
        sys.waitUntil("NETWORK_SWITCHED", 30000)
        connect_mqtt()
        upload_data()
    end
end)

全部代码

main.lua

main.lua
PROJECT = "YFAir8000W_MQTT"
VERSION = "1.0.0"

log.info("main", "项目:", PROJECT, "版本:", VERSION)

local YFLink = require "yflink"

local CONFIG = {
    YFLink = {
        MQTT_SERVER = "iot.yfios.net",
        MQTT_PORT = 1883,
        ProjectID = "YOUR_PROJECT_ID",
        ProductID = "YOUR_PRODUCT_ID",
        DeviceID = "YOUR_DEVICE_ID",
        DeviceKey = "YOUR_DEVICE_KEY"
    },
    Network = {
        Ethernet = {
            spi_id = 1, cs_pin = 12, irq_pin = 21,
            spi_config = { cpha = 0, cpol = 0, dataw = 8, bandrate = 25600000 }
        },
        WiFi = { ssid = "YOUR_WIFI_SSID", password = "YOUR_WIFI_PASSWORD" }
    },
    UPLOAD_INTERVAL = 30000,
    NETWORK_CHECK_INTERVAL = 5000
}

local system_state = {
    mqtt_connected = false, upload_count = 0,
    current_network = nil, current_network_name = "无",
    network_switching = false
}

local net_status = {
    [socket.LWIP_GP]  = { ready = false, ip = nil },
    [socket.LWIP_ETH] = { ready = false, ip = nil },
    [socket.LWIP_STA] = { ready = false, ip = nil }
}

local network_priority = { socket.LWIP_GP, socket.LWIP_ETH, socket.LWIP_STA }
local network_names = {
    [socket.LWIP_GP] = "4G", [socket.LWIP_ETH] = "以太网", [socket.LWIP_STA] = "WiFi"
}

-- 初始化4G
local function init_4g()
    log.info("网络", "初始化4G...")
    socket.setDNS(socket.LWIP_GP, 1, "223.5.5.5")
    socket.setDNS(socket.LWIP_GP, 2, "114.114.114.114")
end

-- 初始化以太网
local function init_ethernet()
    log.info("网络", "初始化以太网 CH390H...")
    local eth = CONFIG.Network.Ethernet
    spi.setup(eth.spi_id, nil, eth.spi_config.cpha, eth.spi_config.cpol,
              eth.spi_config.dataw, eth.spi_config.bandrate)
    local result = netdrv.setup(socket.LWIP_ETH, netdrv.CH390, {
        spi = eth.spi_id, cs = eth.cs_pin, irq = eth.irq_pin
    })
    if result then
        netdrv.dhcp(socket.LWIP_ETH, true)
        socket.setDNS(socket.LWIP_ETH, 1, "223.5.5.5")
        socket.setDNS(socket.LWIP_ETH, 2, "114.114.114.114")
        log.info("网络", "以太网驱动初始化成功")
    else
        log.error("网络", "以太网驱动初始化失败")
    end
end

-- 初始化WiFi
local function init_wifi()
    log.info("网络", "初始化WiFi...")
    local wifi = CONFIG.Network.WiFi
    if wifi.ssid and wifi.ssid ~= "" then
        wlan.init()
        wlan.connect(wifi.ssid, wifi.password)
        log.info("网络", "WiFi连接中...")
    end
end

-- 选择最佳网络
local function select_best_network()
    for _, adapter in ipairs(network_priority) do
        if net_status[adapter] and net_status[adapter].ready then
            return adapter
        end
    end
    return nil
end

-- 切换网络
local function switch_network(new_adapter)
    if system_state.current_network == new_adapter then return false end
    local new_name = network_names[new_adapter] or "无"
    log.info("网络", "切换: " .. system_state.current_network_name .. " -> " .. new_name)
    socket.dft(new_adapter)
    system_state.current_network = new_adapter
    system_state.current_network_name = new_name
    if system_state.mqtt_connected then
        system_state.mqtt_connected = false
        sys.publish("NETWORK_SWITCHED")
    end
    return true
end

-- 检查并切换网络
local function check_and_switch_network()
    local best = select_best_network()
    if best and best ~= system_state.current_network then
        switch_network(best)
    elseif not best and system_state.current_network then
        log.warn("网络", "所有网络不可用")
        system_state.current_network = nil
        system_state.current_network_name = "无"
        system_state.mqtt_connected = false
    end
end

-- 网络事件处理
local function on_ip_ready(adapter, ip)
    log.info("网络", network_names[adapter] .. " 获取IP: " .. ip)
    net_status[adapter].ready = true
    net_status[adapter].ip = ip
    check_and_switch_network()
end

local function on_ip_lose(adapter)
    log.warn("网络", network_names[adapter] .. " 断开")
    net_status[adapter].ready = false
    net_status[adapter].ip = nil
    check_and_switch_network()
end

-- 初始化所有网络
local function init_all_networks()
    log.info("网络", "初始化所有网络...")
    sys.subscribe("IP_READY", function(ip, adapter) on_ip_ready(adapter, ip) end)
    sys.subscribe("IP_LOSE", function(adapter) on_ip_lose(adapter) end)

    init_4g()
    sys.wait(500)
    init_ethernet()
    sys.wait(500)
    init_wifi()

    local timeout = 0
    while timeout < 60 do
        if select_best_network() then break end
        sys.wait(1000)
        timeout = timeout + 1
    end

    local best = select_best_network()
    if best then
        switch_network(best)
        log.info("网络", "初始化完成，当前网络:", system_state.current_network_name)
        return true
    end
    log.error("网络", "所有网络初始化失败")
    return false
end

-- MQTT连接
local function connect_mqtt()
    if system_state.mqtt_connected then return true end
    if not system_state.current_network then return false end

    log.info("MQTT", "连接服务器:", CONFIG.YFLink.MQTT_SERVER, "网络:", system_state.current_network_name)
    YFLink.init({
        project_id = CONFIG.YFLink.ProjectID, product_id = CONFIG.YFLink.ProductID,
        device_id = CONFIG.YFLink.DeviceID, device_key = CONFIG.YFLink.DeviceKey
    })
    YFLink.connect(CONFIG.YFLink.MQTT_SERVER, CONFIG.YFLink.MQTT_PORT, function(success)
        system_state.mqtt_connected = success
        if success then
            log.info("MQTT", "================ 连接成功 ================")
        else
            log.error("MQTT", "连接失败")
        end
    end, system_state.current_network)

    sys.waitUntil("YFLINK_MQTT_CONACK", 15000)
    return system_state.mqtt_connected
end

-- 数据上传
local function get_device_data()
    local traffic = YFLink.get_traffic()
    local rssi = (system_state.current_network == socket.LWIP_GP) and (mobile.rssi() or 0) or 0
    return {
        WrtmICCID = mobile.iccid() or "", WrtmIMSI = mobile.imsi() or "",
        WrtmRSSI = rssi, YFLinkConnectNum = system_state.upload_count,
        YFLinkDownlinkFlow = traffic.downlink, YFLinkUplinkFlow = traffic.uplink,
        NetworkType = system_state.current_network_name
    }
end

local function upload_data()
    if not system_state.mqtt_connected then return false end
    local data = get_device_data()
    YFLink.post_property(data, function(ok, msg_id)
        if ok then
            system_state.upload_count = system_state.upload_count + 1
            log.info("上传", "成功，计数:", system_state.upload_count)
        else
            log.error("上传", "失败")
        end
    end)
    return true
end

-- 主任务
sys.taskInit(function()
    log.info("系统", "启动...")
    if not init_all_networks() then return end
    connect_mqtt()
    if system_state.mqtt_connected then upload_data() end
    while true do
        sys.wait(CONFIG.UPLOAD_INTERVAL)
        if not system_state.current_network then
            log.warn("系统", "无网络，等待...")
        elseif not system_state.mqtt_connected then
            connect_mqtt()
        else
            upload_data()
        end
    end
end)

-- 网络切换处理任务
sys.taskInit(function()
    while true do
        sys.waitUntil("NETWORK_SWITCHED", 30000)
        if system_state.current_network and not system_state.mqtt_connected then
            log.info("网络", "网络已切换，重新连接MQTT...")
            sys.wait(1000)
            connect_mqtt()
            if system_state.mqtt_connected then upload_data() end
        end
    end
end)

sys.run()

yflink.lua

yflink.lua
local YFLink = {}

local device_info = {}
local mqtt_client = nil
local message_id = 0
local traffic = { uplink = 0, downlink = 0 }

local function gen_message_id()
    message_id = message_id + 1
    if message_id > 2147483647 then message_id = 1 end
    return message_id
end

local function get_timestamp()
    return os.time() * 1000
end

local function gen_client_id()
    return device_info.project_id .. "-" .. device_info.product_id .. "-" .. device_info.device_id
end

local function gen_user_name()
    return device_info.project_id .. "&" .. device_info.product_id .. "&" .. device_info.device_id
end

local function gen_password()
    local content = gen_client_id() .. gen_user_name()
    local hmac = crypto.hmac_sha1(content, device_info.device_key)
    return string.lower(hmac)
end

local function build_topic(suffix)
    return string.format("%s/%s/%s/%s",
        device_info.project_id, device_info.product_id,
        device_info.device_id, suffix)
end

local function publish(topic, payload, qos)
    if mqtt_client and mqtt_client:ready() then
        mqtt_client:publish(topic, payload, qos or 1)
        traffic.uplink = traffic.uplink + #payload
        return true
    end
    return false
end

function YFLink.init(config)
    device_info.project_id = config.project_id
    device_info.product_id = config.product_id
    device_info.device_id = config.device_id
    device_info.device_key = config.device_key
    log.info("YFLink", "初始化完成")
end

function YFLink.connect(server, port, callback, adapter)
    local client_id = gen_client_id()
    local user_name = gen_user_name()
    local password = gen_password()
    log.info("YFLink", "连接MQTT:", server, port, "网络:", adapter or "默认")

    mqtt_client = mqtt.create(adapter, server, port or 1883)
    if not mqtt_client then
        if callback then callback(false) end
        return
    end

    mqtt_client:auth(client_id, user_name, password)
    mqtt_client:keepalive(60)
    mqtt_client:autoreconn(true, 3000)

    mqtt_client:on(function(mqttc, event, data, payload)
        if event == "conack" then
            mqttc:subscribe(build_topic("property/set"), 1)
            mqttc:subscribe(build_topic("server/send"), 1)
            sys.publish("YFLINK_MQTT_CONACK")
            if callback then callback(true) end
        elseif event == "disconnect" then
            if callback then callback(false) end
        elseif event == "recv" and payload then
            traffic.downlink = traffic.downlink + #payload
        end
    end)

    mqtt_client:connect()
end

function YFLink.post_property(properties, callback)
    local id = gen_message_id()
    local topic = build_topic("property/post")
    local payload = json.encode({
        id = id, ver = "1.0.0", timestamp = get_timestamp(), params = properties
    })
    if publish(topic, payload) then
        if callback then callback(true, id) end
        return true
    else
        if callback then callback(false, id) end
        return false
    end
end

function YFLink.is_connected()
    return mqtt_client and mqtt_client:ready()
end

function YFLink.get_traffic()
    return traffic
end

return YFLink

上传数据字段

字段	说明	来源
`WrtmICCID`	SIM 卡 ICCID	`mobile.iccid()`
`WrtmIMSI`	SIM 卡 IMSI	`mobile.imsi()`
`WrtmRSSI`	信号强度（仅 4G）	`mobile.rssi()`
`YFLinkConnectNum`	上传计数	本地累加
`YFLinkDownlinkFlow`	下行流量（字节）	MQTT recv 累计
`YFLinkUplinkFlow`	上行流量（字节）	MQTT publish 累计
`NetworkType`	当前网络	"4G" / "以太网" / "WiFi"

关键技术点

技术点	API	说明
MQTT 客户端	`mqtt.create(adapter, server, port)`	绑定指定网络适配器
YFLink 认证	`crypto.hmac_sha1(content, key)`	HMAC-SHA1 签名
属性上传	`YFLink.post_property(data, cb)`	JSON 序列化 → MQTT publish
网络事件	`sys.subscribe("IP_READY"/"IP_LOSE")`	实时感知网络变化
网卡选择	`socket.dft(adapter)`	设置默认网卡
以太网	`netdrv.setup(LWIP_ETH, netdrv.CH390, ...)`	CH390H SPI 驱动
自动重连	`mqtt_client:autoreconn(true, 3000)`	3 秒自动重连
流量统计	`YFLink.get_traffic()`	上下行字节累计

概述​

系统架构​

程序流程​

项目结构​

核心代码详解​

1. 三网自动切换​

2. 以太网 CH390H 初始化​

3. YFLink MQTT 认证​

4. 属性上传格式​

5. 网络切换 → MQTT 重连​

全部代码​

main.lua​

yflink.lua​

上传数据字段​

关键技术点​

概述

系统架构

程序流程

项目结构

核心代码详解

1. 三网自动切换

2. 以太网 CH390H 初始化

3. YFLink MQTT 认证

4. 属性上传格式

5. 网络切换 → MQTT 重连

全部代码

main.lua

yflink.lua

上传数据字段

关键技术点