#include <errno.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>

#include "can_inf.h"
#include "xieneng_can_mqtt.h"
#include "bmser_log.h"
#include "bmser_dev.h"
#include "toolkit.h"
#include "cJSON.h"
#include "bmser_utils.h"
#include "bmser_error_code.h"
#include "bmser_project_para.h"

struct mosquitto *g_can_session = NULL;
char *g_sn_str = NULL;
int32_t g_mqttConnectState = 0;

static int32_t get_bankid_from_topic(char *topic, uint16_t *bankId, uint16_t *rackId)
{
    if (!topic || 0 == strlen(topic) || !bankId || !rackId) {
        return OPRT_PARA_NULL;
    }

    int32_t ret = 0, devSn = 0;
    uint16_t bank_id = 0, rack_id = 0;
    ret = sscanf(topic, "%*[^/]/%*[^/]/%*[^/]/%08d", &devSn);
    if (1 != ret) {
        return OPRT_DATA_INVALID;
    }

    rack_id = devSn % 1000;
    bank_id = (devSn / 1000 % 100);
    dy_syslog(LOG_DEBUG, "bank_id = %u, rack_id = %u", bank_id, rack_id);
    *bankId = bank_id;
    *rackId = rack_id;

    return OPRT_OK;
}

static int publish_binary_data(struct mosquitto* session, unsigned char *data, int len, char *cmd, char *devsn)
{
    if(g_sn_str == NULL)
    {
        return -1;
    }

    char topic[MQTT_TOPIC_MAX_LEN] = {0};
    snprintf(topic, MQTT_TOPIC_MAX_LEN, "ipc/%s/bms/%s/upgrade/SetResp/%s", g_sn_str, devsn, cmd);
    bmser_mqtt_session_publish_raw_new(session, topic, data, len, MQTT_QOS_LEVEL1, false);
    return 0;
}

#ifdef WITH_BMSER_PRODUCT_TEST
static int publish_binary_data_productTest(struct mosquitto* session, unsigned char *data, int len, unsigned char port, char *cmd)
{
    if(g_sn_str == NULL)
    {
        return -1;
    }
    char topic[MQTT_TOPIC_MAX_LEN] = {0};
    snprintf(topic, MQTT_TOPIC_MAX_LEN, "dev/%s/%s/data/%s", g_sn_str, port_enum2char(port), cmd);
    bmser_mqtt_session_publish_raw(session, topic, data, len);

    return 0;
}
#endif

int publish_critical_binary_data(char *sn_str, uint32_t canObjId, unsigned char layer, \
                                 uint16_t bankId, uint16_t rackId, unsigned short start_addr, unsigned short reg_cnt, unsigned char *buf, \
                                 int len, unsigned char cmdreply)
{
    uint8_t data[BUF_LEN] = {0};
    char devsn[MAX_DEV_SN_LEN] = {0};
    xieneng_header_t *xieneng_header = (xieneng_header_t *)data;

    int ret = get_cache(canObjId, bankId, rackId, start_addr, reg_cnt, data, BUF_LEN);
    if (ret == -1)
    {
        dy_syslog(LOG_DEBUG, "get cache failed");
    }
    else
    {
        if (cmdreply == READ_RSP)
        {
            if (memcmp(buf, data, len) == 0)
            {
                return 0;
            }
        }
    }
    xieneng_header->dev_id = rackId;
    xieneng_header->layer = layer;
    // xieneng_header->dev_id = dev_id;
    xieneng_header->func = cmdreply;
    xieneng_header->start_addr_hi = (start_addr >> 8) & 0xFF;
    xieneng_header->start_addr_lo = start_addr & 0xFF;
    xieneng_header->reg_cnt_hi = (reg_cnt >> 8) & 0xFF;
    xieneng_header->reg_cnt_lo = reg_cnt & 0xFF;

    memcpy(xieneng_header->data, buf, len);
    len = bmser_calculate_crc(data, reg_cnt * 2 + sizeof(xieneng_header_t));
    snprintf(devsn, sizeof(devsn), "000%02d%03d", bankId, rackId);

    publish_binary_data(g_can_session, data, len, TOPIC_BINARY_CMD, devsn);

    return 0;
}

