esp_alox/main/communication_handler.c

#include "esp_log.h"
#include "esp_now.h"
#include "esp_timer.h"
#include "freertos/idf_additions.h"

#include "client_handler.h"
#include "communication_handler.h"
#include <stdbool.h>
#include <stdint.h>
#include <stdlib.h>
#include <string.h>
#include <sys/types.h>

static const char *TAG = "ALOX - COM";

static struct ESP_MessageBroker mr;
static QueueHandle_t ESP_recieved_message_queue;

void ESP_InitMessageBroker(QueueHandle_t msg_queue_handle) {
  mr.num_direct_callbacks = 0;
  mr.num_task_callbacks = 0;
  ESP_recieved_message_queue = msg_queue_handle;
  return;
}

void ESP_RegisterFunction(CommandPages command,
                          ESP_RegisterFunctionCallback callback) {
  mr.FunctionList[mr.num_direct_callbacks].MSGID = command;
  mr.FunctionList[mr.num_direct_callbacks].callback = callback;
  mr.num_direct_callbacks++;
  return;
}

void ESP_RegisterTask(CommandPages command, ESP_RegisterTaskCallback callback) {
  mr.TaskList[mr.num_task_callbacks].MSGID = command;
  mr.TaskList[mr.num_task_callbacks].task = callback;
  mr.num_task_callbacks++;
}

void ESP_MessageBrokerTask(void *param) {
  ESPNOW_MessageInfo received_msg;
  ESP_MessageBrokerTaskParams_t *task_params =
      (ESP_MessageBrokerTaskParams_t *)param;

  // Extrahiere die einzelnen Parameter
  QueueHandle_t msg_queue = task_params->message_queue;

  if (msg_queue == NULL) {
    ESP_LOGE(TAG, "Message queue not initialized. Terminating task.");
    vTaskDelete(NULL);
  }

  ESP_LOGI(TAG, "Message broker task started.");

  while (1) {
    if (xQueueReceive(msg_queue, &received_msg, portMAX_DELAY)) {

      ESP_LOGI(TAG, "Broker got message trying to relay it now");
      const BaseMessage *message = (const BaseMessage *)received_msg.data;
      ESP_LOGI(TAG, "Broker searching for command page %d",
               message->commandPage);
      for (int i = 0; i < mr.num_direct_callbacks; i++) {
        if (mr.FunctionList[i].MSGID == message->commandPage) {
          mr.FunctionList[i].callback(&received_msg.esp_now_info,
                                      received_msg.data, received_msg.data_len);
          ESP_LOGI(TAG, "Broker found matching msgid %d",
                   mr.FunctionList[i].MSGID);
        }
      }
      for (int i = 0; i < mr.num_direct_callbacks; i++) {
        // if (mr.FunctionList[i].MSGID == received_msg.msgid) {
        //  TODO: Not yet implemented
        //  Only send data to task, task should be created beforhead and wait
        //  for new data in the queue.
        //}
      }
    }
  }
}

QueueHandle_t messageQueue = NULL; // Warteschlange für empfangene Nachrichten
static bool hasMaster = false;
static ClientList *esp_client_list;
static uint8_t channelNumber = 0;

#define MAC_STRING_BUFFER_SIZE 18

int init_com(ClientList *clients, uint8_t wifi_channel) {
  // Initialisiere die Kommunikations-Warteschlange
  messageQueue = xQueueCreate(MESSAGE_QUEUE_SIZE, sizeof(BaseMessage));
  if (messageQueue == NULL) {
    ESP_LOGE(TAG, "Message queue creation failed");
    return -1;
  }

  esp_client_list = clients;
  hasMaster = false;
  channelNumber = wifi_channel;
  return 0;
}

int add_peer(uint8_t *macAddr) {
  esp_now_peer_info_t peerInfo = {
      .channel = channelNumber,
      .ifidx = ESP_IF_WIFI_STA,
      .encrypt = false, // Keine Verschlüsselung (kann geändert werden)
  };
  memcpy(peerInfo.peer_addr, macAddr, ESP_NOW_ETH_ALEN);

  esp_err_t result = esp_now_add_peer(&peerInfo);
  if (result == ESP_OK) {
    ESP_LOGI(TAG, "Peer added: " MACSTR, MAC2STR(peerInfo.peer_addr));

    if (!IS_BROADCAST_ADDR(macAddr)) {
      int ret = add_client(esp_client_list, peerInfo.peer_addr);
      if (ret < 0) {
        ESP_LOGE(TAG, "Client could not be added to client handler, removing "
                      "it from esp now client list!");
        esp_now_del_peer(peerInfo.peer_addr);
        return -1;
      }
      ESP_LOGI(TAG, "New client added.");
    }
  } else if (result == ESP_ERR_ESPNOW_EXIST) {
    ESP_LOGW(TAG, "Peer already exists.");
    int id = get_client_id(esp_client_list, peerInfo.peer_addr);
    if (id >= 0) {
      ESP_LOGI(TAG, "Client found again, welcome back!");
      esp_client_list->Clients[id].isAvailable = true;
    }
  } else {
    ESP_LOGE(TAG, "Failed to add peer: %s", esp_err_to_name(result));
    return -1;
  }
  return 0;
}

