EEPW论坛

感谢EEPW推出的“换取手持示波器”活动，本次分享的主题是结合Zephyr与MQTT来远程控制LED并获取BUTTON状态。

硬件需求：Nordic公司的Nrf7002 dk。

这个开发板提供了包括Wi-Fi，BLE，Thread以及NFC等常用的无线协议，包含一颗nrf7002 Wi-Fi协处理器，主控芯片是nrf5340，这是一个Cortex-M33内核的安全处理器。

nRF7002-DK是用于nRF7002 Wi-Fi 6协同IC的开发套件，该开发套件采用nRF5340多协议片上系统 (SoC) 作为nRF7002的主处理器，在单一的电路板上包含了开发工作所需的一切，可让开发人员轻松开启基于nRF7002 的物联网项目。该 DK 包括 Arduino 连接器、两个可编程按钮、一个 Wi-Fi 双频段天线和一个低功耗蓝牙天线，以及电流测量引脚。

这款DK支持低功耗 Wi-Fi 应用开发，并实现了多项 Wi-Fi 6 功能，比如 OFDMA、波束成型和 TWT。nRF7002 Wi-Fi 6配套IC为另一个主机添加了低功耗Wi-Fi 6功能，提供无缝连接和基于Wi-Fi的定位（本地Wi-Fi集线器的SSID嗅探）功能。该IC设计用于搭配Nordic现有的nRF52^®和nRF53^®系列多协议片上系统 (SoC) 和nRF91^®系列蜂窝物联网系统级封装 (SiP) 使用。nRF7002 IC 还可与非nordic主机器件搭配使用。通过SPI或QSPI与主机通信，并带有额外的共存功能，可与其他协议如蓝牙、Thread或Zigbee无缝共存。

本次项目的目标分解为如下子任务：

• Sub1: 实现联网功能

• Sub2:实现MQTT功能

• Sub3:增加LED控制

• Sub4:增加BUTTON状态上传

• Sub5:实物展示

软件框架使用的是Nordic的nRF Connect SDK v2.4.0版本。nRF Connect SDK 是一个可扩展的统一软件开发套件，用于构建基于我们所有 nRF52、nRF53, nRF70 和 nRF91 系列无线设备的产品。它为开发人员提供了一个可扩展的框架，用于为内存受限的设备构建尺寸优化的软件，以及为更高级的设备和应用程序构建强大而复杂的软件。它集成了 Zephyr RTOS 和各种示例、应用程序协议、协议栈、库和硬件驱动程序。nRF Connect SDK内嵌Zephyr RTOS，并沿用了Zephyr project的编译系统。Zephyr Project是Linux基金会推出的一个Apache2.0开源项目，版权非常友好，适合用于商业项目开发。Zephyr Project是一个合作社区，其产品就是Zephyr，具体包括Zephyr RTOS，Zephyr组件以及Zephyr编译系统等。Zephyr很多地方都模拟了Linux，比如使用了DeviceTree和Kconfig。

Zephyr zbus基本概念介绍：

zbus：Zephyr消息总线(Zephyr message bus)）是一个轻量级、灵活和可扩展的消息传递机制，用于在不同的设备和应用程序之间进行通信。在3.3.0 release正式出现。它提供了一种简单的方式，使得线程之间可以相互发送和接收消息，从而实现数据共享和协作。zbus提供轻量级和灵活的消息总线，可以使线程之间进行简单的通信。它实现了消息传递和发布/订阅通信范例，允许线程通过共享内存同步或异步地进行通信。zbus的通信是基于通道的，其中线程使用消息进行发布和读取。下图为zephyr官方文档给出的zbus典型应用架构：

定义

通道（Channel）：通道是zbus中的基本单元，用于在不同线程之间传递消息。每个通道都有一个唯一的名称和一个与之相关联的消息类型。
观察者（Observer）：观察者订阅特定通道并接收其消息。观察者可以是异步订阅者或同步监听者。
消息（Message）：消息是在zbus中传递的数据单元。每个消息都与特定通道相关联，并且必须与该通道关联的消息类型匹配。
消息类型（Message Type）：消息类型是用于定义特定通道所使用的消息结构体类型。每个通道只有一个唯一的消息类型。
发布（Publishing）：发布是将消息发送到特定通道以供观察者接收的过程。
订阅（Subscription）：订阅是指观察者向特定通道注册以接收其发布的所有消息的过程，订阅是异步过程。
监听（Listening）：监听是指观察者等待并接收来自特定通道发布的所有消息的过程，监听是同步过程。
验证器（Validator）：验证器是用于验证消息是否有效并符合预期格式和内容规则的函数或回调函数。

