代码¶
Warning
以下代码应基于发布代码中的代码，可能已更新。
sensing_command.h¶
/**
* @file sensing_command.h
* @author SHUAIWEN CUI (SHUAIWEN001@e.ntu.edu.sg)
* @brief MQTT remote control command handler for sensing modules
* @version 1.0
* @date 2025-12-17
* @copyright Copyright (c) 2025
*
* @details
* This module handles MQTT commands for controlling online and offline sensing.
* Command format: SENSE,ONLINE or SENSE,OFFLINE with parameters
*
* Supported commands:
* - SENSE,ONLINE,F=frequency,D=duration
* - SENSE,ONLINE,STOP
* - SENSE,ONLINE,STATUS
* - SENSE,OFFLINE,F=frequency,D=duration
* - SENSE,OFFLINE,F=frequency,D=duration,DL=delay_seconds
* - SENSE,OFFLINE,F=frequency,D=duration,TIME=YYMMDDHHMMSS
* - SENSE,OFFLINE,STOP
* - SENSE,OFFLINE,STATUS
*/
#pragma once
#include <stdint.h>
#include <stdbool.h>
#include "esp_err.h"
#include "node_acc_adxl355.h"
#ifdef __cplusplus
extern "C"
{
#endif
/* ============================================================================
* FUNCTION DECLARATIONS
* ============================================================================ */
/**
* @brief Process MQTT sensing command
* @param command Command string (e.g., "SENSE,ONLINE,F=20,D=60")
* @param command_len Length of command string
* @return ESP_OK on success, error code on failure
* @note This function parses and executes the command, then publishes response via MQTT
*/
esp_err_t sensing_command_process(const char *command, int command_len);
/**
* @brief Initialize sensing command handler
* @return ESP_OK on success, error code on failure
* @note Must be called after MQTT is initialized
*/
esp_err_t sensing_command_init(void);
/**
* @brief Set sensor handle for command handler
* @param handle ADXL355 sensor handle
* @return ESP_OK on success, error code on failure
* @note Must be called before processing commands
*/
esp_err_t sensing_command_set_sensor_handle(adxl355_handle_t *handle);
#ifdef __cplusplus
}
#endif
sensing_command.c¶
/**
* @file sensing_command.c
* @author SHUAIWEN CUI (SHUAIWEN001@e.ntu.edu.sg)
* @brief MQTT remote control command handler implementation
* @version 1.0
* @date 2025-12-17
* @copyright Copyright (c) 2025
*
*/
#include "sensing_command.h"
#include "online_sensing.h"
#include "offline_sensing.h"
#include "node_mqtt.h"
#include "node_acc_adxl355.h"
#include "esp_log.h"
#include "esp_heap_caps.h"
#include "freertos/FreeRTOS.h"
#include "freertos/task.h"
#include "freertos/queue.h"
#include <string.h>
#include <stdio.h>
#include <stdlib.h>
#include <time.h>
/* ============================================================================
* PRIVATE DEFINITIONS
* ============================================================================ */
static const char *TAG = "SensingCmd";
/* ============================================================================
* PRIVATE VARIABLES
* ============================================================================ */
static adxl355_handle_t *s_sensor_handle = NULL;
static TaskHandle_t s_offline_sensing_task_handle = NULL;
static TaskHandle_t s_online_stop_task_handle = NULL;
static TaskHandle_t s_command_task_handle = NULL;
static QueueHandle_t s_command_queue = NULL;
/* ============================================================================
* COMMAND QUEUE STRUCTURE
* ============================================================================ */
#define COMMAND_QUEUE_SIZE 5
#define MAX_COMMAND_LENGTH 256
typedef struct
{
char command[MAX_COMMAND_LENGTH];
int command_len;
} command_queue_item_t;
/* ============================================================================
* TASK PARAMETER STRUCTURES
* ============================================================================ */
/**
* @brief Parameters for offline sensing task with delay
*/
typedef struct
{
offline_sensing_config_t config;
int delay_ms;
} offline_sensing_task_params_t;
/* ============================================================================
* PRIVATE FUNCTION DECLARATIONS
* ============================================================================ */
static esp_err_t parse_float_param(const char *str, const char *prefix, float *value);
static esp_err_t parse_int_param(const char *str, const char *prefix, int *value);
static esp_err_t parse_time_param(const char *str, const char *prefix, time_t *timestamp);
static esp_err_t send_mqtt_response(const char *response);
static void offline_sensing_task_wrapper(void *pvParameters);
static void online_sensing_stop_task(void *pvParameters);
static void command_processing_task(void *pvParameters);
static esp_err_t handle_online_command(const char *cmd);
static esp_err_t handle_offline_command(const char *cmd);
/* ============================================================================
* PRIVATE FUNCTION IMPLEMENTATIONS
* ============================================================================ */
/**
* @brief Parse float parameter from command string
*/
static esp_err_t parse_float_param(const char *str, const char *prefix, float *value)
{
char search_str[32];
snprintf(search_str, sizeof(search_str), "%s=", prefix);
const char *pos = strstr(str, search_str);
if (pos == NULL)
{
return ESP_ERR_NOT_FOUND;
}
pos += strlen(search_str);
// Find the end of the value (comma or end of string)
const char *end = strchr(pos, ',');
if (end == NULL)
{
end = pos + strlen(pos);
}
// Extract value string
char value_str[64];
size_t len = end - pos;
if (len >= sizeof(value_str))
{
len = sizeof(value_str) - 1;
}
strncpy(value_str, pos, len);
value_str[len] = '\0';
// Parse float
char *parse_end;
*value = strtof(value_str, &parse_end);
// Check if parsing was successful
if (parse_end == value_str || *parse_end != '\0')
{
return ESP_ERR_INVALID_ARG;
}
return ESP_OK;
}
/**
* @brief Parse int parameter from command string
*/
static esp_err_t parse_int_param(const char *str, const char *prefix, int *value)
{
char search_str[32];
snprintf(search_str, sizeof(search_str), "%s=", prefix);
const char *pos = strstr(str, search_str);
if (pos == NULL)
{
return ESP_ERR_NOT_FOUND;
}
pos += strlen(search_str);
// Find the end of the value (comma or end of string)
const char *end = strchr(pos, ',');
if (end == NULL)
{
end = pos + strlen(pos);
}
// Extract value string
char value_str[64];
size_t len = end - pos;
if (len >= sizeof(value_str))
{
len = sizeof(value_str) - 1;
}
strncpy(value_str, pos, len);
value_str[len] = '\0';
// Parse int
char *parse_end;
*value = (int)strtol(value_str, &parse_end, 10);
// Check if parsing was successful
if (parse_end == value_str || *parse_end != '\0')
{
return ESP_ERR_INVALID_ARG;
}
return ESP_OK;
}
/**
* @brief Parse TIME parameter (YYMMDDHHMMSS format) and convert to Unix timestamp
*/
static esp_err_t parse_time_param(const char *str, const char *prefix, time_t *timestamp)
{
char search_str[32];
snprintf(search_str, sizeof(search_str), "%s=", prefix);
const char *pos = strstr(str, search_str);
if (pos == NULL)
{
return ESP_ERR_NOT_FOUND;
}
pos += strlen(search_str);
// Find the end of the value (comma or end of string)
const char *end = strchr(pos, ',');
if (end == NULL)
{
end = pos + strlen(pos);
}
// Extract value string (should be exactly 12 digits)
size_t len = end - pos;
if (len != 12)
{
return ESP_ERR_INVALID_ARG;
}
char time_str[13];
strncpy(time_str, pos, 12);
time_str[12] = '\0';
// Parse components
int yy, mm, dd, hh, min, ss;
if (sscanf(time_str, "%2d%2d%2d%2d%2d%2d", &yy, &mm, &dd, &hh, &min, &ss) != 6)
{
return ESP_ERR_INVALID_ARG;
}
// Validate ranges
if (mm < 1 || mm > 12 || dd < 1 || dd > 31 || hh > 23 || min > 59 || ss > 59)
{
return ESP_ERR_INVALID_ARG;
}
// Convert to full year (YY: 00-99 -> 2000-2099)
int year = 2000 + yy;
// Create tm structure
struct tm timeinfo = {0};
timeinfo.tm_year = year - 1900;
timeinfo.tm_mon = mm - 1;
timeinfo.tm_mday = dd;
timeinfo.tm_hour = hh;
timeinfo.tm_min = min;
timeinfo.tm_sec = ss;
timeinfo.tm_isdst = -1; // Let system determine DST
// Convert to Unix timestamp
*timestamp = mktime(&timeinfo);
if (*timestamp == -1)
{
return ESP_ERR_INVALID_ARG;
}
return ESP_OK;
}
/**
* @brief Send MQTT response message
* @note Optimized to minimize stack usage - uses minimal logging
*/
static esp_err_t send_mqtt_response(const char *response)
{
if (!s_is_mqtt_connected || s_mqtt_client == NULL)
{
return ESP_ERR_INVALID_STATE;
}
// Publish to the same topic (or a response topic)
// esp_mqtt_client_publish is non-blocking and should not need much stack
int ret = esp_mqtt_client_publish(s_mqtt_client, MQTT_PUBLISH_TOPIC,
response, strlen(response), 1, 0);
if (ret < 0)
{
return ESP_FAIL;
}
return ESP_OK;
}
/**
* @brief Task wrapper for offline sensing (to avoid blocking)
* @note Handles delay before starting sensing
*/
static void offline_sensing_task_wrapper(void *pvParameters)
{
offline_sensing_task_params_t *params = (offline_sensing_task_params_t *)pvParameters;
// Wait for delay if specified (moved here to avoid blocking command handler)
if (params->delay_ms > 0)
{
int delay_sec = params->delay_ms / 1000;
int remaining_ms = params->delay_ms % 1000;
ESP_LOGI(TAG, "Countdown: %d sec (start in %d sec)", delay_sec, delay_sec);
// Countdown display with dynamic interval
int remaining_sec = delay_sec;
while (remaining_sec > 0)
{
// Determine interval based on remaining time (not initial delay)
int interval_sec = 1; // Default: 1 second
if (remaining_sec > 300) // > 5 minutes: show every 30 seconds
{
interval_sec = 30;
}
else if (remaining_sec > 60) // > 1 minute: show every 10 seconds
{
interval_sec = 10;
}
else if (remaining_sec > 10) // > 10 seconds: show every 5 seconds
{
interval_sec = 5;
}
else // <= 10 seconds: show every 1 second
{
interval_sec = 1;
}
int wait_sec = (remaining_sec >= interval_sec) ? interval_sec : remaining_sec;
vTaskDelay(pdMS_TO_TICKS(wait_sec * 1000));
remaining_sec -= wait_sec;
if (remaining_sec > 0)
{
ESP_LOGI(TAG, "Countdown: %d sec", remaining_sec);
}
}
// Wait for remaining milliseconds
if (remaining_ms > 0)
{
vTaskDelay(pdMS_TO_TICKS(remaining_ms));
}
ESP_LOGI(TAG, "Countdown: 0 sec - Starting now");
}
ESP_LOGI(TAG, "OFFLINE: F=%.2f Hz, D=%.2f sec",
params->config.sampling_frequency_hz, params->config.sampling_duration_sec);
// Initialize offline sensing
esp_err_t ret = offline_sensing_init(¶ms->config);
if (ret != ESP_OK)
{
char resp[128];
snprintf(resp, sizeof(resp), "SENSE,ERROR,OFFLINE,INIT_FAILED");
send_mqtt_response(resp);
free(params);
s_offline_sensing_task_handle = NULL;
vTaskDelete(NULL);
return;
}
// Set sensor handle
ret = offline_sensing_set_sensor_handle(s_sensor_handle);
if (ret != ESP_OK)
{
char resp[128];
snprintf(resp, sizeof(resp), "SENSE,ERROR,OFFLINE,SET_HANDLE_FAILED");
send_mqtt_response(resp);
offline_sensing_deinit();
free(params);
s_offline_sensing_task_handle = NULL;
vTaskDelete(NULL);
return;
}
// Start offline sensing (this will block until complete)
ret = offline_sensing_start();
// Get report
offline_sensing_report_t report;
if (ret == ESP_OK)
{
offline_sensing_get_report(&report);
char resp[256];
// Use configured values (not actual measured values) for FREQ and DUR
snprintf(resp, sizeof(resp),
"SENSE,OK,OFFLINE,SAMPLES=%lu,FREQ=%.2f,DUR=%.2f,SD=%s",
(unsigned long)report.total_samples,
params->config.sampling_frequency_hz, // Use configured frequency
params->config.sampling_duration_sec, // Use configured duration
report.sd_storage_success ? "OK" : "FAIL");
send_mqtt_response(resp);
}
else
{
char resp[128];
snprintf(resp, sizeof(resp), "SENSE,ERROR,OFFLINE,START_FAILED");
send_mqtt_response(resp);
}
// Cleanup
offline_sensing_deinit();
free(params);
s_offline_sensing_task_handle = NULL;
vTaskDelete(NULL);
}
/**
* @brief Task to stop online sensing after specified duration
* @note Optimized to minimize stack usage
*/
static void online_sensing_stop_task(void *pvParameters)
{
float duration_sec = *(float *)pvParameters;
free(pvParameters);
// Wait for the specified duration
vTaskDelay(pdMS_TO_TICKS((uint32_t)(duration_sec * 1000)));
// Stop online sensing (simple operation, should not need much stack)
esp_err_t ret = online_sensing_stop();
// Send response using static buffer to minimize stack usage
static char resp[128];
if (ret == ESP_OK)
{
snprintf(resp, sizeof(resp), "SENSE,OK,ONLINE,AUTO_STOPPED");
}
else
{
snprintf(resp, sizeof(resp), "SENSE,ERROR,ONLINE,AUTO_STOP_FAILED");
}
send_mqtt_response(resp);
s_online_stop_task_handle = NULL;
vTaskDelete(NULL);
}
/**
* @brief Handle ONLINE sensing commands
*/
static esp_err_t handle_online_command(const char *cmd)
{
// Check for STOP command
if (strstr(cmd, "STOP") != NULL)
{
// Cancel auto-stop task if running
if (s_online_stop_task_handle != NULL)
{
vTaskDelete(s_online_stop_task_handle);
s_online_stop_task_handle = NULL;
}
esp_err_t ret = online_sensing_stop();
if (ret == ESP_OK)
{
send_mqtt_response("SENSE,OK,ONLINE,STOPPED");
}
else
{
send_mqtt_response("SENSE,ERROR,ONLINE,STOP_FAILED");
}
return ret;
}
// Check for STATUS command
if (strstr(cmd, "STATUS") != NULL)
{
bool is_running = false;
online_sensing_is_running(&is_running);
online_sensing_config_t config;
online_sensing_get_config(&config);
char resp[128];
snprintf(resp, sizeof(resp), "SENSE,STATUS,ONLINE,RUNNING=%s,F=%.2f",
is_running ? "YES" : "NO", config.sampling_frequency_hz);
send_mqtt_response(resp);
return ESP_OK;
}
// Parse F and D parameters
float freq = 0.0f;
float duration = 0.0f;
if (parse_float_param(cmd, "F", &freq) != ESP_OK)
{
send_mqtt_response("SENSE,ERROR,ONLINE,INVALID_FREQ");
return ESP_ERR_INVALID_ARG;
}
if (parse_float_param(cmd, "D", &duration) != ESP_OK)
{
duration = 0.0f; // Default: continuous
}
// Validate frequency
if (freq <= 0.0f || freq > 300.0f)
{
send_mqtt_response("SENSE,ERROR,ONLINE,INVALID_FREQ");
return ESP_ERR_INVALID_ARG;
}
// Stop if already running
bool is_running = false;
online_sensing_is_running(&is_running);
if (is_running)
{
online_sensing_stop();
vTaskDelay(pdMS_TO_TICKS(100)); // Give it time to stop
}
// Update frequency
online_sensing_set_frequency(freq);
// Start online sensing
esp_err_t ret = online_sensing_start();
if (ret == ESP_OK)
{
char resp[128];
snprintf(resp, sizeof(resp), "SENSE,OK,ONLINE,F=%.2f,D=%.2f", freq, duration);
send_mqtt_response(resp);
// If duration is specified and > 0, schedule auto-stop
if (duration > 0.0f)
{
// Cancel existing stop task if any
if (s_online_stop_task_handle != NULL)
{
vTaskDelete(s_online_stop_task_handle);
s_online_stop_task_handle = NULL;
}
// Allocate memory for duration parameter
float *duration_ptr = malloc(sizeof(float));
if (duration_ptr != NULL)
{
*duration_ptr = duration;
// Create task to stop after duration
// Note: online_sensing_stop() is simple (just stops timer), but MQTT publish may need some stack
// Using 3072 bytes should be sufficient (reduced from 6144 after optimization)
BaseType_t task_ret = xTaskCreate(
online_sensing_stop_task,
"online_stop",
3072, // Stack size (3KB should be enough after optimization)
duration_ptr,
5, // Priority
&s_online_stop_task_handle);
if (task_ret != pdPASS)
{
free(duration_ptr);
ESP_LOGW(TAG, "Failed to create auto-stop task, online sensing will run continuously");
}
else
{
ESP_LOGI(TAG, "Online sensing will auto-stop after %.2f seconds", duration);
}
}
else
{
ESP_LOGW(TAG, "Failed to allocate memory for auto-stop task");
}
}
}
else
{
char resp[128];
snprintf(resp, sizeof(resp), "SENSE,ERROR,ONLINE,START_FAILED");
send_mqtt_response(resp);
}
return ret;
}
/**
* @brief Handle OFFLINE sensing commands
*/
static esp_err_t handle_offline_command(const char *cmd)
{
// Check for STOP command
if (strstr(cmd, "STOP") != NULL)
{
bool is_running = false;
offline_sensing_is_running(&is_running);
if (is_running)
{
esp_err_t ret = offline_sensing_stop();
if (ret == ESP_OK)
{
send_mqtt_response("SENSE,OK,OFFLINE,STOPPED");
}
else
{
send_mqtt_response("SENSE,ERROR,OFFLINE,STOP_FAILED");
}
return ret;
}
else
{
send_mqtt_response("SENSE,ERROR,OFFLINE,NOT_RUNNING");
return ESP_ERR_INVALID_STATE;
}
}
// Check for STATUS command
if (strstr(cmd, "STATUS") != NULL)
{
bool is_running = false;
offline_sensing_is_running(&is_running);
char resp[128];
snprintf(resp, sizeof(resp), "SENSE,STATUS,OFFLINE,RUNNING=%s",
is_running ? "YES" : "NO");
send_mqtt_response(resp);
return ESP_OK;
}
// Check if already running
bool is_running = false;
offline_sensing_is_running(&is_running);
if (is_running)
{
send_mqtt_response("SENSE,ERROR,OFFLINE,ALREADY_RUNNING");
return ESP_ERR_INVALID_STATE;
}
// Parse parameters
float freq = 0.0f;
float duration = 0.0f;
int delay_sec = 0;
time_t target_time = 0;
bool has_delay = false;
bool has_time = false;
if (parse_float_param(cmd, "F", &freq) != ESP_OK)
{
send_mqtt_response("SENSE,ERROR,OFFLINE,INVALID_FREQ");
return ESP_ERR_INVALID_ARG;
}
if (parse_float_param(cmd, "D", &duration) != ESP_OK)
{
send_mqtt_response("SENSE,ERROR,OFFLINE,INVALID_DURATION");
return ESP_ERR_INVALID_ARG;
}
// Validate parameters (must match offline_sensing.c limits: 1-4000 Hz)
// Note: Maximum frequency of 4000 Hz is limited by ADXL355 sensor's maximum output data rate (ODR)
if (freq <= 0.0f || freq > 4000.0f)
{
send_mqtt_response("SENSE,ERROR,OFFLINE,INVALID_FREQ");
return ESP_ERR_INVALID_ARG;
}
if (duration <= 0.0f)
{
send_mqtt_response("SENSE,ERROR,OFFLINE,INVALID_DURATION");
return ESP_ERR_INVALID_ARG;
}
// Check for delay or time parameter
if (parse_int_param(cmd, "DL", &delay_sec) == ESP_OK)
{
has_delay = true;
if (delay_sec < 0)
{
send_mqtt_response("SENSE,ERROR,OFFLINE,INVALID_DELAY");
return ESP_ERR_INVALID_ARG;
}
}
else if (parse_time_param(cmd, "TIME", &target_time) == ESP_OK)
{
has_time = true;
time_t now = time(NULL);
if (target_time <= now)
{
send_mqtt_response("SENSE,ERROR,OFFLINE,TIME_PAST");
return ESP_ERR_INVALID_ARG;
}
}
// Calculate actual delay
int actual_delay_ms = 0;
if (has_time)
{
time_t now = time(NULL);
actual_delay_ms = (int)((target_time - now) * 1000);
if (actual_delay_ms < 0)
{
send_mqtt_response("SENSE,ERROR,OFFLINE,TIME_PAST");
return ESP_ERR_INVALID_ARG;
}
}
else if (has_delay)
{
actual_delay_ms = delay_sec * 1000;
}
// Create configuration
offline_sensing_config_t config = {
.sampling_frequency_hz = freq,
.sampling_duration_sec = duration,
.enable_memory = true,
.enable_sd = true,
.enable_mqtt_report = true,
.sd_file_path = NULL,
.mqtt_report_topic = NULL
};
// Pre-check memory availability before creating task
// This allows immediate feedback if memory is insufficient
if (config.enable_memory)
{
// Calculate required buffer size (same as in offline_sensing_init)
uint32_t buffer_size = (uint32_t)(freq * duration) + 100; // Add margin
size_t required_bytes = buffer_size * sizeof(offline_sensing_sample_t); // Each sample is ~24 bytes
// Check PSRAM availability (we use PSRAM for large buffers)
size_t psram_free = heap_caps_get_free_size(MALLOC_CAP_SPIRAM);
if (psram_free < required_bytes)
{
char resp[256];
snprintf(resp, sizeof(resp),
"SENSE,ERROR,OFFLINE,INSUFFICIENT_MEMORY,NEED=%zu,AVAILABLE=%zu",
required_bytes, psram_free);
send_mqtt_response(resp);
ESP_LOGE(TAG, "Insufficient PSRAM: need %zu bytes, available %zu bytes",
required_bytes, psram_free);
return ESP_ERR_NO_MEM;
}
ESP_LOGI(TAG, "Memory check passed: need %zu bytes, available %zu bytes (PSRAM)",
required_bytes, psram_free);
}
// Allocate parameters for task (includes config and delay)
offline_sensing_task_params_t *task_params = malloc(sizeof(offline_sensing_task_params_t));
if (task_params == NULL)
{
send_mqtt_response("SENSE,ERROR,OFFLINE,MEMORY_FAILED");
return ESP_ERR_NO_MEM;
}
memcpy(&task_params->config, &config, sizeof(offline_sensing_config_t));
task_params->delay_ms = actual_delay_ms;
// Create task for offline sensing (non-blocking, delay handled in task)
BaseType_t ret = xTaskCreate(
offline_sensing_task_wrapper,
"offline_sensing_cmd",
8192, // Stack size
task_params,
5, // Priority
&s_offline_sensing_task_handle);
if (ret != pdPASS)
{
free(task_params);
send_mqtt_response("SENSE,ERROR,OFFLINE,TASK_CREATE_FAILED");
return ESP_FAIL;
}
// Send acknowledgment (delay is handled in task, so we don't wait here)
char resp[128];
if (has_time)
{
snprintf(resp, sizeof(resp), "SENSE,OK,OFFLINE,F=%.2f,D=%.2f,TIME=%lld,DELAY=%d",
freq, duration, (long long)target_time, actual_delay_ms / 1000);
}
else if (has_delay)
{
snprintf(resp, sizeof(resp), "SENSE,OK,OFFLINE,F=%.2f,D=%.2f,DL=%d",
freq, duration, delay_sec);
}
else
{
snprintf(resp, sizeof(resp), "SENSE,OK,OFFLINE,F=%.2f,D=%.2f",
freq, duration);
}
send_mqtt_response(resp);
return ESP_OK;
}
/* ============================================================================
* PUBLIC FUNCTION IMPLEMENTATIONS
* ============================================================================ */
esp_err_t sensing_command_init(void)
{
// Create command queue
s_command_queue = xQueueCreate(COMMAND_QUEUE_SIZE, sizeof(command_queue_item_t));
if (s_command_queue == NULL)
{
ESP_LOGE(TAG, "Failed to create command queue");
return ESP_ERR_NO_MEM;
}
// Create command processing task
BaseType_t ret = xTaskCreate(
command_processing_task,
"cmd_processor",
4096, // Stack size (increased for command processing)
NULL,
5, // Priority
&s_command_task_handle);
if (ret != pdPASS)
{
ESP_LOGE(TAG, "Failed to create command processing task");
vQueueDelete(s_command_queue);
s_command_queue = NULL;
return ESP_FAIL;
}
ESP_LOGI(TAG, "Sensing command handler initialized");
return ESP_OK;
}
/**
* @brief Command processing task (runs in separate task with larger stack)
*/
static void command_processing_task(void *pvParameters)
{
command_queue_item_t item;
ESP_LOGI(TAG, "Command processing task started");
while (1)
{
// Wait for command from queue
if (xQueueReceive(s_command_queue, &item, portMAX_DELAY) == pdTRUE)
{
// Check sensor handle first
if (s_sensor_handle == NULL)
{
send_mqtt_response("SENSE,ERROR,SENSOR_NOT_INITIALIZED");
ESP_LOGE(TAG, "Sensor handle not set, cannot process command");
continue;
}
// Trim whitespace
int len = item.command_len;
while (len > 0 && (item.command[len-1] == ' ' || item.command[len-1] == '\t' ||
item.command[len-1] == '\r' || item.command[len-1] == '\n'))
{
item.command[--len] = '\0';
}
// Check for empty command
if (len == 0)
{
send_mqtt_response("SENSE,ERROR,EMPTY_COMMAND");
continue;
}
ESP_LOGD(TAG, "Processing command: %s", item.command);
// Check if it's a SENSE command
if (strncmp(item.command, "SENSE,", 6) != 0)
{
// Not a sensing command, ignore
continue;
}
const char *cmd_part = item.command + 6; // Skip "SENSE,"
// Check for empty command after SENSE,
if (strlen(cmd_part) == 0)
{
send_mqtt_response("SENSE,ERROR,INVALID_COMMAND");
continue;
}
// Check for ONLINE or OFFLINE
if (strncmp(cmd_part, "ONLINE,", 7) == 0)
{
handle_online_command(cmd_part + 7);
}
else if (strncmp(cmd_part, "OFFLINE,", 8) == 0)
{
handle_offline_command(cmd_part + 8);
}
else
{
send_mqtt_response("SENSE,ERROR,UNKNOWN_COMMAND");
}
}
}
}
/**
* @brief Process MQTT sensing command (queues command for processing)
*/
esp_err_t sensing_command_process(const char *command, int command_len)
{
if (command == NULL || command_len <= 0)
{
return ESP_ERR_INVALID_ARG;
}
if (s_command_queue == NULL)
{
ESP_LOGE(TAG, "Command queue not initialized");
return ESP_ERR_INVALID_STATE;
}
// Prepare queue item
command_queue_item_t item;
int len = command_len < MAX_COMMAND_LENGTH - 1 ? command_len : MAX_COMMAND_LENGTH - 1;
strncpy(item.command, command, len);
item.command[len] = '\0';
item.command_len = len;
// Send to queue (non-blocking, drop if queue is full)
if (xQueueSend(s_command_queue, &item, 0) != pdTRUE)
{
ESP_LOGW(TAG, "Command queue full, dropping command");
return ESP_ERR_NO_MEM;
}
return ESP_OK;
}
/**
* @brief Set sensor handle (must be called before processing commands)
*/
esp_err_t sensing_command_set_sensor_handle(adxl355_handle_t *handle)
{
if (handle == NULL)
{
return ESP_ERR_INVALID_ARG;
}
s_sensor_handle = handle;
ESP_LOGI(TAG, "Sensor handle set");
return ESP_OK;
}
关键实现¶
关键函数¶
- 命令处理
sensing_command_process(): 命令处理的主入口点command_processing_task(): 从队列处理命令的FreeRTOS任务handle_online_command(): 处理在线传感命令-
handle_offline_command(): 处理离线传感命令 -
参数解析
parse_float_param(): 从命令字符串中提取浮点参数parse_int_param(): 从命令字符串中提取整数参数-
parse_time_param(): 解析YYMMDDHHMMSS格式的TIME参数 -
任务管理
offline_sensing_task_wrapper(): 支持延迟的离线传感包装任务-
online_sensing_stop_task(): 在持续时间后自动停止在线传感的任务 -
响应处理
send_mqtt_response(): 通过MQTT发布命令响应
命令队列结构¶
#define COMMAND_QUEUE_SIZE 5
#define MAX_COMMAND_LENGTH 256
typedef struct
{
char command[MAX_COMMAND_LENGTH];
int command_len;
} command_queue_item_t;
初始化流程¶
esp_err_t sensing_command_init(void)
{
// 创建命令队列
s_command_queue = xQueueCreate(COMMAND_QUEUE_SIZE, sizeof(command_queue_item_t));
// 创建命令处理任务
BaseType_t ret = xTaskCreate(
command_processing_task,
"cmd_processor",
4096, // 栈大小
NULL,
5, // 优先级
&s_command_task_handle);
return ret == pdPASS ? ESP_OK : ESP_FAIL;
}
命令处理流程¶
- 通过MQTT回调接收命令
- 命令排队等待处理
- 命令处理任务出队并验证
- 命令解析并路由到相应的处理器
- 通过MQTT发布响应
参数解析示例¶
static esp_err_t parse_float_param(const char *str, const char *prefix, float *value)
{
char search_str[32];
snprintf(search_str, sizeof(search_str), "%s=", prefix);
const char *pos = strstr(str, search_str);
if (pos == NULL)
{
return ESP_ERR_NOT_FOUND;
}
pos += strlen(search_str);
const char *end = strchr(pos, ',');
if (end == NULL)
{
end = pos + strlen(pos);
}
char value_str[64];
size_t len = end - pos;
if (len >= sizeof(value_str))
{
len = sizeof(value_str) - 1;
}
strncpy(value_str, pos, len);
value_str[len] = '\0';
char *parse_end;
*value = strtof(value_str, &parse_end);
if (parse_end == value_str || *parse_end != '\0')
{
return ESP_ERR_INVALID_ARG;
}
return ESP_OK;
}
在线命令处理器¶
static esp_err_t handle_online_command(const char *cmd)
{
// 检查STOP命令
if (strstr(cmd, "STOP") != NULL)
{
// 如果正在运行，取消自动停止任务
if (s_online_stop_task_handle != NULL)
{
vTaskDelete(s_online_stop_task_handle);
s_online_stop_task_handle = NULL;
}
esp_err_t ret = online_sensing_stop();
if (ret == ESP_OK)
{
send_mqtt_response("SENSE,OK,ONLINE,STOPPED");
}
else
{
send_mqtt_response("SENSE,ERROR,ONLINE,STOP_FAILED");
}
return ret;
}
// 检查STATUS命令
if (strstr(cmd, "STATUS") != NULL)
{
bool is_running = false;
online_sensing_is_running(&is_running);
online_sensing_config_t config;
online_sensing_get_config(&config);
char resp[128];
snprintf(resp, sizeof(resp), "SENSE,STATUS,ONLINE,RUNNING=%s,F=%.2f",
is_running ? "YES" : "NO", config.sampling_frequency_hz);
send_mqtt_response(resp);
return ESP_OK;
}
// 解析F和D参数
float freq = 0.0f;
float duration = 0.0f;
if (parse_float_param(cmd, "F", &freq) != ESP_OK)
{
send_mqtt_response("SENSE,ERROR,ONLINE,INVALID_FREQ");
return ESP_ERR_INVALID_ARG;
}
if (parse_float_param(cmd, "D", &duration) != ESP_OK)
{
duration = 0.0f; // 默认：持续运行
}
// 验证频率
if (freq <= 0.0f || freq > 300.0f)
{
send_mqtt_response("SENSE,ERROR,ONLINE,INVALID_FREQ");
return ESP_ERR_INVALID_ARG;
}
// 如果已在运行，先停止
bool is_running = false;
online_sensing_is_running(&is_running);
if (is_running)
{
online_sensing_stop();
vTaskDelay(pdMS_TO_TICKS(100)); // 给停止操作一些时间
}
// 更新频率并启动在线传感
online_sensing_set_frequency(freq);
esp_err_t ret = online_sensing_start();
// ... (如果duration > 0，创建自动停止任务)
return ret;
}
离线命令处理器¶
static esp_err_t handle_offline_command(const char *cmd)
{
// 检查STOP命令
if (strstr(cmd, "STOP") != NULL)
{
bool is_running = false;
offline_sensing_is_running(&is_running);
if (is_running)
{
esp_err_t ret = offline_sensing_stop();
send_mqtt_response(ret == ESP_OK ?
"SENSE,OK,OFFLINE,STOPPED" :
"SENSE,ERROR,OFFLINE,STOP_FAILED");
return ret;
}
// ... (错误处理)
}
// 解析参数（F, D, DL, TIME）
// ... (实现细节)
// 创建支持延迟的离线传感任务
// ... (任务创建)
}
使用示例¶
// 初始化命令处理器
esp_err_t ret = sensing_command_init();
if (ret != ESP_OK)
{
ESP_LOGE(TAG, "初始化命令处理器失败");
return;
}
// 设置传感器句柄
ret = sensing_command_set_sensor_handle(&adxl355_handle);
if (ret != ESP_OK)
{
ESP_LOGE(TAG, "设置传感器句柄失败");
return;
}
// 在MQTT消息回调中
void mqtt_message_callback(esp_mqtt_event_handle_t event)
{
if (strncmp(event->data, "SENSE,", 6) == 0)
{
sensing_command_process(event->data, event->data_len);
}
}