int publish_binaryFlow_data(unsigned char *buf, int len, uint32_t canId)
{
    uint8_t data[BUF_LEN] = {0};
    char devsn[MAX_DEV_SN_LEN] = {0};
    canrbus_header_t *xieneng_header = (canrbus_header_t *)data;

    can_canrbusMsgFlowHeader(canId,xieneng_header);
    memcpy(xieneng_header->data, buf, len);
    len = bmser_calculate_crc(data, len + sizeof(canrbus_header_t));
    // dy_syslog(LOG_DEBUG, "publish_binary_data---->");
    snprintf(devsn,sizeof(devsn),"00000%03d",xieneng_header->msgFlowBcuId);
    //ipc/[GW SN]/bms/[Dev_Sn]/upgrade/SetResp/binaryFlow
    publish_binary_data(g_can_session, data, len, TOPIC_BINARY_FLOW, devsn);
    return 0;
}

#ifdef WITH_BMSER_PRODUCT_TEST
int publish_cmd_binary_data_7ff(uint32_t canObjId, unsigned char *buf, int len, unsigned char cmdreply)
{
    char data[BUF_LEN] = {0};

    xieneng_header_t *xieneng_header = (xieneng_header_t *)data;
    xieneng_header->func = cmdreply;
    memcpy(xieneng_header->data, buf, len);
    len = bmser_calculate_crc(data, len + sizeof(xieneng_header_t));
    dy_syslog(LOG_DEBUG, "publish_7ff_binary_data---->");
    publish_binary_data_productTest(g_can_session, data, len, canObjId, BINARY_CMD_TOPIC);
    return 0;
}
#endif

int publish_binaryCmd_func_read(unsigned char *buf, int reg_cnt, unsigned char dev_id)
{
    char devsn[MAX_DEV_SN_LEN] = {0};
    int len = bmser_calculate_crc((uint8_t *)buf, reg_cnt * 2 + sizeof(xieneng_header_t));
    snprintf(devsn,sizeof(devsn),"00000%03d", dev_id);
    publish_binary_data(g_can_session, (uint8_t *)buf, len, TOPIC_BINARY_CMD, devsn);
    return 0;
}

int publish_binaryFlow_Data2TD(unsigned char *data, int len, uint16_t bankId, uint16_t rackId)
{
    if(g_sn_str == NULL)
    {
        return -1;
    }

    char topic[MQTT_TOPIC_MAX_LEN] = {0};
    char devsn[MAX_DEV_SN_LEN] = {0};
    snprintf(devsn,sizeof(devsn),"000%02d%03d", bankId, rackId);
    snprintf(topic, MQTT_TOPIC_MAX_LEN, "ipc/%s/bms/%s/data/Set/%s", g_sn_str, devsn, TOPIC_BINARY_FLOW);
    bmser_mqtt_session_publish_raw_new(g_can_session, topic, data, len, MQTT_QOS_LEVEL1, false);
    return 0;
}

int publish_canrbusVer_Info(bmser_canrbus_ver_s ver_msg)
{
    if(g_sn_str == NULL)
    {
        return -1;
    }
    int mi = 0;
    char in_buf[128] = {0};
    char topic[MQTT_TOPIC_MAX_LEN] = {0};
    snprintf(in_buf, sizeof(in_buf), \
             "{\"identifier\":\"%s\",\"mi\":%d,\"sn\":\"%s\",\"mode\":{\"bau\":%d,\"bcu\":%d,\"bmu\":%d,\"bdu\":%d}}",\
             TOPIC_CANRBUS_VER, mi, g_sn_str, ver_msg.bau_ver, ver_msg.bcu_ver, ver_msg.bmu_ver, ver_msg.bdu_ver);

    snprintf(topic, MQTT_TOPIC_MAX_LEN, "ipc/%s/bms/00000000/upgrade/GetResp/%s", g_sn_str, TOPIC_CANRBUS_VER);
    bmser_mqtt_session_publish_raw_new(g_can_session, topic, in_buf, strlen(in_buf), MQTT_QOS_LEVEL1, false);
}