Nordic工具具有强大的生态，官方网站提供了非常丰富的案例，Nordic DevZone提供了大量的实际问题案例供参考。因此本项目的起步是基于官方的例程：MQTT（Samples\Networking\samples MQTT:  https://developer.nordicsemi.com/nRF_Connect_SDK/doc/2.4.0/nrf/samples/net/mqtt/README.html）。

这个例程具有模块化结构，其中每个模块都有定义的责任范围。 模块之间的通信由 Zephyr 消息总线 (zbus) 使用通过通道传递的消息进行处理。 如果模块具有内部状态处理，则它是使用 Zephyr 状态机框架来实现的。 下图是本项目的整体架构，说明了示例中模块、通道和网络堆栈之间的关系：

程序流程图如下：

项目使用到了Zephyr的Zbus虚拟总线，定义了如下4个Channel（通道）

Name

Channel payload

Payload description

同时也定义了如下的模块，并且列出了通道的订阅情况。比如Sampler模块订阅了通道“Trigger”。Transport模块订阅了通道“Network”和“Payload”。

Module name

Observes channel

Subscriber / Listener

Description

源文件结构：

├─common
│      CMakeLists.txt
│      message_channel.c
│      message_channel.h
│
└─modules
    ├─error
    │      CMakeLists.txt
    │      error.c
    │      Kconfig.error
    │
    ├─led
    │      CMakeLists.txt
    │      Kconfig.led
    │      led.c
    │
    ├─network
    │      CMakeLists.txt
    │      Kconfig.network
    │      network_emulation.c
    │      network_lte.c
    │      network_wifi.c
    │
    ├─sampler
    │      CMakeLists.txt
    │      Kconfig.sampler
    │      sampler.c
    │
    ├─transport
    │  │  CMakeLists.txt
    │  │  Kconfig.transport
    │  │  transport.c
    │  │
    │  ├─client_id
    │  │      client_id.c
    │  │      client_id.h
    │  │      CMakeLists.txt
    │  │
    │  ├─credentials
    │  │      ca-cert.pem
    │  │
    │  └─credentials_provision
    │          CMakeLists.txt
    │          credentials_provision.c
    │
    └─trigger
            CMakeLists.txt
            Kconfig.trigger
            trigger.c

Sub1:实现联网（使用模块：network）

本项目的实现是基于Zephyr RTOS的，所以首先需要使用K_THREAD_DEFINE来初始化一个用于网络连接到线程。zephyr支持两种创建线程的方式，分别是静态创建和动态创建，静态创建使用K_THREAD_DEFINE宏，动态创建则调用k_thread_create()函数。

K_THREAD_DEFINE(network_task_id,
  CONFIG_MQTT_SAMPLE_NETWORK_THREAD_STACK_SIZE,
  network_task, NULL, NULL, NULL, 3, 0, 0);

之后在network_task函数中进行Wifi网络连接。从代码中可以看到接着调用connect（）函数进行网络连接。

static void network_task(void)
{
 net_mgmt_init_event_callback(&net_mgmt_callback, wifi_mgmt_event_handler, MGMT_EVENTS);
 net_mgmt_add_event_callback(&net_mgmt_callback);
 net_mgmt_init_event_callback(&net_mgmt_ipv4_callback, ipv4_mgmt_event_handler,
     NET_EVENT_IPV4_ADDR_ADD | NET_EVENT_IPV4_ADDR_DEL);
 net_mgmt_add_event_callback(&net_mgmt_ipv4_callback);

 /* Add temporary fix to prevent using Wi-Fi before WPA supplicant is ready. */
 k_sleep(K_SECONDS(1));

 connect();
}

在编译之前，需要配置好如下跟WIFI网络相关的参数，比如上面创建线程时用到的“CONFIG_MQTT_SAMPLE_NETWORK_THREAD_STACK_SIZE”就是在如下位置进行配置，否则编译会失败。

