物联网健康马桶开源分享

描述

介绍

由于文明的发展以及工业和汽车的不洁排放量增加，大气状况每年都在恶化。尽管空气是生命不可缺少的资源，但许多人对空气污染的严重程度漠不关心，或者直到最近才意识到这个问题。在水、土壤、热、噪声等各类污染物中，空气污染是最危险、最严重的，它会引起气候变化和危及生命的疾病。根据世界卫生组织 (WHO) 的数据，现在 90% 的人口呼吸着被污染的空气，空气污染是每年 700 万人死亡的原因。污染对健康的影响非常严重，会导致中风、肺癌和心脏病。此外，空气污染物对人类和地球生态系统有负面影响。

根据美国环境保护署 (EPA) 的数据，室内空气的污染程度是室外空气的 100 倍。大多数现代人将 80% 到 90% 的时间花在室内；在这次 COVID-19 大流行中，发病率甚至更高，全世界所有人的情况都是一样的。因此，室内空气比室外空气对人体健康的直接影响更大。此外，与大气污染相比，室内污染物传播到肺部的可能性要高出约 1000 倍，从而导致病态建筑综合症、多重化学敏感性和头晕等疾病。

厕所是公共设施之一，是人们经常使用的公共设施之一，位于室内。因此，保持厕所良好的空气质量对于保持厕所卫生和卫生至关重要。这与威尔克提到的说法一致，“为了创造一个更健康、更安全的环境，第一步是在洗手间。” 空气质量差的厕所可能是微生物、空气传播细菌的理想场所，最近还添加了 COVID-19 来传播。

我的健康厕所项目和动机

通风不足、人流量大、公厕管理不善是公厕室内空气污染的主要来源。为减少接触空气污染，可以采取新措施，包括开发空气质量测量设备、持续监测空气质量数据，以及根据数据采取必要措施。

物联网（IoT）和云计算揭示了各个领域实时监控的新能力。由于这些威廉希尔官方网站 采用无线传感器网络来自动传输、处理、分析和可视化数据，因此融合这些新威廉希尔官方网站 也可以为改善室内空气质量提供巨大优势。

必须衡量用户使用厕所的满意度，才能对厕所进行适当的管理。通常，马桶会以固定的时间间隔进行清洁，但有时即使还没有清洁计划，马桶也需要清洁，因为这会给用户带来不适。分析空气质量数据，可以在厕所需要清洁时检测到这种情况。

通风可以通过降低室内空气污染物和 COVID-19 等病毒的浓度来提高舒适度并改善室内空气质量 (IAQ)。但通风涉及电力消耗，需要高效智能运行以节省能源。

该项目意义重大，因为它通过通知厕所服务员或所有者何时需要清洁来帮助保持公共/私人厕所的清洁和良好的空气质量。它还根据空气质量数据智能地操作排气扇，以保持马桶内的舒适环境，从而将能耗降至最低。基于云的机器学习用于分析数据并做出决策。

健康厕所项目特点

• 将重要的空气质量数据（温度、湿度、氨气、二氧化硫、一氧化碳、二氧化氮、甲烷和噪音水平）安全地发布到 AWS。
• 分析空气质量数据并根据参数对排气扇进行无线控制，以保持厕所内舒适健康的环境，并且非常有效地消耗最少的能量。
• 使用机器学习，进一步分析数据以持续监控排气扇是否能够维持健康的环境。如果污染程度如此之高，以至于排气扇和通风系统无法保持舒适和卫生，则会将手动清洁请求发送到清洁人员或当局的电子邮件中。
• 将机器学习应用于噪音水平和其他参数，检测到漏水并向业主发送通知。
• 根据内部空气的状况，系统会产生颜色信号，以便用户知道马桶是否可以安全使用。绿灯意味着厕所环境使用起来安全舒适。黄灯表示空气不够好，用户应等到排气扇将其恢复到更好的状态。红灯表示厕所在有人清理之前无法使用。

健康马桶如何运作？

该项目的完整框图如下所示。所有的传感器都是

通过端口 A 和端口 B 连接到 Core2 EduKit。Core2 EduKit 读取传感器并使用 MQTT 将所有传感器的值发送到 AWS IoT Core。IoT 规则将数据转发到 IoT Analytics。IoT Analytics 预处理数据，将数据存储到 AWS S3，并根据数据准备数据集。AWS SageMaker 使用该数据集并构建机器学习模型，以根据 Core2 发布的未来数据确定厕所状态（是否干净或需要清洁）。一个单独的规则调用 Lambda 函数来使用 ML 模型读取新空气质量数据的预测。然后，该规则会根据预测结果更新设备影子，并且设备会收到有关当前状态的通知。

如果需要清洁马桶或有任何漏水，则使用另一个 IoT 规则使用 AWS SNS 向用户发送电子邮件/短信通知。Lambda 函数用于从原始 JSON 数据格式化电子邮件。

按照相同的程序，从噪音数据中检测到漏水，并通过电子邮件/短信通知业主。

专用的 IoT 规则用于根据空气质量数据确定排气扇状态，IoT Core 将数据发送到排气扇的单独主题集。我们对项目现在如何运作有了一个基本的了解，

让我们成功吧

连接硬件

主要硬件是 Core2 EduKit 和传感器。我使用 DHT11 传感器收集温度和湿度，使用 MICS6814 传感器检测一氧化碳、二氧化氮、氨和甲烷。该传感器通过三个独立的interwetten与威廉的赔率体系 通道提供数据。为了测量二氧化硫，我使用了 2SH12 传感器，它还提供模拟数据。因此，为了将这两个传感器连接到 Core2 套件，我使用了 4 通道 16 位 I2C ADC (ADS1115) 模块。我将此模块连接到 Core2 Kit 的端口 A。此端口支持 I2C

通信并且与 Grove 兼容。因此，我使用 Grove 电缆将传感器模块连接到 Core2。

DHT11 传感器提供数字数据。因此，我通过 Grove 电缆将此传感器连接到端口 B 的 GPIO 26 引脚。如果您使用 Grove DHT11 模块，您需要将信号线与 NC 线交替使用，以使其连接到 GPIO 26。

要将 MISC6814 和 2SH12 与 ADS1115 连接，我使用了一块穿孔板并将三个公对母排针焊接到穿孔板上。连接

将此 PCB 传感器板连接到 Core2 EduKit 我将 Grove 电缆焊接到 PCB 上，以维持 Grove I2C 端口的 I2C 接线顺序。最后，我把 DHT11

传感器连接到端口 B，I2C 传感器连接到 Core2 套件的端口 A，如下图所示。如果您注意到 PCB，您会发现 PCB 上焊接了三个电阻器。

MICS6814 传感器需要这些电阻器，您可以在代码部分所附的MICS6814.odt文件中找到传感器的详细信息和电阻器所需的值计算。有关详细连接，请参见下面的示意图。

通风/排气扇的连接

为了无线控制排气扇，我使用了 Node MCU 和继电器板。节点 MCU 将通过 MQTT 主题从 AWS IoT Core 接收风扇状态，并根据状态打开或关闭继电器。Realy 用于通过来自 Node MCU 的低压信号来控制大功率排气扇。下面的 Fritzing 草图显示了连接。

我为什么选择那些特定的传感器？

我选择了一些重要的传感器来测量厕所内的空气质量。对马桶来说重要的参数是氨、二氧化硫、二氧化碳、甲烷、温度、湿度等。

二氧化硫 (SO2)：二氧化硫是一种无色、具有强烈气味的活性空气污染物。这种气体可能对人类健康、动物健康和植物生命构成威胁。二氧化硫会刺激眼睛、鼻子、喉咙和肺部的皮肤和粘膜。高浓度的 SO2 会引起呼吸系统的炎症和刺激，尤其是在剧烈的体力活动期间。由此产生的症状可能包括深呼吸时的疼痛、咳嗽、喉咙发炎和呼吸困难。高浓度的 SO2 会影响肺功能，加重哮喘发作，并加重敏感人群现有的心脏病。这种气体还可以与空气中的其他化学物质发生反应，变成可以进入肺部并造成类似健康影响的小颗粒。超过1.0 ppm被认为是不健康的。

氨 (NH3)：氨是一种无色气体，具有强烈、刺鼻的刺激性气味。它通常用于水溶液、肥料、制冷剂、纺织品等。氨会在吸入时影响您。接触会严重刺激和灼伤皮肤和眼睛，并可能对眼睛造成伤害。吸入氨水会刺激鼻子、喉咙和肺部。反复接触可能会导致哮喘样过敏并导致肺损伤。超过25 ppm会产生上述症状。