static void xieneng_can_subscribe_all()
{
    char topic[MQTT_TOPIC_MAX_LEN] = {0};
    int i = 0;
    char sn[SN_MAX_LEN + 1] = {0};
    bmser_dev_get_board_sn_ex(sn, sizeof(sn));

    for (i = 0; i < CAN_OBJ_ID_MAX; i++)
    {
        memset(topic, 0, MQTT_TOPIC_MAX_LEN);
        snprintf(topic, MQTT_TOPIC_MAX_LEN, "ipc/%s/bms/+/upgrade/Set/%s", sn, TOPIC_BINARY_FLOW);
        bmser_mqtt_session_subscribe(g_can_session, topic);

        memset(topic, 0, MQTT_TOPIC_MAX_LEN);
        snprintf(topic, MQTT_TOPIC_MAX_LEN, "ipc/%s/bms/+/upgrade/Set/%s", sn, TOPIC_BINARY_CMD);
        bmser_mqtt_session_subscribe(g_can_session, topic);

#ifdef WITH_BMSER_PRODUCT_TEST
        memset(topic, 0, MQTT_TOPIC_MAX_LEN);
        snprintf(topic, MQTT_TOPIC_MAX_LEN, "to_dev/+/%s/data/%s", port_enum2char(i), TOPIC_SEND_RGLT_SIGNAL_RAW_DATA_7FF);
        bmser_mqtt_session_subscribe(g_can_session, topic);
#endif
        //订阅写项目配置参数TOPIC
        memset(topic, 0, MQTT_TOPIC_MAX_LEN);
        snprintf(topic, MQTT_TOPIC_MAX_LEN, "ipc/%s/bms/+/data/Set/%s", sn, TOPIC_SET_PROJECT_PARA);
        bmser_mqtt_session_subscribe(g_can_session, topic);

        //订阅获取canrbus版本TOPIC
        memset(topic, 0, MQTT_TOPIC_MAX_LEN);
        snprintf(topic, MQTT_TOPIC_MAX_LEN, "ipc/%s/bms/+/upgrade/Get/%s", sn, TOPIC_CANRBUS_VER);
        bmser_mqtt_session_subscribe(g_can_session, topic);
    }
}


static int xieneng_can_mqtt_handle_recv_msg(void* userdata, const struct mosquitto_message* mqtt_msg)
{
    if (strstr(mqtt_msg->topic, TOPIC_BINARY_FLOW))
    {
        // dy_syslog(LOG_DEBUG, "rcv topic: %s ", mqtt_msg->topic);
        xieneng_can_send_to_dev(mqtt_msg);
    }
    else if (strstr(mqtt_msg->topic, TOPIC_BINARY_CMD))
    {
        uint16_t bankId = 0, rackId = 0, devId = 0;
        uint16_t racks = bmser_jsonGetSingleBankRackNum();
        get_bankid_from_topic(mqtt_msg->topic, &bankId, &rackId);
        devId = bankId * racks + rackId;
        xieneng_can_msg_handle_binary(devId, mqtt_msg);
    }
#ifdef WITH_BMSER_PRODUCT_TEST
    else if (strstr(mqtt_msg->topic, TOPIC_SEND_RGLT_SIGNAL_RAW_DATA_7FF))
    {
        xieneng_can_send_to_dev_7ff(mqtt_msg);
    }
#endif
    else if (strstr(mqtt_msg->topic, TOPIC_SET_PROJECT_PARA)) 
    {
        dy_syslog(LOG_DEBUG, "rcv topic: %s ", mqtt_msg->topic);
        xieneng_can_set_config_para(mqtt_msg);
    }
    else if (strstr(mqtt_msg->topic, TOPIC_CANRBUS_VER)) 
    {
        dy_syslog(LOG_DEBUG, "rcv topic: %s, payload: %s\n", mqtt_msg->topic,(char*)mqtt_msg->payload);
        bmser_canrbus_protocol_ver_publish();
    }
    return 0;
}

static int xieneng_can_state_change(int state)
{
    char topic[MQTT_TOPIC_MAX_LEN] = {0};
    dy_syslog(LOG_DEBUG, "xieneng_can_state_change = %d.", state);
    if(state == 0)
    {
        g_mqttConnectState = 1;
        xieneng_can_subscribe_all();
    }
    return 0;
}

// 建立与内部broker之间的MQTT连接
int xieneng_can_mqtt_client_init(char *sn_str)
{
    char clientId[MAX_CLIENT_ID_LEN] = {0};
    if(sn_str != NULL)
    {
        g_sn_str = sn_str;
        snprintf(clientId, MAX_CLIENT_ID_LEN, "xieneng_can_%s", sn_str);
    }
    g_can_session = bmser_mqtt_session_create(clientId, "localhost",MQTT_CONNECT_PORT, xieneng_can_mqtt_handle_recv_msg,xieneng_can_state_change);
    if (g_can_session == NULL)
    {
        return -1;
    }
    // 压力测试发现会出现连接失败的情况，所以这里做了重连
    while (!g_mqttConnectState)
    {
        sleep(1);
        mosquitto_reconnect(g_can_session);
    }
    return 0;
}