日志栏的输出显示已经成功连接到Wi-Fi网络，并且获取到了IPv4地址。

[00:00:00.531,158] <inf> fs_nvs: 6 Sectors of 4096 bytes
[00:00:00.531,188] <inf> fs_nvs: alloc wra: 0, fe8
[00:00:00.531,219] <inf> fs_nvs: data wra: 0, 0
*** Booting Zephyr OS build v3.3.99-ncs1 ***
[00:00:00.533,905] <inf> zbus: Inside zbus.c line 123, after memcpy
[00:00:00.533,996] <inf> sampler: Inside Samplec LIne 30
[00:00:00.534,088] <inf> zbus: Inside zbus.c line 123, after memcpy
[00:00:01.534,973] <wrn> wifi_credentials: Cannot retrieve WiFi credentials, no entry found for the provided SSID
[00:00:01.535,003] <inf> wifi_mgmt_ext: Adding statically configured WiFi network [TP-LINK_XXX] to internal list.
[00:00:07.756,713] <inf> network: Wi-Fi Connected, waiting for IP address
[00:00:07.914,825] <inf> network: IPv4 address acquired

Sub2:实现连接到MQTT Broker（使用模块：transport）

在wifi连接成功后以及成功获取IPV4地址后，会调用zbus_chan_pub(&NETWORK_CHAN, &status, K_SECONDS(1))将网络状态发布到通道“NETWORK_CHAN”上。根据架构图，transport模块订阅了这个通道，所以这个模块会读取该通道的payload，然后做进一步处理。

transport模块中首先需要定义相关的进程：

K_THREAD_DEFINE(transport_task_id, CONFIG_MQTT_SAMPLE_TRANSPORT_THREAD_STACK_SIZE, transport_task, NULL, NULL, NULL, 3, 0, 0);

MQTT模块还定义了在“断开”与“连接”两种状态下相关的回调函数：

/* Construct state table */
static const struct smf_state state[] = {
 [MQTT_DISCONNECTED] = SMF_CREATE_STATE(disconnected_entry, disconnected_run, NULL),
 [MQTT_CONNECTED] = SMF_CREATE_STATE(connected_entry, connected_run, connected_exit),
};

接着对MQTT相关的回调函数进行注册，然后调用Zephyr的MQTT组件的接口进行mqtt初始化：

static void transport_task(void)
{
 int err;
 const struct zbus_channel *chan;
 enum network_status status;
 struct payload payload;
 struct mqtt_helper_cfg cfg = {
  .cb = {
   .on_connack = on_mqtt_connack,
   .on_disconnect = on_mqtt_disconnect,
   .on_publish = on_mqtt_publish,
   .on_suback = on_mqtt_suback,
  },
 };

 /* Initialize and start application workqueue.
  * This workqueue can be used to offload tasks and/or as a timer when wanting to
  * schedule functionality using the 'k_work' API.
  */
 k_work_queue_init(&transport_queue);
 k_work_queue_start(&transport_queue, stack_area,
    K_THREAD_STACK_SIZEOF(stack_area),
    K_HIGHEST_APPLICATION_THREAD_PRIO,
    NULL);

 err = mqtt_helper_init(&cfg);
 if (err) {
  LOG_ERR("mqtt_helper_init, error: %d", err);
  SEND_FATAL_ERROR();
  return;
 }

 /* Set initial state */
 smf_set_initial(SMF_CTX(&s_obj), &state[MQTT_DISCONNECTED]);
...
}

从上面代码片段的smf_set_initial()可以看到，transport模块初始化的时候的状态首先是“断开”。“Transport”模块订阅了通道“NETWORK_CHAN”，所以每次网络状态有变化，Transport模块都会收到一个notification。

下面是来自transport.c模块的代码，transport模块通过调用zbus_sub_wait()接口来等待来自通道“NETWORK_CHAN”的通知。进而根据网络状态的变化来做相应的处理。

 while (!zbus_sub_wait(&transport, &chan, K_FOREVER)) {

  s_obj.chan = chan;

  if (&NETWORK_CHAN == chan) {

   err = zbus_chan_read(&NETWORK_CHAN, &status, K_SECONDS(1));
   if (err) {
    LOG_ERR("zbus_chan_read, error: %d", err);
    SEND_FATAL_ERROR();
    return;
   }

   s_obj.status = status;

   err = smf_run_state(SMF_CTX(&s_obj));
   if (err) {
    LOG_ERR("smf_run_state, error: %d", err);
    SEND_FATAL_ERROR();
    return;
   }
  }
...
}