一氧化碳（CO）：一氧化碳 (CO) 是一种无色无味的气体。它存在于加热器、壁炉、汽车消音器、空间加热器、木炭烤架、汽车发动机等产生的燃烧（废气）烟雾中。每个人全天都接触到少量的一氧化碳。然而，吸入过多会导致一氧化碳中毒。当燃烧烟雾被困在通风不良或封闭的空间（如车库）时，一氧化碳可能会增加到危险水平。吸入这些烟雾会导致一氧化碳在您的血液中积聚，从而导致严重的组织损伤。一氧化碳中毒非常严重，可能危及生命。如果您吸入大量一氧化碳，您的身体将开始用一氧化碳替换血液中的氧气。当这种情况发生时，您可能会失去知觉。在这些情况下可能会发生死亡。一旦 CO 水平增加到百万分之 70 (ppm)及以上，症状变得更加明显。这些症状可能包括恶心、头晕和无意识。

CO2（二氧化碳）是呼吸的副产品，其生物效应与 CO（一氧化碳）非常不同，CO（一氧化碳）是碳氢化合物燃烧的副产品，例如燃气灶和燃气或燃油锅炉。一氧化碳（但不是二氧化碳）会在通风不良的家中迅速积聚，而且是致命的。这就是为什么建议对室内条件进行 CO（一氧化碳）监测的原因。

二氧化氮 (NO2)：NO2 是最常见和研究最充分的污染物之一。接触二氧化氮——即使是短期接触的小幅增加——也会加剧呼吸系统问题，尤其是哮喘，尤其是儿童。长期接触二氧化氮会导致“心血管影响、糖尿病、较差的出生结果、过早死亡和癌症”。研究已将持续的二氧化氮暴露与认知能力下降联系起来，尤其是在儿童中。

为 Core2 AWS EduKit 开发代码

完成硬件连接后，下一步是为 Core2 AWS EduKit 开发固件。为此，请按顺序执行所有步骤。

第 1 步：完成Cloud Connected Blinky示例

在进一步继续之前，您必须从官方链接完成Cloud Connected Blinky示例。该教程解释得很好，如果您有嵌入式系统和 AWS 云的基本知识，您应该不会遇到任何困难。此处使用的所有步骤和技能将为在本教程中取得成功奠定基础。具体来说，您将：

• 在您的主机上安装和配置AWS CLI以远程管理您的 AWS 服务并使用提供的帮助程序脚本。
• 使用板载安全元件上预先配置的安全证书在 AWS IoT Core 中注册“事物” 。
• 配置您的参考硬件以连接到 Wi-Fi，连接到您账户的AWS IoT 终端节点，并将 MQTT 消息发送到 AWS IoT Core。
• 在指定的参考硬件上接收来自云端的 MQTT 消息以触发 LED 闪烁。

第 2 步：完成智能恒温器示例

下一步是打开、编译和测试官方智能恒温器固件。我们将使用此代码作为我们代码的模板。因此，在开始修改代码之前，我们希望通过运行 Core2 套件中的固件来确保一切正常。在进行下一步之前，这也是一个解释得很好且易于理解的教程。

在该教程结束时，您将了解：

• 如何从 Core2 for AWS IoT EduKit 设备获取温度和声级。
• MQTT 的工作原理以及如何使用 MQTT 将温度和声音测量值从设备发布到 AWS IoT Core。
• 什么是设备影子以及如何将测量值报告给设备影子？
• 如何使用 AWS IoT Core 规则引擎执行消息转换。
• 如何构建响应输入和检测复杂事件的无服务器应用程序。
• 如何通过设备影子向您的设备发送命令。

第 3 步：为 Core2 AWS IoT EduKit 开发固件

正如我已经说过的，我们将使用智能恒温器固件作为开发我们自己的固件的模板。因此，我假设您已经成功编译和测试了该固件。您可以像示例程序一样直接从 GitHub 下载和使用我修改后的固件（链接附在代码部分），也可以按照以下步骤自行自定义智能恒温器示例。

由于我们将使用一些外部传感器，我们需要编写代码来连接这些传感器。我为 Core2 IoT EduKit 开发了（实际上是修改了现有的 Arduino 库）两个库。一个用于 DHT11 温度和湿度传感器，另一个用于 ADS1115 I2C 模数转换器模块。按照接下来的步骤了解该过程。

步骤 3.1：创建库

为了创建一个库，我们需要创建两个文件。一个是C头文件，我们需要将它添加到主目录的include子目录中。另一个是C源文件需要添加到主目录的根目录。所以，让我们先创建一个头文件（.h）文件。

步骤 3.1.1：转到资源管理器->主->包含并单击鼠标右键并选择新建文件

输入带有 ah 扩展名的名称，然后从键盘按 Enter。例如dht11.h.

步骤 3.1.2：单击文件将其打开并在那里写入或粘贴您的代码。

例如，我为我的dht11.h头文件添加了以下代码。

#pragma once
#include 
#include 
#ifdef __cplusplus
extern "C" {
#endif
/**
* Sensor type
*/
typedef enum
{
DHT_TYPE_DHT11 = 0,   //!< DHT11
DHT_TYPE_AM2301,      //!< AM2301 (DHT21, DHT22, AM2302, AM2321)
DHT_TYPE_SI7021 //!< Itead Si7021
} dht_sensor_type_t;

esp_err_t dht_read_data(dht_sensor_type_t sensor_type, gpio_num_t pin,
int16_t *humidity, int16_t *temperature);

esp_err_t dht_read_float_data(dht_sensor_type_t sensor_type, gpio_num_t pin,
float *humidity, float *temperature);

#ifdef __cplusplus
}
#endif

步骤 3.1.3：按Ctrl+S保存文件。我们的头文件创建完成。现在我们将为库创建 C 源文件。

步骤 3.1.4:转到explorer -> main并单击鼠标右键并选择New File

输入一个带有 ac 扩展名的名称，然后从键盘上按回车键。例如dht11.c.

步骤 3.1.5：单击文件将其打开，然后在编辑器中编写或粘贴代码。

例如，我为我的dht11.c文件使用了以下代码行