BaseMessage MessageBuilder(CommandPages commandPage, PayloadUnion payload,
                           size_t payload_size) {
  BaseMessage message;

  // Initialisierung der BaseMessage
  message.commandPage = commandPage;
  message.version = 1;
  message.length = (uint16_t)payload_size;

  // Kopieren des Payloads in die Union
  memset(&message.payload, 0, sizeof(message.payload)); // Sicherheitsmaßnahme
  memcpy(&message.payload, &payload, payload_size);

  return message;
}

void master_broadcast_task(void *param) {
  while (1) {
    BroadCastPayload payload = {};

    BaseMessage message = MessageBuilder(
        BroadCastPage, *(PayloadUnion *)&payload, sizeof(payload));

    ESP_ERROR_CHECK(esp_now_send(broadcast_address, (uint8_t *)&message,
                                 sizeof(BaseMessage)));
    ESP_LOGI(TAG, "Broadcast Message sent");
    vTaskDelay(pdMS_TO_TICKS(5000));
  }
}

void master_broadcast_ping(void *param) {
  while (1) {
    PingPayload payload = {};
    payload.timestamp = esp_timer_get_time();
    BaseMessage message =
        MessageBuilder(PingPage, *(PayloadUnion *)&payload, sizeof(payload));
    ESP_ERROR_CHECK(esp_now_send(broadcast_address, (uint8_t *)&message,
                                 sizeof(BaseMessage)));
    ESP_LOGI(TAG, "Broadcast PING Message sent");
    vTaskDelay(pdMS_TO_TICKS(2500));
  }
}

void master_ping_task(void *param) {
  while (1) {
    for (size_t i = 0; i < MAX_CLIENTS; i++) {
      if (esp_client_list->Clients[i].isAvailable) {
        ESP_LOGI(TAG, "SEND PING TO %zu: " MACSTR, i,
                 MAC2STR(esp_client_list->Clients[i].macAddr));
        PingPayload payload = {};
        payload.timestamp = esp_timer_get_time();
        BaseMessage message = MessageBuilder(
            PingPage, *(PayloadUnion *)&payload, sizeof(payload));
        esp_now_send(esp_client_list->Clients[i].macAddr, (uint8_t *)&message,
                     sizeof(BaseMessage));
        ESP_LOGI(TAG, "SENDING PING!!!!");
      }
    }
    vTaskDelay(pdMS_TO_TICKS(1000));
  }
}

void master_StatusCallback(const esp_now_recv_info_t *esp_now_info,
                           const uint8_t *data, int data_len) {
  const BaseMessage *message = (const BaseMessage *)data;

  ESP_LOGI(TAG, "SRC " MACSTR, MAC2STR(esp_now_info->src_addr));
  ESP_LOGI(TAG,
           "Status Message Received: status: %d, runningPartition: %d, uptime: "
           "%d, version: %d",
           message->payload.status_payload.status,
           message->payload.status_payload.runningPartition,
           message->payload.status_payload.uptime,
           message->payload.status_payload.version);
}

void master_RegisterCallback(const esp_now_recv_info_t *esp_now_info,
                             const uint8_t *data, int data_len) {
  BaseMessage replyMessage = {};
  const BaseMessage *message = (const BaseMessage *)data;

  ESP_LOGI(TAG, "WILL REGISTER DEVICE");
  esp_now_peer_info_t checkPeerInfo;
  esp_err_t checkPeer =
      esp_now_get_peer(esp_now_info->src_addr, &checkPeerInfo);
  switch (checkPeer) {
  case (ESP_OK):
    ESP_LOGI(TAG, "CLIENT BEKANNT");
    int id = get_client_id(esp_client_list, esp_now_info->src_addr);
    esp_client_list->Clients[id].isAvailable = true;
    esp_client_list->Clients[id].lastSuccessfullPing = xTaskGetTickCount();
    ESP_LOGI(TAG, "Updated client %d last ping time to %lu", id,
             esp_client_list->Clients[id].lastSuccessfullPing);
    break;
  case (ESP_ERR_ESPNOW_NOT_INIT):
    ESP_LOGI(TAG, "Not initalised");
    break;
  case (ESP_ERR_ESPNOW_ARG):
    ESP_LOGI(TAG, "ESP ERR ESPNOW_ARG");
    break;
  case (ESP_ERR_ESPNOW_NOT_FOUND):
    ESP_LOGI(TAG, "CLIENT WIRD IN DIE LISTE AUFGENOMMEN");
    add_peer(esp_now_info->src_addr);
    ESP_LOGI(TAG, "FRAGE CLIENT STATUS AN");

    GetStatusPayload payload = {};
    replyMessage = MessageBuilder(GetStatusPage, *(PayloadUnion *)&payload,
                                  sizeof(payload));
    ESP_ERROR_CHECK(esp_now_send(
        esp_now_info->src_addr, (uint8_t *)&replyMessage, sizeof(BaseMessage)));

    break;
  default:
    ESP_LOGI(TAG, "Unknown Message %i", checkPeer);
  }
}