上面的smf_run_state(SMF_CTX(&s_obj))，会继续调用之前注册的回调函数，在if语句中判断当前的网络状态是否为NETWORK_CONNECTED，以及通知是否来自通道“NETWORK_CHAN”。

因此之后会调用在“断开”状态下的入口函数来尝试去建立与MQTT Broker的TCP连接，使用的端口是1883。通过函数connect_work_fn连接到broker:

连接成功后，会调用“连接”状态下的入口函数来打印相关的输出，进而调用subscribe()来获取nrf7002dk所订阅的相关的主题。

/* Function executed when the module enters the connected state. */
static void connected_entry(void *o)
{
 LOG_INF("Connected to MQTT broker");
 LOG_INF("Hostname: %s", CONFIG_MQTT_SAMPLE_TRANSPORT_BROKER_HOSTNAME);
 LOG_INF("Client ID: %s", client_id);
 LOG_INF("Port: %d", CONFIG_MQTT_HELPER_PORT);
 LOG_INF("TLS: %s", IS_ENABLED(CONFIG_MQTT_LIB_TLS) ? "Yes" : "No");

 ARG_UNUSED(o);

 /* Cancel any ongoing connect work when we enter connected state */
 k_work_cancel_delayable(&connect_work);

 subscribe();
}

其中订阅的主题与发布的主题是在kconfig里面定义：

串口打印输出：

[00:00:15.780,853] <inf> transport: Connected to MQTT broker
[00:00:15.780,883] <inf> transport: Hostname: broker.hivemq.com
[00:00:15.780,914] <inf> transport: Client ID: hifriday
[00:00:15.780,944] <inf> transport: Port: 1883
[00:00:15.780,944] <inf> transport: TLS: No
[00:00:15.781,005] <inf> transport: Subscribing to: hifriday/my/subscribe/topic
[00:00:16.046,691] <inf> transport: Subscribed to topic hifriday/my/subscribe/topic

至此，wifi网络连接成功，mqtt客户端成功连接到Broker。接来下增加led控制，button状态判断。

Sub3:增加LED控制

针对LED与button，Nordic已经封装好了相关的操作函数，只需要在kconfig里面勾选相关的功能即可。

另外，MQTT连接到Broker之后，如果有mqtt事件发生，并且为MQTT_EVT_PUBLISH即本客户端订阅的主题被发布了，那么在相应的回调函数中，就可以根据payload的内容进行LED小灯的亮灭操作（通过strncmp（）函数来对比固定位数的两个字符串是否相同，如果匹配成功，对相应的LED小灯进行操作）。

本项目中，电脑上用MQTTX软件来模拟一个MQTT的client，这个client会发布主题：hifriday/my/subscribe/topic，发布的内容主要是LED控制相关的，比如：

• LED1ON: 点亮板载的LED1

• LED1OFF: 熄灭板载的LED1

• LED2ON: 点亮板载的LED2

• LED2OFF: 熄灭板载的LED2

Nrf7002dk也是作为一个MQTT的客户端，它会订阅主题：hifriday/my/subscribe/topic。MQTT模块中的on_mqtt_publish（）函数在mqtt客户端订阅的主题有更新时候被调用。

Sub4:增加BUTTON状态上传

对于按键的状态，本项目是在trigger.c模块中进行的处理。跟别的模块一样，首先需要创建一个线程，用于处理button按下的触发操作。注意如下的函数，在while()循环中，让进程进行休眠。等待有按键按下后，在进行相关主题的发布。

static void trigger_task(void)
{
#if CONFIG_DK_LIBRARY
 int err = dk_buttons_init(button_handler);
#endif /* CONFIG_DK_LIBRARY */

 while (true) {
  message_send();
  //k_sleep(K_SECONDS(CONFIG_MQTT_SAMPLE_TRIGGER_TIMEOUT_SECONDS));K_FOREVER
  k_sleep(K_FOREVER);
 }
}