#include 
#include 
#include 
#include 
#include 
#include "dht.h"
// DHT timer precision in microseconds
#define DHT_TIMER_INTERVAL 2
#define DHT_DATA_BITS 40
#define DHT_DATA_BYTES (DHT_DATA_BITS / 8)
/*
*  Note:
*  A suitable pull-up resistor should be connected to the selected GPIO line
*
*  __           ______          _______                              ___________________________
*    \    A    /      \   C    /       \   DHT duration_data_low    /                           \
*     \_______/   B    \______/    D    \__________________________/   DHT duration_data_high    \__
*
*
*  Initializing communications with the DHT requires four 'phases' as follows:
*
*  Phase A - MCU pulls signal low for at least 18000 us
*  Phase B - MCU allows signal to float back up and waits 20-40us for DHT to pull it low
*  Phase C - DHT pulls signal low for ~80us
*  Phase D - DHT lets signal float back up for ~80us
*
*  After this, the DHT transmits its first bit by holding the signal low for 50us
*  and then letting it float back high for a period of time that depends on the data bit.
*  duration_data_high is shorter than 50us for a logic '0' and longer than 50us for logic '1'.
*
*  There are a total of 40 data bits transmitted sequentially. These bits are read into a byte array
*  of length 5.  The first and third bytes are humidity (%) and temperature (C), respectively.  Bytes 2 and 4
*  are zero-filled and the fifth is a checksum such that:
*
*  byte_5 == (byte_1 + byte_2 + byte_3 + byte_4) & 0xFF
*
*/
static const char *TAG = "DHT";
static portMUX_TYPE mux = portMUX_INITIALIZER_UNLOCKED;
#define PORT_ENTER_CRITICAL() portENTER_CRITICAL(&mux)
#define PORT_EXIT_CRITICAL() portEXIT_CRITICAL(&mux)
#define CHECK_ARG(VAL) do { if (!(VAL)) return ESP_ERR_INVALID_ARG; } while (0)
#define CHECK_LOGE(x, msg, ...) do { \
esp_err_t __; \
if ((__ = x) != ESP_OK) { \
PORT_EXIT_CRITICAL(); \
ESP_LOGE(TAG, msg, ## __VA_ARGS__); \
return __; \
} \
} while (0)
/**
* Wait specified time for pin to go to a specified state.
* If timeout is reached and pin doesn't go to a requested state
* false is returned.
* The elapsed time is returned in pointer 'duration' if it is not NULL.
*/
static esp_err_t dht_await_pin_state(gpio_num_t pin, uint32_t timeout,
int expected_pin_state, uint32_t *duration)
{
/* XXX dht_await_pin_state() should save pin direction and restore
* the direction before return. however, the SDK does not provide
* gpio_get_direction().
*/
gpio_set_direction(pin, GPIO_MODE_INPUT);
// Enabling pull-up is required if the sensor has no physical pull-up resistor
gpio_set_pull_mode(pin, GPIO_PULLUP_ONLY);
for (uint32_t i = 0; i < timeout; i += DHT_TIMER_INTERVAL)
{
// need to wait at least a single interval to prevent reading a jitter
ets_delay_us(DHT_TIMER_INTERVAL);
if (gpio_get_level(pin) == expected_pin_state)
{
if (duration)
*duration = i;
return ESP_OK;
}
}
return ESP_ERR_TIMEOUT;
}
/**
* Request data from DHT and read raw bit stream.
* The function call should be protected from task switching.
* Return false if error occurred.
*/
static inline esp_err_t dht_fetch_data(dht_sensor_type_t sensor_type, gpio_num_t pin, uint8_t data[DHT_DATA_BYTES])
{
uint32_t low_duration;
uint32_t high_duration;
// Phase 'A' pulling signal low to initiate read sequence
gpio_set_direction(pin, GPIO_MODE_OUTPUT_OD);
gpio_set_level(pin, 0);
ets_delay_us(sensor_type == DHT_TYPE_SI7021 ? 500 : 20000);
gpio_set_level(pin, 1);
// Step through Phase 'B', 40us
CHECK_LOGE(dht_await_pin_state(pin, 40, 0, NULL),
"Initialization error, problem in phase 'B'");
// Step through Phase 'C', 88us
CHECK_LOGE(dht_await_pin_state(pin, 88, 1, NULL),
"Initialization error, problem in phase 'C'");
// Step through Phase 'D', 88us
CHECK_LOGE(dht_await_pin_state(pin, 88, 0, NULL),
"Initialization error, problem in phase 'D'");
// Read in each of the 40 bits of data...
for (int i = 0; i < DHT_DATA_BITS; i++)
{
CHECK_LOGE(dht_await_pin_state(pin, 65, 1, &low_duration),
"LOW bit timeout");
CHECK_LOGE(dht_await_pin_state(pin, 75, 0, &high_duration),
"HIGH bit timeout");
uint8_t b = i / 8;
uint8_t m = i % 8;
if (!m)
data[b] = 0;
data[b] |= (high_duration > low_duration) << (7 - m);
}
return ESP_OK;
}
/**
* Pack two data bytes into single value and take into account sign bit.
*/
static inline int16_t dht_convert_data(dht_sensor_type_t sensor_type, uint8_t msb, uint8_t lsb)
{
int16_t data;
if (sensor_type == DHT_TYPE_DHT11)
{
data = msb * 10;
}
else
{
data = msb & 0x7F;
data <<= 8;
data |= lsb;
if (msb & BIT(7))
data = -data;       // convert it to negative
}
return data;
}
esp_err_t dht_read_data(dht_sensor_type_t sensor_type, gpio_num_t pin,
int16_t *humidity, int16_t *temperature)
{
CHECK_ARG(humidity || temperature);
uint8_t data[DHT_DATA_BYTES] = { 0 };
gpio_set_direction(pin, GPIO_MODE_OUTPUT_OD);
gpio_set_level(pin, 1);
PORT_ENTER_CRITICAL();
esp_err_t result = dht_fetch_data(sensor_type, pin, data);
if (result == ESP_OK)
PORT_EXIT_CRITICAL();
/* restore GPIO direction because, after calling dht_fetch_data(), the
* GPIO direction mode changes */
gpio_set_direction(pin, GPIO_MODE_OUTPUT_OD);
gpio_set_level(pin, 1);
if (result != ESP_OK)
return result;
if (data[4] != ((data[0] + data[1] + data[2] + data[3]) & 0xFF))
{
ESP_LOGE(TAG, "Checksum failed, invalid data received from sensor");
return ESP_ERR_INVALID_CRC;
}
if (humidity)
*humidity = dht_convert_data(sensor_type, data[0], data[1]);
if (temperature)
*temperature = dht_convert_data(sensor_type, data[2], data[3]);
ESP_LOGD(TAG, "Sensor data: humidity=%d, temp=%d", *humidity, *temperature);
return ESP_OK;
}
esp_err_t dht_read_float_data(dht_sensor_type_t sensor_type, gpio_num_t pin,
float *humidity, float *temperature)
{
CHECK_ARG(humidity || temperature);
int16_t i_humidity, i_temp;
esp_err_t res = dht_read_data(sensor_type, pin, humidity ? &i_humidity : NULL, temperature ? &i_temp : NULL);
if (res != ESP_OK)
return res;
if (humidity)
*humidity = i_humidity / 10.0;
if (temperature)
*temperature = i_temp / 10.0;
return ESP_OK;
}

步骤 3.1.6：按Ctrl+S保存文件。我们的库创建完成。

步骤 3.1.7：将新创建的源文件添加到CMakeList. 单击CMakeLists.txt将其打开并添加您的库源文件的名称，如下面的屏幕截图所示。

现在，您可以使用您的库了。按照相同的过程根据需要创建更多库。例如我们案例中的 ADS1115 库。该库的所有源文件都添加到代码部分以及 GitHub 存储库中。

步骤 3.2：测试库

让我们测试我们创建的 dht11 库以了解它是否正常工作。它还将证明我们的 DHT11 传感器的工作原理。

步骤 3.2.1：打开main.c文件，使用Ctrl+A选择整个代码，然后使用Ctrl+X剪切代码，使用Ctrl+V将代码粘贴到记事本文本文件中并保存。我们将再次使用它。

步骤 3.2.2：将以下代码片段粘贴到main.c文件中。此代码将使用新创建的 dht11 库从 dht11 传感器读取温度和湿度，并在终端中打印结果。

#include 
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "esp_log.h"
#include "esp_err.h"
#include "core2forAWS.h"

#include "dht.h"

static const dht_sensor_type_t sensor_type = DHT_TYPE_DHT11;
static const gpio_num_t dht_gpio = GPIO_NUM_26;
void temperature_task(void *arg) {
int16_t temperature = 0;
int16_t humidity = 0;
while (1)
    {
    if (dht_read_data(sensor_type, dht_gpio, &humidity, &temperature) == ESP_OK)
        printf("Humidity: %d%% Temp: %dC\n", humidity / 10, temperature / 10);
    else
        printf("Could not read data from sensor\n");
    vTaskDelay(pdMS_TO_TICKS(2000));
    }
}

void app_main()
{
    Core2ForAWS_Init();
    xTaskCreatePinnedToCore(&temperature_task, "temperature_task", 4096, NULL, 5, NUL,  1);
}

步骤 3.2.3：打开一个新终端并键入pio run --environment core2foraws以构建程序。

如果一切正常，您将收到成功消息。

步骤 3.2.4：使用以下命令将固件上传到 Core2 AWS IoT EduKit

pio run --environment core2foraws --target upload

步骤 3.2.5：将 DHT11 传感器连接到 Core2 Kit 的端口 B 并使用以下命令监控结果

pio run --environment core2foraws --target monitor

如果您从终端得到以下结果，那么恭喜！您的磁带库和 DHT11 传感器都运行良好。

步骤 3.3：修改 main.c 文件

将新的main.c源代码替换为我们存储在记事本文件中的原始main.c源代码。由于我们要发布比演示项目更多的数据，我们需要增加 JSON 缓冲区大小，如下所示：

#define MAX_LENGTH_OF_UPDATE_JSON_BUFFER 300

添加了以下变量用于存储空气质量参数。

float temperature = STARTING_ROOMTEMPERATURE;
float humidity = STARTING_ROOMHUMIDITY;
float nitrogen_dioxide = STARTING_ROOMNO2;
float ammonia = STARTING_ROOMNH3;
float carbon_monoxide = STARTING_ROOMCO;
float sulfur_dioxide = STARTING_ROOMSO2;
float methane = STARTING_ROOMCH4;
uint8_t soundBuffer = STARTING_SOUNDLEVEL;
uint8_t reportedSound = STARTING_SOUNDLEVEL;
bool fan_status = STARTING_FANSTATUS;
char toilet_status[14] = STARTING_TOILETSTATUS;

我们需要更多类型的处理程序变量jsonStruct_t来打包我们所有的传感器值。因此，已创建以下处理程序变量。

jsonStruct_t temperatureHandler;
temperatureHandler.cb = NULL;
temperatureHandler.pKey = "temperature";
temperatureHandler.pData = &temperature;
temperatureHandler.type = SHADOW_JSON_FLOAT;
temperatureHandler.dataLength = sizeof(float);

jsonStruct_t humidityHandler;
humidityHandler.cb = NULL;
humidityHandler.pKey = "humidity";
humidityHandler.pData = &humidity;
humidityHandler.type = SHADOW_JSON_FLOAT;
humidityHandler.dataLength = sizeof(float);

jsonStruct_t carbonMonoxideHandler;
carbonMonoxideHandler.cb = NULL;
carbonMonoxideHandler.pKey = "carbon_monoxide";
carbonMonoxideHandler.pData = &carbon_monoxide;
carbonMonoxideHandler.type = SHADOW_JSON_FLOAT;
carbonMonoxideHandler.dataLength = sizeof(float);

jsonStruct_t ammoniaHandler;
ammoniaHandler.cb = NULL;
ammoniaHandler.pKey = "ammonia";
ammoniaHandler.pData = &ammonia;
ammoniaHandler.type = SHADOW_JSON_FLOAT;
ammoniaHandler.dataLength = sizeof(float);

jsonStruct_t nitrogenDioxideHandler;
nitrogenDioxideHandler.cb = NULL;
nitrogenDioxideHandler.pKey = "nitrogen_dioxide";
nitrogenDioxideHandler.pData = &nitrogen_dioxide;
nitrogenDioxideHandler.type = SHADOW_JSON_FLOAT;
nitrogenDioxideHandler.dataLength = sizeof(float);

jsonStruct_t sulfurDioxideHandler;
sulfurDioxideHandler.cb = NULL;
sulfurDioxideHandler.pKey = "sulfur_dioxide";
sulfurDioxideHandler.pData = &sulfur_dioxide;
sulfurDioxideHandler.type = SHADOW_JSON_FLOAT;
sulfurDioxideHandler.dataLength = sizeof(float);

jsonStruct_t methaneHandler;
methaneHandler.cb = NULL;
methaneHandler.pKey = "methane";
methaneHandler.pData = &methane;
methaneHandler.type = SHADOW_JSON_FLOAT;
methaneHandler.dataLength = sizeof(float);

jsonStruct_t soundHandler;
soundHandler.cb = NULL;
soundHandler.pKey = "sound";
soundHandler.pData = &reportedSound;
soundHandler.type = SHADOW_JSON_UINT8;
soundHandler.dataLength = sizeof(uint8_t);

jsonStruct_t exhaustFanActuator;
exhaustFanActuator.cb = exhaustFan_Callback;
exhaustFanActuator.pKey = "fan_status";
exhaustFanActuator.pData = &fan_status;
exhaustFanActuator.type = SHADOW_JSON_BOOL;
exhaustFanActuator.dataLength = sizeof(bool);

jsonStruct_t toiletStatusActuator;
toiletStatusActuator.cb = toilet_status_Callback;
toiletStatusActuator.pKey = "toilet_status";
toiletStatusActuator.pData = &toilet_status;
toiletStatusActuator.type = SHADOW_JSON_STRING;
toiletStatusActuator.dataLength = strlen(toilet_status)+1;

以下第一个函数将我们想要发布到云的任何值打包到 IoT Core 影子服务所期望的影子文档中。第二个函数实际上将封送的影子文档作为有效负载通过网络发布到 IoT Core 的主题上$aws/things/<>/shadow/update，其中<>每个设备的唯一 ID。这两个函数修改如下。

if(SUCCESS == rc) {
    rc = aws_iot_shadow_add_reported(JsonDocumentBuffer, 
         sizeOfJsonDocumentBuffer, 10, &temperatureHandler,
         &humidityHandler, &carbonMonoxideHandler, 
         &ammoniaHandler, &nitrogenDioxideHandler, &sulfurDioxideHandler,
         &methaneHandler, &soundHandler, &toiletStatusActuator, &exhaustFanActuator);
    if(SUCCESS == rc) {
        rc = aws_iot_finalize_json_document(JsonDocumentBuffer, 
             sizeOfJsonDocumentBuffer);
        if(SUCCESS == rc) {
            ESP_LOGI(TAG, "Update Shadow: %s", JsonDocumentBuffer);
            rc = aws_iot_shadow_update(&iotCoreClient, client_id, JsonDocumentBuffer,
                 ShadowUpdateStatusCallback, NULL, 10, true);
            shadowUpdateInProgress = true;
            }
        }
}

以下代码表示我们的马桶状态执行器的回调函数。这是当设备接收到包含键值对的新消息时执行的代码。state.desired.toilet_status厕所状态是根据空气质量数据从 ML 模型中确定的。根据 IoT Core 返回的状态，颜色会发生变化。

void toilet_status_Callback(const char *pJsonString, uint32_t JsonStringDataLen, 
                   jsonStruct_t *pContext) {
    IOT_UNUSED(pJsonString);
    IOT_UNUSED(JsonStringDataLen);
    char * status = (char *) (pContext->pData);
    if(pContext != NULL) {
        ESP_LOGI(TAG, "Delta - toiletStatus state changed to %s", status);
    }
    if(strcmp(status, BUSY) == 0) {
        ESP_LOGI(TAG, "setting side LEDs to Yellow");
        Core2ForAWS_Sk6812_SetSideColor(SK6812_SIDE_LEFT, 0xFFFF00);
        Core2ForAWS_Sk6812_SetSideColor(SK6812_SIDE_RIGHT, 0xFFFF00);
        Core2ForAWS_Sk6812_Show();
    } else if(strcmp(status, UNCLEAN) == 0) {
        ESP_LOGI(TAG, "setting side LEDs to Red");
        Core2ForAWS_Sk6812_SetSideColor(SK6812_SIDE_LEFT, 0xFF0000);
        Core2ForAWS_Sk6812_SetSideColor(SK6812_SIDE_RIGHT, 0xFF0000);
        Core2ForAWS_Sk6812_Show();
    } else if(strcmp(status, READY) == 0) {
        ESP_LOGI(TAG, "clearing side Green");
        Core2ForAWS_Sk6812_SetSideColor(SK6812_SIDE_LEFT, 0x00FF00);
        Core2ForAWS_Sk6812_SetSideColor(SK6812_SIDE_RIGHT, 0x00FF00);
        //Core2ForAWS_Sk6812_Clear();
        Core2ForAWS_Sk6812_Show();
    }
}

使用以下两个函数来读取温度和空气质量。

void read_temperature(){
    int16_t temperature_data = 0;
    int16_t humidity_data = 0;
    if (dht_read_data(sensor_type, dht_gpio, &humidity_data, 
        &temperature_data) == ESP_OK){
    temperature = (float) temperature_data/10;
    humidity = (float) humidity_data/10;
    }
}

void read_airquality(){
    int16_t adc0, adc1, adc2;
    //float nitrogen_dioxide, ammonia, carbon_monoxide;
    adc0 = ADS1115_readADC_SingleEnded(CO_CHNNEL);
    carbon_monoxide = ADS1115_computeVolts(adc0);
    adc1 = ADS1115_readADC_SingleEnded(NH3_CHNNEL);
    ammonia = ADS1115_computeVolts(adc1);
    adc2 = ADS1115_readADC_SingleEnded(NO2_CHNNEL);
    nitrogen_dioxide = ADS1115_computeVolts(adc2);
}

以下函数将空气质量数据的 ppm 值接收到全局变量中。

void read_airquality_ppm(){
    carbon_monoxide = measure_in_ppm(CO);
    nitrogen_dioxide = measure_in_ppm(NO2);
    ammonia = measure_in_ppm(NH3);
    methane = measure_in_ppm(CH4);
}

app_main() 函数更新如下

void app_main()
{
    Core2ForAWS_Init();
    Core2ForAWS_Display_SetBrightness(80);
    Core2ForAWS_LED_Enable(1);
    ADS1115_I2CInit();
    ADS1115_setGain(GAIN_TWOTHIRDS);
    airquality_calibrate ();
    xMaxNoiseSemaphore = xSemaphoreCreateMutex();
    ui_init();
    initialise_wifi();
    xTaskCreatePinnedToCore(&aws_iot_task, "aws_iot_task", 4096*2, NULL, 5, NULL, 1);
}

完整的main.c程序附在代码部分和 GitHub ripo 中。

步骤 3.4：测试固件

完成所有修改后，构建程序并将其上传到 Core2。将 I2C 传感器连接到端口 A，将 DHT11 传感器连接到端口 B，然后运行调试终端来监控输出。

如果一切顺利，您将从终端获得以下输出。

打开 AWS IoT Core 控制台测试页面，订阅主题$aws/things/<>/shadow/update/accepted，您应该会看到新消息与您的vTaskDelay()一起及时到达。（用屏幕上打印的设备客户端 ID/序列号替换 «CLIENT_ID»。）输出如下图所示。

在 AWS IoT Core 控制台测试页面中，单击 Publish to a topic 选项卡并在该主题上发布以下新影子消息$aws/things/<>/shadow/update。您应该会看到 Core for AWS IoT EduKit 的 LED 条从绿色变为红色。请参阅下面的示例影子消息。每次发布消息时，通过更改厕所状态（设置为BUSY或READY ）和/或fan_status值（设置为true或false ）来测试效果。

{ "state": { "desired": { "toilet_status": "UNCLEAN", "fan_status": true } } }

效果也会出现在终端中，如下所示。

从设备方面来看，一切都很好。现在我们将配置 AWS IoT Cloud 以转换和路由从设备接收的数据。

开发机器学习模型以自动检测厕所状态（UNCLEAN、BUSY、READY）

我们正在为我们的项目考虑厕所的三种状态。一种是就绪状态，表示空气质量数据良好，马桶使用安全舒适。在这种情况下，LED 条显示绿灯。另一种状态是BUSY状态。当某些空气质量参数不够好时，用户可能会感到不舒服。可能的原因可能是存在较高的湿度、气味或难闻的气味。在这种情况下，运行排气扇几分钟可以使情况变得更好。因此，这是不建议使用马桶时的状态，系统会通过花一些时间和运行风扇来尝试使其变得更好。在此状态下启用黄灯。另一种状态是 UNCLEAN 状态，此时某些空气参数不在可容忍的范围内，在这种情况下使用马桶会产生健康问题。即使只有通风也不足以使其变得更好，但需要清洁工手动清洁。设备在此状态下显示红灯，并通过电子邮件向相关人员发送清洁请求。

问题是这三种状态的确定并不容易。没有公式可以找到厕所处于哪种状态。感谢机器学习。一个简单的机器学习模型可以帮助从空气质量数据的分类中智能地识别状态。同样，训练机器学习模型并不容易，构建一个好的模型需要专业知识、对输入的全面理解，以及经过多个周期的努力来优化它。但如今，借助现代数据科学工具链，无需任何 ML 专业知识即可制作 ML 模型，而 Amazon SageMaker 工具链就是其中之一，但我们不能指望第一次尝试就能生成一个好的模型。

为了建立一个好的模型，我们需要大量的数据。越多越好。构建 ML 模型是一个两步过程，收集数据和训练模型。为了准确检测，我们需要为每种情况收集大量数据，之前讨论过的三种状态。因此，我们需要在干净的马桶中部署我们的设备至少几个小时（最好是 24 小时），在不舒服的马桶中部署几个小时，在不卫生的马桶中部署几个小时，以收集足够的数据来构建准确的 ML 模型。第二个不干涉步骤是在您收集到足够的数据后，ML 模型正在经历一个自动训练过程。此培训过程可能需要几个小时，我们建议您在早上开始，下午返回，或者让它运行一整夜。

我们将使用设备遥测的聚合数据集来支持自动 ML 训练实验，然后使用从训练的 ML 模型推断出的新厕所状态值对新设备报告进行分类。

我们将遵循的工作流程具有以下关键组件：

• 我们将创建一个新的 IoT Core 规则，它将报告的设备数据转发到名为 AWS IoT Analytics 的服务。IoT Analytics 服务批量存储原始 IoT 数据，对其进行转换和清理以将原始数据转换为处理后的数据，并提供查询引擎来对处理后的数据进行切片，以用于分析工作流或 ML 训练。
• 我们的 IoT Analytics 项目将包含四个链接在一起的资源：一个用于存储来自设备影子的原始 IoT 数据的通道、一个用于转换/过滤/丰富数据的管道、一个用于处理数据的数据存储，以及一个运行保存的查询和可以发送结果进行处理。
• Amazon SageMaker Studio 是一个集成的机器学习环境，您可以在其中构建、训练、部署和分析您的模型，所有这些都在同一个应用程序中进行。
• 一个 Amazon SageMaker 终端节点，将您的训练模型作为可使用的 API 托管。
• 我们将创建一个 AWS Lambda 函数，该函数将运行一些简单的代码来处理已发布的消息并对我们的新 ML 模型端点进行推理。

步骤 1：创建 IoT Core 规则以将遥测数据转发到 IoT Analytics

• 一世。在AWS IoT Core 控制台中，依次选择Act 、Rules和Create 。
• ii. 给你的规则起一个名字，比如厕所DataToAnalyticsForState和描述。
• iii. 使用以下查询。请务必将«CLIENT_ID»替换为打印在 Core2 for AWS IoT Edukit 参考硬件套件屏幕上的客户端 ID/序列号。

SELECT current.state.reported.temperature, current.state.reported.humidity, current.state.reported.carbon_monoxide, current.state.reported.ammonia, current.state.reported.nitrogen_dioxide, current.state.reported.sulfur_dioxide, current.state.reported.methane, current.state.reported.sound, current.state.reported.toilet_status, current.state.reported.fan_status, timestamp FROM '$aws/things/<>/shadow/update/documents'

• iv. 为设置一项或多项操作选择添加操作。
• v.选择向 IoT Analytics 发送消息，然后选择配置操作。
• 六。选择快速创建 IoT Analytics 资源并为Resource prefix提供项目名称。进一步的模块步骤假设前缀是toiletAnalyticsResourceForState。选择快速创建，您的所有 AWS IoT Analytics 资源将自动创建和配置。
• 七。选择添加操作以配置此操作并返回到规则创建表单。
• 八。选择创建规则以创建新规则。

验证步骤

在继续下一章之前，您可以验证您的无服务器应用程序是否按预期配置：

• 确保您的 Core2 设备已开机、发布数据并部署在您要训练的厕所中。
• 使用 AWS IoT Analytics 控制台，查看最新的数据集内容，并验证所有参数（如温度、湿度、氨气、二氧化硫等）的历史记录以及时间戳。要检查这一点，请在IoT Analytics 控制台中找到您的数据集，选择Actions和Run now ，然后等待查询完成。在“内容”选项卡中，单击最近创建的包含最新内容的内容预览。您应该会看到类似于以下内容的结果：

第 2 步：设置 Amazon SageMaker Studio 以进行自动模型训练

首先，您将设置 Amazon SageMaker Studio，以便为自动模型训练配置新实验。

• 一世。转到 Amazon SageMaker 控制台并选择Amazon SageMaker Studio 。
• ii. 选择Quick start并可选择输入一个新的用户名，如厕所状态用户。
• iii. 对于Execution role选择下拉菜单并选择Create a new IAM role 。
• iv. 对于您指定的 S3 存储桶，选择None ，然后选择Create role 。名称中带有“sagemaker”的存储桶的其他默认值足以满足此项目。
• v.选择提交以开始 SageMaker Studio 的配置过程。代表您完成此步骤需要几分钟时间。

步骤 3：在 SageMaker Studio中配置新项目

• 一世。从 SageMaker Studio 控制面板中，为新创建的用户选择打开 Studio 。
• ii. 在 Launcher 选项卡中，选择New project 。
• iii. 在SageMaker 项目模板下，选择用于模型构建、训练和部署的 MLOps 模板，然后选择选择项目模板。

• iv. 给你的项目起一个名字，比如厕所StateSageMakerProject和描述，然后选择Create project 。

创建项目后，我们将获得一个项目仪表板，其中包含存储库、管道、实验等选项卡。让我们将此浏览器选项卡对 SageMaker Studio 保持打开状态，以便您可以快速返回此页面。

步骤 4：导出 AWS IoT Analytics 数据

• 一世。打开AWS IoT Analytics 控制台并选择您的数据集（假定名称为Toiletanalyticsresourceforstate_dataset）。
• ii. 在数据集内容交付规则下选择编辑。
• iii. 选择Add rule ，然后选择Deliver result to S3 。
• iv. 在S3 存储桶下选择请选择一个资源并找到为您的 SageMaker Studio 项目创建的 S3 存储桶。它将被命名为sagemaker-project-p-wzq6773hm0gv. 如果有多个这样命名的存储桶，您需要检查 SageMaker Studio 项目以获取项目的随机哈希 ID。您可以在项目的其他资源（例如存储库和管道选项卡）中查看哈希值。

• v.在Bucket key 表达式下使用这个表达式：data/healthytoilet/Version/!{iotanalytics:scheduleTime}_!{iotanalytics:versionId}.csv
• 六。在Role下，选择Create new并为 IAM 角色提供一个名称，该角色将授予 IoT Analytics 访问权限以将数据写入您的 S3 存储桶。选择创建角色。
• 七。选择保存以完成新的递送规则。

• 八。要生成将保存到您的新 Amazon S3 存储桶以进行训练的数据集，请选择Actions然后Run now 。数据集内容生成完成后，您应该会看到结果预览更新。

我们现在已准备好在 SageMaker Studio 中开始您的 ML 实验。实验将使用我们的 IoT Analytics 数据集刚刚导出的报告恒温器数据作为输入。我们将配置实验以寻找准确预测现有厕所状态列的方法。自动训练作业将分析您的数据以尝试相关算法，然后运行 ​​250 个具有不同超参数的训练作业，选择最适合您的输入训练数据的一个。

在开始您的 ML 实验之前，我们应该从我们要分析的马桶中的设备报告几个小时的数据，并且在那段时间马桶应该有准备好的、忙碌的和不干净的情况。一个自动 ML 实验需要至少 500 行数据才能工作，但我们带来的数据越多，结果就会越好。如果在继续之前我们仍需要生成更多数据，我们需要重新运行 IoT Analytics 控制台中的数据集（上一个指令列表的最后一步），以便我们的项目 S3 存储桶中的 SageMaker 可以使用这些结果。

第 4 步：在 SageMaker Studio 中启动 ML 实验

• 一世。返回您的 SageMaker Studio，打开您的项目，选择Experiments选项卡并选择Create autopilot Experiment 。

• ii. 为您的实验命名。
• iii. 在项目下，从列表中选择您的项目。
• iv. 在连接您的数据和S3 存储桶名称下，在列表中找到并选择您项目的 S3 存储桶。这与您在上一步中为 IoT Analytics 数据集内容交付规则选择的规则相同。
• v.在数据集文件名下找到并选择您的 IoT Analytics 数据集内容，例如data/healthytoilet/Version/1607276270943_3b4eb6bb-8533-4ac0-b8fd-1b62ac0020a2.csv.
• 六。在目标下选择toilet_status。
• 七。在输出数据位置和S3 存储桶名称下，在此列表中查找并选择您在步骤 4 中选择的相同项目 S3 存储桶。
• 八。在Dataset directory name下输入output/healthytoilet并选择Use input as S3 object key prefix “output/healthytoilet” 。这会在 S3 存储桶中定义一个新前缀，用于您的输出文件。
• 十四。选择Create Experiment以启动自动化 ML 实验。

运行实验可能需要几分钟到几小时。您可以在 SageMaker Studio 浏览器选项卡中跟踪实验的进度，但关闭选项卡并稍后返回查看进度也是安全的。

实验结束后，结果输出是 250 次试验，SageMaker 使用这些试验来寻找最佳调整作业参数。对试验表进行排序以找到标记为Best的那一项。下一个里程碑是将此试用版部署为模型端点，以便您可以将其作为 API 调用。

第 5 步：部署最佳 ML 模型

• 一世。选择标记为Best的试用版，然后选择Deploy model 。

• ii. 为您的端点命名。本模块中的进一步步骤假定名称 healthToiletStateEndpoint。
• iii. 在Inference Response Content下，同时选择predict_label和probability 。predict_label可能已经添加到列表中。
• iv. 选择部署模型以告诉 SageMaker 将您的模型部署为新的可消耗 API 端点。这将需要几分钟。

现在，您的机器学习模型已部署为 API 终端节点，由 Amazon SageMaker 管理。在下一章“使用 ML 模型”中，您将使用无服务器函数使用 API 端点，并用模型生成的推理替换 IoT Core 规则中确定值的简单阈值逻辑。toilet_status

验证步骤

在继续下一步之前，您可以验证您的无服务器应用程序是否按预期配置......

• 使用Amazon SageMaker 控制台在Endpoints页面上查看状态为InService的新终端节点。

第 6 步：设置 AWS Lambda 并调用 SageMaker Endpoint

以下步骤将引导我们在 AWS Lambda 中创建无服务器函数。该函数定义了一小段代码，期望来自 IoT Core 的设备影子消息，将消息转换为与我们的 ML 端点一起使用的格式，然后调用 ML 端点以返回厕所状态的分类和推理的置信度分数。

• 一世。在 AWS Lambda 控制台中，选择Create function 。
• ii. 输入函数的名称。进一步的步骤假定名称classifyToiletStatus。
• iii. 在运行时，选择Python 3.8 。
• iv. 选择创建函数。

• v.在Function code下，在文件lambda_function.py中，复制并粘贴以下代码以替换占位符代码：

import json
import boto3
import os

# Receives a device shadow Accepted document from IoT Core rules engine.
# Event has signature like {"state": {"reported": {"sound": 5}}}.
# See expectedAttributes for full list of attributes expected in state.reported.
# Builds CSV input to send to SageMaker endpoint, name of which stored in
#   environment variable SAGEMAKER_ENDPOINT.
#
# Returns the prediction and confidence score from the ML model endpoint.
def lambda_handler(event, context):
    client = boto3.client('sagemaker-runtime')
    
    print('event received: {}'.format(event))
    
    # Order of attributes must match order expected by ML model endpoint. E.g.
    #   the same order of columns used to train the model.
    expectedAttributes = ['temperature', 'humidity', 'carbon_monoxide', 'ammonia', 'nitrogen_dioxide', 'sulfur_dioxide', 'methane', 'sound', 'toilet_status', 'fan_status', 'timestamp']
    reported = event['state']['reported']
    reported['timestamp'] = event['timestamp']
    reportedAttributes = reported.keys()
    
    # Validates the input event has all the expected attributes.
    if(len(set(expectedAttributes) & set(reportedAttributes)) < len(expectedAttributes)):
        return {
            'statusCode': 400,
            'body': 'Error: missing attributes from event. Expected: {}. Received: {}.'.format(','.join(expectedAttributes), ','.join(reportedAttributes))
        }
    
    # Build the input CSV string to send to the ML model endpoint.
    reportedValues = []
    for attr in expectedAttributes:
        reportedValues.append(str(reported[attr]))
    input = ','.join(reportedValues)
    print('sending this input for inference: {}'.format(input))

    endpoint_name = os.environ['SAGEMAKER_ENDPOINT']
    content_type = "text/csv"
    accept = "application/json"
    payload = input
    response = client.invoke_endpoint(
        EndpointName=endpoint_name, 
        ContentType=content_type,
        Accept=accept,
        Body=payload
        )
        
    body = response['Body'].read()
        
    print('received this response from inference endpoint: {}'.format(body))
    
    return {
        'statusCode': 200,
        'body': json.loads(body)['predictions'][0]
    }

• 六。在 Configuration 选项卡下单击Environment variables ，然后选择Edit并添加一个新的环境变量。

• 七。对于Key输入SAGEMAKER_ENDPOINT，对于Value输入您的 SageMaker 终端节点的名称。您将此资源命名为部署模型的最后一步，并且此模块假定名称为healthyToiletStateEndpoint.

• 八。选择保存以提交此新环境变量并返回主 Lambda 编辑器界面。
• 九。在设计器面板中，选择+ 添加触发器。
• X。对于触发器配置，从列表中选择AWS IoT 。
• 十一。对于IoT 类型，选择自定义 IoT 规则。
• 十二。对于Rule ，选择Create a new rule，提供一个角色名称，如厕所IotLambdaInvoke并将以下文本粘贴到 Rule 查询语句中。将 DEVICE_ID 替换为您自己的设备 ID，然后单击Add 。

SELECT cast(get(get(aws_lambda("arn:aws:lambda:REGION:ACCOUNT_ID:function:FUNCTION_NAME", *), "body"), 
"predicted_label") AS String) AS state.desired.toilet_status 
FROM '$aws/things/<>/shadow/update/accepted' WHERE state.reported.temperature <> Null

确保替换占位符：将 REGION 更改为您当前的区域，如控制台标题中所示（它必须是格式us-west-2而不是Oregon）；将 ACCOUNT_ID 更改为您的 12 位帐户 ID，不带连字符，这也显示在打印您的用户名的控制台标题菜单中；并将 FUNCTION_NAME 更改为您创建的 AWS Lambda 函数的名称（假定名称为classifyToiletStatus）。不要忘记更新 FROM 主题中的 «CLIENT_ID» 占位符。

• 十三。选择权限选项卡，然后选择角色名称下的链接，以便您可以为此 Lambda 函数添​​加权限以调用您的 SageMaker 终端节点。
• 十四。从打开到 IAM 控制台的新选项卡中，在Permissions policies下选择Add inline policy 。
• 十五。对于服务，选择SageMaker 。
• 十六。对于操作，选择InvokeEndpoint 。
• 十七。对于资源，选择所有资源。
• 十八。选择审查政策。
• 十九。为您的策略命名，invokeSageMakerEndpoint然后选择Create policy 。您现在可以关闭这个新的浏览器选项卡。

测试 Lambda 函数

• 一世。转到您的 Lambda 函数并选择代码选项卡。如果您没有部署您的 Lambda 代码，请单击Deploy

• ii. 您将收到Changes 已部署的确认信息。单击测试并选择配置测试事件。

• iii. 给出配置的名称，如 prediction_test 并在代码编辑器中粘贴示例 JSON 数据（如设备上传的 JSON），删除之前的数据：

{
  "state": {
    "reported": {
      "temperature": 29,
      "humidity": 39,
      "carbon_monoxide": 4.384765,
      "ammonia": 0.687,
      "nitrogen_dioxide": 0.141511,
      "sulfur_dioxide": 0.441511,
      "methane": 837.172485,
      "sound": 23,
      "toilet_status": "BUSY",
      "fan_status": false
    }
  },
  "metadata": {
    "reported": {
      "temperature": {
        "timestamp": 1630912876
      },
      "humidity": {
        "timestamp": 1630912876
      },
      "carbon_monoxide": {
        "timestamp": 1630912876
      },
      "ammonia": {
        "timestamp": 1630912876
      },
      "nitrogen_dioxide": {
        "timestamp": 1630912876
      },
      "sulfur_dioxide": {
        "timestamp": 1630912876
      },
      "methane": {
        "timestamp": 1630912876
      },
      "sound": {
        "timestamp": 1630912876
      },
      "toiletStatus": {
        "timestamp": 1630912876
      },
      "fanStatus": {
        "timestamp": 1630912876
      }
    }
  },
  "version": 560,
  "timestamp": 1630912876,
  "clientToken": "01231c94f550fe1c01-3"
}

• iv. 单击以创建。

v.现在单击测试，您将获得以下 JSON 响应。它返回带有概率的厕所状态的预测标签。请注意，在我们提供的数据中，厕所状态为 BUSY，但 ML 模型根据空气质量数据预测它为 READY。

{
  "statusCode": 200,
  "body": {
    "predicted_label": "READY",
    "probability": "0.4254942536354065"
  }
}

此时，我们的 IoT 工作流程现在正在使用来自其部署的端点的经过训练的机器学习模型，以将设备发布的消息分类为新的厕所状态值！

在厕所IotLambdaInvoke规则上添加操作

转到toiletIotLambdaInvoke规则添加一个操作以根据预测更新设备阴影。

• 选择添加操作。
• 选择将消息重新发布到 AWS IoT 主题，然后选择配置操作。
• 对于主题，使用$$aws/things/<>/shadow/update. 请务必将«CLIENT_ID»替换为您设备的客户端 ID/序列号。
• 对于选择或创建角色以授予 AWS IoT 访问权限以执行此操作。选择Create Role并在弹出窗口中为您的新 IAM 角色命名，然后选择Create role 。
• 选择添加操作以完成操作配置并返回到规则创建表单。
• 单击创建规则以在 AWS IoT 规则引擎中创建此规则。

这将更新设备影子，如下所示：

{
  "state": {
    "desired": {
      "toilet_status": "UNCLEAN",
      "fan_status": false
    }
  },
  "metadata": {
    "desired": {
      "toilet_status": {
        "timestamp": 1632032337
      },
      "fan_status": {
        "timestamp": 1632032337
      }
    }
  },
  "version": 6735,
  "timestamp": 1632032337
}

发送厕所不干净的通知

当设备发布遥测数据时，IoT 规则会触发 Lambda 函数来读取 SageMaker 预测并相应地更新设备影子。我们希望在厕所状态预测级别为 UNCLEAN 时向涉及清洁的人员发送通知（电子邮件）。每次厕所状态从另一个状态变为 UNCLEAN 时，我们只想发送一封电子邮件。因此，我们将订阅delta主题。

步骤1：创建发送 SMS 文本消息的 Amazon SNS 主题

创建一个 Amazon SNS 主题。

• 登录到Amazon SNS 控制台。
• 在左侧导航窗格中，选择主题。
• 在主题页面上，选择创建主题。
• 在Details中，选择Standard类型。默认情况下，控制台会创建一个 FIFO 主题。
• 在Name中，输入 SNS 主题名称。对于本教程，输入toiletUncleanNotice
• 滚动到页面末尾并选择Create topic 。控制台将打开新主题的详细信息页面。

创建 Amazon SNS 订阅。

• 在toiletUncleanNotice主题的详细信息页面中，选择Create subscription 。
• 在Create subscription的Details部分的Protocol列表中，选择SMS 。
• 在Endpoint中，输入可以接收短信的电话号码。确保输入以 开头+，包括国家和地区代码，并且不包括任何其他标点符号。
• 选择创建订阅。

测试 Amazon SNS 通知。

• 在Amazon SNS 控制台的左侧导航窗格中，选择主题。
• 要打开主题的详细信息页面，请在主题列表中的主题列表中选择toiletUncleanNotice。
• 要打开Publish message to topic页面，请在high_temp_notice详细信息页面中，选择Publish message 。
• 在Publish message to topic的Message body部分的Message body to send to the endpoint中，输入一条短消息。
• 向下滚动到页面底部并选择Publish message 。
• 在您之前创建订阅时使用的号码的手机上，确认已收到消息。

如果您没有收到测试消息，请仔细检查电话号码和手机设置。

步骤2：创建 AWS IoT 规则以发送短信

• 在AWS IoT Core 控制台中，依次选择Act 、Rules和Create 。
• 为您的规则命名，如sendSnsUnclean和描述。
• 使用以下查询。请务必将«CLIENT_ID»替换为打印在 Core2 for AWS IoT Edukit 参考硬件套件屏幕上的客户端 ID/序列号。

SELECT state.toilet_status AS state.toilet_status FROM '$aws/things/«CLIENT_ID»/shadow/update/delta' WHERE state.toilet_status = "UNCLEAN"

当马桶状态从另一个状态变为另一个状态时，上述查询语句将生成以下 JSON 消息UNCLEAN。

{
  "state": {
    "toilet_status": "UNCLEAN"
  }
}

• 要打开此规则的规则操作列表，请在设置一个或多个操作中，选择添加操作。
• 在选择操作中，选择将消息作为 SNS 推送通知发送。
• 要打开所选操作的配置页面，请在操作列表底部选择配置操作。
• 在配置操作中：
• 在SNS 目标中，选择Select ，找到名为high_temp_notice的 SNS 主题，然后选择Select 。
• 在消息格式中，选择RAW 。
• 在选择或创建角色以授予 AWS IoT 访问权限以执行此操作中，选择创建角色。
• 在Create a new role 中，在Name中，输入新角色的唯一名称。对于本教程，使用sns_rule_role.
• 选择创建角色。

此规则将以原始 JSON 格式发送电子邮件。我们可以使用 Lambda 将其格式化为用户友好的格式。在这种情况下，我们将选择向 Lambda 函数发送消息，而不是作为规则操作将消息作为 SNS 推送通知发送，并且 Lambda 会将消息作为 SNS 发送。所以让我们先创建一个 Lambda 函数。

使用 AWS Lambda 函数格式化通知

上面的教程是关于如何发送 Amazon SNS 通知，由规则的查询语句产生的 JSON 文档作为文本消息的正文发送。结果是一条类似于以下示例的文本消息：

{
  "state": {
    "toilet_status": "UNCLEAN"
  }
}

在本教程中，您将使用 AWS Lambda 规则操作来调用 AWS Lambda 函数，该函数将规则查询语句中的数据格式化为更友好的格式，例如以下示例：

The toilet is very UNCLEAN and need to clean immediately. So, you are requested to take immediate action for cleaning. Thank you.

此项目中的 AWS Lambda 函数接收规则查询语句的结果，将元素插入文本字符串，并将结果字符串作为通知中的消息发送到 Amazon SNS。