void ESPNOW_RegisterMasterCallbacks() {
  ESP_RegisterFunction(StatusPage, master_StatusCallback);
  ESP_RegisterFunction(RegisterPage, master_RegisterCallback);
}

void master_receive_callback(const esp_now_recv_info_t *esp_now_info,
                             const uint8_t *data, int data_len) {
  ESP_LOGI(TAG, "MASTER GOT MESSAGE");

  // Allokiere Speicher für die Daten und kopiere sie
  uint8_t *copied_data = (uint8_t *)malloc(data_len);
  if (copied_data == NULL) {
    ESP_LOGE(TAG, "Failed to allocate memory for message data.");
    return;
  }
  memcpy(copied_data, data, data_len);

  // Fülle die neue Struktur mit kopierten Daten
  ESPNOW_MessageInfo msg_info;
  memcpy(&msg_info.esp_now_info, esp_now_info, sizeof(esp_now_recv_info_t));
  msg_info.data = copied_data;
  msg_info.data_len = data_len;

  if (xQueueSend(ESP_recieved_message_queue, &msg_info, portMAX_DELAY) !=
      pdPASS) {
    // Fehlerbehandlung: Queue voll oder Senden fehlgeschlagen
    ESP_LOGE(TAG, "Failed to send parsed message to queue.");
  }

  return;
  // TODO: just testing if callback queue logic works here

  BaseMessage replyMessage = {};
  const BaseMessage *message = (const BaseMessage *)data;
  int id;
  switch (message->commandPage) {
  case StatusPage:
    ESP_LOGI(TAG, "GOT STATUS MESSAGE");
    id = get_client_id(esp_client_list, esp_now_info->src_addr);
    if (id >= 0) {
      esp_client_list->Clients[id].clientVersion =
          message->payload.status_payload.version;
    }

    break;
  case PingPage:
    ESP_LOGI(TAG, "GOT PING MESSAGE");
    uint32_t currentTime = esp_timer_get_time();
    uint32_t diff = currentTime - message->payload.ping_payload.timestamp;

    ESP_LOGI(TAG, "Start: %lu, End: %lu, Diff: %lu, Ping: %lu",
             message->payload.ping_payload.timestamp, currentTime, diff,
             diff / 1000); // ping in ms

    id = get_client_id(esp_client_list, esp_now_info->src_addr);
    if (id >= 0) {
      esp_client_list->Clients[id].lastSuccessfullPing = xTaskGetTickCount();
      esp_client_list->Clients[id].lastPing = (diff / 1000);
      ESP_LOGI(TAG, "Updated client %d: " MACSTR " last ping time to %lu", id,
               MAC2STR(esp_now_info->src_addr),
               esp_client_list->Clients[id].lastSuccessfullPing);
    }
    break;
  case BroadCastPage:
    ESP_LOGI(TAG, "MASTER SHOULD NOT GET BROADCAST MESSAGE, is there another "
                  "master calling?");
    break;
  case RegisterPage:
    ESP_LOGI(TAG, "WILL REGISTER DEVICE");
    esp_now_peer_info_t checkPeerInfo;
    esp_err_t checkPeer =
        esp_now_get_peer(esp_now_info->src_addr, &checkPeerInfo);
    switch (checkPeer) {
    case (ESP_OK):
      ESP_LOGI(TAG, "CLIENT BEKANNT");
      int id = get_client_id(esp_client_list, esp_now_info->src_addr);
      esp_client_list->Clients[id].isAvailable = true;
      esp_client_list->Clients[id].lastSuccessfullPing = xTaskGetTickCount();
      ESP_LOGI(TAG, "Updated client %d last ping time to %lu", id,
               esp_client_list->Clients[id].lastSuccessfullPing);
      break;
    case (ESP_ERR_ESPNOW_NOT_INIT):
      ESP_LOGI(TAG, "Not initalised");
      break;
    case (ESP_ERR_ESPNOW_ARG):
      ESP_LOGI(TAG, "ESP ERR ESPNOW_ARG");
      break;
    case (ESP_ERR_ESPNOW_NOT_FOUND):
      ESP_LOGI(TAG, "CLIENT WIRD IN DIE LISTE AUFGENOMMEN");
      add_peer(esp_now_info->src_addr);
      ESP_LOGI(TAG, "FRAGE CLIENT STATUS AN");

      GetStatusPayload payload = {};
      replyMessage = MessageBuilder(GetStatusPage, *(PayloadUnion *)&payload,
                                    sizeof(payload));
      ESP_ERROR_CHECK(esp_now_send(esp_now_info->src_addr,
                                   (uint8_t *)&replyMessage,
                                   sizeof(BaseMessage)));

      break;
    default:
      ESP_LOGI(TAG, "Unknown Message %i", checkPeer);
    }
    break;
  default:
    ESP_LOGI(TAG, "Unknown CommandPage %i", message->commandPage);
    break;
  }
}

void client_receive_callback(const esp_now_recv_info_t *esp_now_info,
                             const uint8_t *data, int data_len) {
  ESP_LOGI(TAG, "SLAVE GOT MESSAGE");
  ESP_LOGI(TAG, "Received message from: " MACSTR,
           MAC2STR(esp_now_info->src_addr));
  ESP_LOGI(TAG, "Message: %.*s", data_len, data);

  BaseMessage replyMessage = {};

  const BaseMessage *message = (const BaseMessage *)data;
  switch (message->commandPage) {
  case StatusPage:
    ESP_LOGI(TAG, "GOT STATUS MESSAGE");
    break;
  case GetStatusPage: {
    StatusPayload payload = {
        .status = 1,
        .runningPartition = 1,
        .uptime = 100,
        .version = 0x0002,

    };
    replyMessage =
        MessageBuilder(StatusPage, *(PayloadUnion *)&payload, sizeof(payload));
    ESP_ERROR_CHECK(esp_now_send(
        esp_now_info->src_addr, (uint8_t *)&replyMessage, sizeof(BaseMessage)));

  } break;
  case PingPage:
    ESP_LOGI(TAG, "GOT PING MESSAGE");
    replyMessage = MessageBuilder(PingPage, *(PayloadUnion *)&message->payload,
                                  sizeof(message->payload));
    ESP_ERROR_CHECK(esp_now_send(
        esp_now_info->src_addr, (uint8_t *)&replyMessage, sizeof(BaseMessage)));
    break;
  case BroadCastPage:
    ESP_LOGI(TAG, "GOT BROADCAST MESSAGE");
    if (!hasMaster) {
      if (IS_BROADCAST_ADDR(esp_now_info->des_addr)) {
        ESP_LOGI(TAG,
                 "GOT BROADCAST MESSAGE ATTEMPTING TO REGISTER TO MASTER!");
        add_peer(esp_now_info->src_addr);
        replyMessage =
            MessageBuilder(RegisterPage, *(PayloadUnion *)&message->payload,
                           sizeof(message->payload));
        ESP_ERROR_CHECK(esp_now_send(esp_now_info->src_addr,
                                     (uint8_t *)&replyMessage,
                                     sizeof(BaseMessage)));
        hasMaster = true;
      }
      break;
    }
    // check if its an update message

    break;
  case RegisterPage:
    break;
  default:
    ESP_LOGI(TAG, "GOT UNKONW MESSAGE");
    break;
  }
}

void client_data_sending_task(void *param) {
  while (1) {
    const char *dataToSend = "DATA:42";
    ESP_LOGI(TAG, "SEND DATA");
    esp_now_send(NULL, (uint8_t *)dataToSend, strlen(dataToSend));
    vTaskDelay(pdMS_TO_TICKS(5000));
  }
}

void client_monitor_task(void *pvParameters) {
  TickType_t timeout_ticks = pdMS_TO_TICKS(CLIENT_TIMEOUT_MS);
  TickType_t interval_ticks = pdMS_TO_TICKS(CHECK_INTERVAL_MS);

  while (1) {
    TickType_t now = xTaskGetTickCount();

    for (int i = 0; i < MAX_CLIENTS; i++) {
      if (esp_client_list->Clients[i].isAvailable) {
        TickType_t time_diff =
            now - esp_client_list->Clients[i].lastSuccessfullPing;

        if (time_diff > timeout_ticks) {
          esp_client_list->Clients[i].isAvailable = false;
          ESP_LOGW(TAG, "Client %d (MAC: " MACSTR ") is unavailable",
                   MAC2STR(esp_client_list->Clients[i].macAddr));
        }
      }
    }
    vTaskDelay(interval_ticks);
  }
}