K_THREAD_DEFINE(trigger_task_id, CONFIG_MQTT_SAMPLE_TRIGGER_THREAD_STACK_SIZE, trigger_task, NULL, NULL, NULL, 3, 0, 0);

本模块中，针对两个按键分别定义了两个全局变量，每当有按键按下后，会更新这两个全局变量的值。之所以定义两个全局变量，就是希望在按键按下后，nrf7002dk进行相关主题发布的时候，发布的内容会跟哪个按键被按下有关。比如button1被按下，我希望的发布内容是：button1 is pressed。以此类推。

uint8_t Button1Pressed = 0;
uint8_t Button2Pressed = 0;

#if CONFIG_DK_LIBRARY
static void button_handler(uint32_t button_states, uint32_t has_changed)
{
 if (has_changed & button_states) {
  LOG_INF("Inside trigger.c line 39, button_handler");
  if (button_states & DK_BTN1_MSK){
   Button1Pressed = 1;
   Button2Pressed = 0;
   LOG_INF("Inside trigger.c line 43, button 1 pressed");
  }else{
   Button1Pressed = 0;
   Button2Pressed = 1;
   LOG_INF("Inside trigger.c line 47, button 2 pressed");
  }
  message_send();
 }
}
#endif /* CONFIG_DK_LIBRARY */

在上述的message_send()函数中，会调用如下的接口把message“not_used”发布到通道“TRIGGER_CHAN”上。

zbus_chan_pub(&TRIGGER_CHAN, &not_used, K_SECONDS(1));

由于sampler模块订阅了通道“TRIGGER_CHAN”，因此每当收到来自该通道的Notifaction后，都会继续后续操作：

 while (!zbus_sub_wait(&sampler, &chan, K_FOREVER)) {
  if (&TRIGGER_CHAN == chan) {
   sample();
  }
 }

在按键按下后，期望发布的内容是通过宏定义的两个字符串：BTN1_PRESSED_STRING与BTN2_PRESSED_STRING。因此需要根据哪个按键被按下来制作不同的payload发布到通道“Payload”上。

#define FORMAT_STRING "Hello MQTT! Current uptime is: %d"
#define BTN1_PRESSED_STRING "Hello MQTT! Button 1 on DK Board is Pressed"
#define BTN2_PRESSED_STRING "Hello MQTT! Button 2 on DK Board is Pressed"

/* Register subscriber */
ZBUS_SUBSCRIBER_DEFINE(sampler, CONFIG_MQTT_SAMPLE_SAMPLER_MESSAGE_QUEUE_SIZE);

static void sample(void)
{
 struct payload payload = { 0 };
 uint32_t uptime = k_uptime_get_32();
 int err, len;

 /* The payload is user defined and can be sampled from any source.
  * Default case is to populate a string and send it on the payload channel.
  */
 LOG_INF("Inside Samplec LIne 30");

 if(Button1Pressed){
  len = snprintk(payload.string, sizeof(payload.string), BTN1_PRESSED_STRING);
 }else if(Button2Pressed){
  len = snprintk(payload.string, sizeof(payload.string), BTN2_PRESSED_STRING);
 }else{
  len = snprintk(payload.string, sizeof(payload.string), FORMAT_STRING, uptime);
 }

 err = zbus_chan_pub(&PAYLOAD_CHAN, &payload, K_SECONDS(1));
}

transport模块订阅了通道“Payload”，所以sampler模块发布相应的message到该通道后，transport会收到notification。进而调用mqtt相应的回调函数把button相关的字符串作为payload发布到主题：hifriday/my/publish/topic

  if (&PAYLOAD_CHAN == chan) {
   err = zbus_chan_read(&PAYLOAD_CHAN, &payload, K_SECONDS(1));
   s_obj.payload = payload;
   err = smf_run_state(SMF_CTX(&s_obj));
  }

进而调用publish函数接这个mqtt报文发送出去：

/* Function executed when the module is in the connected state. */
static void connected_run(void *o)
{
 struct s_object *user_object = o;

 if (user_object->chan != &PAYLOAD_CHAN) {
  return;
 }

 publish(&user_object->payload);
}

串口与MQTTX客户端：

Sub5:实物展示

LED1 ON:

LED2ON:

LED2OFF:

button 1 press:

button 2 press: