描述

介紹

由于文明的發展以及工業和汽車的不潔排放量增加，大氣狀況每年都在惡化。盡管空氣是生命不可缺少的資源，但許多人對空氣污染的嚴重程度漠不關心，或者直到最近才意識到這個問題。在水、土壤、熱、噪聲等各類污染物中，空氣污染是最危險、最嚴重的，它會引起氣候變化和危及生命的疾病。根據世界衛生組織 (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 傳感器檢測一氧化碳、二氧化氮、氨和甲烷。該傳感器通過三個獨立的模擬通道提供數據。為了測量二氧化硫，我使用了 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 消息控制了。

?

?

?

?