创建发送文本消息的 AWS Lambda 函数

创建一个新的 AWS Lambda 函数。

• 在AWS Lambda 控制台中，选择Create function 。
• 在创建函数中，选择使用蓝图。搜索并选择hello-world-python蓝图，然后选择配置。
• 在Function name中，输入此函数的名称，formatUncleanToiletNotification。
• 在执行角色中，选择从 AWS 策略模板创建新角色。
• 在角色名称中，输入新角色的名称formatUncleanToiletNotificationRole。
• 在Policy templates - optional中，搜索并选择Amazon SNS 发布策略。
• 选择创建函数。

修改蓝图代码以格式化并发送 Amazon SNS 通知。

• 创建函数后，您应该会看到format-h​​igh-temp-notification详细信息页面。如果您不这样做，请从Lambda函数页面打开它。
• 在format-h​​igh-temp-notification详细信息页面中，选择Configuration选项卡并滚动到Function code面板。
• 在函数代码窗口的环境窗格中，选择 Python 文件lambda_function.py.
• 在Function code窗口中，从蓝图中删除所有原始程序代码，并用此代码替换它。将 替换为notify_topic_arn通知主题中的 ARN。

import boto3

def lambda_handler(event, context):

    # Create an SNS client to send notification
    sns = boto3.client('sns')

    # Format text message from data
    message_text = "The toilet is very {0} and need to clean immediately.".format(
            str(event['state']['toilet_status'])
        )

    # Publish the formatted message
    response = sns.publish(
            TopicArn = event['notify_topic_arn'],
            Message = message_text
        )

    return response

• 选择部署。

现在再次转到上一个规则操作，

• 删除之前的操作，然后单击添加操作
• 在选择操作中，选择向 Lambda 函数发送消息。
• 要打开所选操作的配置页面，请在操作列表底部选择配置操作。

在配置操作中：

• 在函数名称中，选择选择。
• 选择format-h​​igh-temp-notification 。
• 在配置操作的底部，选择添加操作。
• 要创建规则，请在创建规则底部选择创建规则。

 

 

所有步骤的视频教程

这是到目前为止解释的所有步骤的屏幕记录。初学者遵循上述步骤将很有帮助。视频中没有声音。

 

使用 SageMaker 检测漏水

按照相同的过程，您可以训练模型来检测漏水。您不需要使用所有参数来开发模型。当然，在这种情况下，噪声级将是最有价值的资源。当我解释我如何使用 IoT Analytics、SageMaker 和 Lambda 开发和训练厕所状态的机器学习模型时，我不再在这里重复这个过程。

将 NodeMCU 连接到 AWS IoT Core

在我们的项目中，Node MCU 将根据从 AWS IoT Core 收到的命令控制排气扇。IoT Core 规则将根据从厕所接收到的空气质量数据将控制消息（真或假）发布到特定的 MQTT 主题。该规则适用于相关重要空气质量参数的预设阈值。如果任何参数超过阈值，则 IoT Core 规则会向node/mcu/fan/state等主题发布一条真实消息。节点 MCU 收到此消息并打开排气扇，反之亦然。

要从 AWS IoT Core 节点 MCU 接收 MQTT 消息，必须连接到 IoT Core。按照链接 ( https://nerdyelectronics.com/iot/how-to-connect-nodemcu-to-aws-iot-core/ ) 中的教程了解如何将节点 MCU 连接到 AWS IoT Core。这是关于这个主题的一个非常好的教程。

你也可以关注视频：

为 Node MCU 开发固件

将 Node MCU 板成功连接到 AWS IoT Core 后，我们需要开发板的固件，以便它可以接收来自我们特定主题的消息并可以相应地控制排气扇。

固件是在 Arduino 中开发的，成功编译需要以下 Arduino 库。

#include "FS.h"
#include   //tested esp8266 core version: 2.5.2
#include  //tested version: 2.7.0
#include     //tested version: 3.2.0
#include 
#include   //tested version: 6.18.4

为了接收消息，我们需要像这样在成功连接到 IoT Core 后订阅一个主题

void reconnect() 
{
  // Loop until we're reconnected
  while (!client.connected()) 
  {
    Serial.print("Attempting MQTT connection...");
    // Attempt to connect
    if (client.connect("ESPthing")) 
    {
      Serial.println("connected");
      client.subscribe("node/mcu/fan/state");
      Serial.println("subscribed");
    } 
    else
    {
      Serial.print("failed, rc=");
      Serial.print(client.state());
      Serial.println(" try again in 5 seconds");
  
      char buf[256];
      espClient.getLastSSLError(buf, 256);
      Serial.print("WiFiClientSecure SSL error: ");
      Serial.println(buf);
  
      // Wait 5 seconds before retrying
      delay(5000);
    }

    
  }
}

以下回调函数接收MQTT消息并控制风扇

void callback(char* topic, byte* payload, unsigned int length) 
{
  Serial.print("Message arrived [");
  Serial.print(topic);
  Serial.print("] ");
  for (int i = 0; i < length; i++) 
  {
    message.concat((char)payload[i]); 
  }
  Serial.print(message);
  deserializeJson(doc, message);
  bool fan_status = doc["state"]["desired"]["fan_status"];
  Serial.println();
  if(fan_status == true){
    //turn on fan
    digitalWrite(RELAY, HIGH);
    Serial.println("Fan is ON");
    }
  else if(fan_status == false){
    //turn off fan
    digitalWrite(RELAY, LOW);
    Serial.println("Fan is OFF");
    }
  message = "";
}

Node MCU 的完整源代码附在代码部分。

验证 NodeMCU 固件

为了验证 Node MCU 正在接收 MQTT 消息并控制风扇，请转到 IoT Core 测试客户端页面并将以下消息发布到主题。node/mcu/fan/state

{ "state": { "desired": {"fan_status": true } } }

您还可以通过将 fan_status 更改为 false 来进行检查。打开 Arduino 串口监视器，您将得到以下响应。

恭喜！它正在工作。为了控制风扇，您需要通过继电器将排气扇连接到节点 MCU 的数字引脚。我使用了 D3，但您可以使用任何其他引脚。请参阅以下连接图以获得更好的理解。

创建 IoT 规则以发布粉丝状态

我们将创建 IoT Core 主题规则，接收设备从厕所发布的消息，检查采样的温度、湿度、氨气、二氧化硫和甲烷水平，并更新设备影子的风扇状态，并重新发布消息到另一个物联网主题。主题规则将使用 SQL 查询中的条件逻辑来构建新的有效负载，并使用 IoT Core 重新发布操作将新的有效负载发送到新主题。

• 转到 AWS IoT Core 管理控制台，选择Act ，选择Rules ，然后选择Create 。
• 为您的规则命名和描述。名字就像healthryToiletFanStateRepublish。
• 使用以下查询并确保将«CLIENT_ID»替换为您的设备客户端 ID/序列号。

SELECT CASE state.reported.temperature > 35 OR state.reported.humidity > 50 OR 
state.reported.ammonia > 3 OR state.reported.sulfur_dioxide > 2 OR 
state.reported.methane > 7 WHEN true THEN true ELSE false END AS 
state.desired.fan_status FROM '$aws/things/<>/shadow/update/accepted' 
WHERE state.reported.temperature <> Null

• 选择添加操作。
• 选择将消息重新发布到 AWS IoT 主题，然后选择配置操作。
• 对于主题，使用node/mcu/fan/state.
• 对于选择或创建角色以授予 AWS IoT 访问权限以执行此操作。选择Create Role并在弹出窗口中为您的新 IAM 角色命名，然后选择Create role 。
• 选择添加操作以完成操作配置并返回到规则创建表单。
• 单击创建规则以在 AWS IoT 规则引擎中创建此规则。

SELECT 子句使用 CASE 语句来实现我们的简单阈值带。如果任何空气质量参数（如温度、湿度、氨、二氧化硫或甲烷）超过比较值，则将风扇状态修改为真 (ON)。您可以根据自己的情况修改解决方案的阈值或参数。

CASE 语句的输出state.desired.fan_status使用 AS 关键字保存到有效负载键。这意味着我们正在创建一个新的有效负载{"state": {"desired": {"fan_status": false}}}，并将此有效负载发送到操作。

state.reported.temperature <> Null防止规则重新触发，因为新的影子更新有效负载仅包含state.desired.fan_status密钥而没有值state.reported.temperature（可能使用其他参数，例如湿度）。

现在所有的排气扇都设置好了，它可以被 IoT Core MQTT 消息控制了。