IntegratedCMU/app/lib/a_process/pd_csg.c

/******************************************************************************
 * file    lib/process/pd_csg.c 
 * author  YuLiang
 * version 1.0.0
 * date    27-Feb-2023
 * brief   This file provides all the csg server operation functions.
 *
 ******************************************************************************
 * Attention
 *
 * <h2><center>&copy; COPYRIGHT(c) 2021 LandPower</center></h2>
 *
 * Redistribution and use in source and binary forms, with or without modification,
 * are permitted provided that the following conditions are met:
 *   1. Redistributions of source code must retain the above copyright notice,
 *      this list of conditions and the following disclaimer.
 *   2. Redistributions in binary form must reproduce the above copyright notice,
 *      this list of conditions and the following disclaimer in the documentation
 *      and/or other materials provided with the distribution.
 *   3. Neither the name of LandPower nor the names of its contributors may be used to 
 *      endorse or promote products derived from this software without specific
 *      prior written permission.
 *
 * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
 * AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
 * IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
 * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
 * FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
 * DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
 * SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
 * CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
 * OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
 * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
 *
 ******************************************************************************/

/* Includes ------------------------------------------------------------------*/
#ifdef HAVE_CONFIG_H
#include "config.h"
#endif

#ifdef CFG_DEV_TYPE_LAND_PD
/* 标准C库头文件. */
#include <sys/types.h>
#include <sys/ipc.h>
#include <sys/msg.h>
#include <sys/stat.h>
#include <sys/socket.h>
#include <sys/time.h>
#include <fcntl.h>
#include <netinet/in.h>
#include <arpa/inet.h>
#include <pthread.h>
#include <sys/prctl.h>
#include <sys/types.h>
#include <dirent.h>
#include <ctype.h>
#include <sys/statvfs.h>

/* 用户代码头文件. */
#include "main.h"
#include "cmd.h"
#include "fifo.h"
#include "pd_main.h"
#include "pd_csg.h"
#include "pd_upgrade.h"
#include "pd_dau.h"
#include "pd_hf.h"
#include "pd_modbus.h"

/* Private define ------------------------------------------------------------*/

/* Private macro -------------------------------------------------------------*/

/* Private typedef -----------------------------------------------------------*/

/* Private variables ---------------------------------------------------------*/
csg_t csg;
static csg_event_t _csg_event;

/* Private function prototypes -----------------------------------------------*/
void _csg_server_set(int32_t ip, uint16_t port);
void _csg_show();
int32_t _csg_send_add_dau(uint8_t slot, void *data);
int32_t _csg_send_reset(uint8_t, void *data);
int32_t _csg_send_info_set(uint8_t, void *data);
int32_t _csg_send_info_get(uint8_t, void *data);
int32_t _csg_send_updata_rv(uint8_t slot, void *data);
int32_t _csg_send_heartbeat(uint8_t slot, void *data);
int32_t _csg_send_cfg_set(uint8_t, void *data);
int32_t _csg_send_cfg_get(uint8_t, void *data);
int32_t _csg_send_cfg_set_port(uint8_t, void *data);
int32_t _csg_send_cfg_get_port(uint8_t, void *data);
int32_t _csg_send_prps_get(uint8_t, void *data);
int32_t _csg_send_trend_data(uint8_t, void *data);
int32_t _csg_send_prps_data(uint8_t, void *data);
int32_t _csg_send_event_data(uint8_t, void *data);

// 命令映射表
static csg_send_fun_cb _csg_send_command[] = 
{
    NULL,                           // 0
    _csg_send_add_dau,              // DAU_SEND_ADD             1
    _csg_send_reset,                // DAU_SEND_RESET           2
    NULL,                           // 3
    _csg_send_info_set,             // DAU_SEND_INFO_SET        4
    _csg_send_info_get,             // DAU_SEND_INFO_GET        5
    _csg_send_updata_rv,            // DAU_SEND_UPDATE_RV       6
    _csg_send_heartbeat,            // DAU_SEND_HEARTBEAT       7
    _csg_send_cfg_set,              // DAU_SEND_CFG_SET         8
    _csg_send_cfg_get,              // DAU_SEND_CFG_GET         9
    _csg_send_cfg_set_port,         // DAU_SEND_PORT_SET        10
    _csg_send_cfg_get_port,         // DAU_SEND_PORT_GET        11
    _csg_send_prps_get,             // DAU_SEND_PRPS_GET        12
    _csg_send_trend_data,           // DAU_SEND_TREND           13
    _csg_send_prps_data,            // DAU_SEND_PRPS            14
    _csg_send_event_data,           // DAU_SEND_EVENT           15  
    NULL
};

/* Internal functions --------------------------------------------------------*/
/* 服务器地址设置 */
CMD(csg_server_set,
    csg_server_set_cmd,
    "csg server A.B.C.D <1-65535>",
    "Csg\n"
    "Server\n"
    "IPv4 address\n"
    "UDP port\n")
{
    _csg_server_set(inet_addr((char*)argv[0]), strtol((char*)argv[1], NULL, 10));
    
    return CMD_SUCCESS;
}

/* 显示模块状态 */
CMD(csg_show,
    csg_show_cmd,
    "show csg",
    "Show\n"
    "CSG\n")
{
    _csg_show();
    
    return CMD_SUCCESS;
}

void _print_sockaddr_in(const struct sockaddr_in *addr) 
{
    // 将IP地址从网络字节序转换为点分十进制格式
    char ip_str[INET_ADDRSTRLEN];
    inet_ntop(AF_INET, &addr->sin_addr, ip_str, sizeof(ip_str));
 
    // 将端口号从网络字节序转换为主机字节序并打印
    unsigned short port = ntohs(addr->sin_port);
 
    DBG(DBG_M_PD_CSG, "IP Address: %s, Port: %u\r\n", ip_str, port);
}

void _csg_show()
{
    printh("CSG connect: %s \n", (csg.is_connect == 1)? "OK" : "FAIL");
    for (int i = 0; i < PD_SLOTS_MAX; i++)
    {
        printh("dau[%d]:is_connect=%d is_insert=%d\n", i, daus[i].state.is_connect,
                daus[i].state.is_insert);
    }
}

void _csg_server_set(int32_t ip, uint16_t port)
{
    /* 比较配置 */
    if (csg.server_ip != ip
        || csg.server_port != port)
    {
        csg.server_ip = ip;
        csg.server_port = port;

        bzero(&csg.server, sizeof(csg.server));
        csg.server.sin_family = AF_INET;
        csg.server.sin_addr.s_addr = csg.server_ip;
        csg.server.sin_port = htons(csg.server_port);
    }
}

/* 封装数据并写fifo */
int32_t _csg_msg_send_dau(int slot, uint32_t type, void *data)
{
    pd_csg_msg_t msg;
    
    /* 封装消息. */
    msg.type = type;
    msg.data = data;

    /* 发送消息 */
    if (fifo_write(daus[slot].fifo_send, (void*)(&msg), sizeof(pd_csg_msg_t)) != sizeof(pd_csg_msg_t))
    {
        DBG(DBG_M_FIFO, "SSG write ERROR! type=%d\r\n", msg.type);
        return E_ERROR;
    }

    return E_NONE;
}


/* 校验收到包的包头, 长度, 校验码. */
int32_t _csg_pkt_check(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;

    /* 对主次设备号进行识别, 次设备号可以是广播. */
    if ((head->dev_type_m != device_info.type_m))
    {
        DBG(DBG_M_PD_CSG_ERR, "@1 type_m=%d  %d \r\n", head->dev_type_m, device_info.type_m);
        return E_ERROR;
    } 

    if (head->len > 1500)
    {
        DBG(DBG_M_PD_CSG_ERR, "@2 receive packet len(%d) is out of range\r\n", head->len);
        return E_ERROR;
    }
    
    return E_NONE;
}

/* 包头填充. */
void _csg_head_init(char *buf, uint16_t len, uint8_t cmdType, uint8_t cmd, uint8_t slot)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)buf;

    /* 封装报文头. */
    head->len = len;
    head->dev_type_m = device_info.type_m;
    head->dev_type_s= device_info.type_s;
    head->dev_id = device_info.dev_id;
    head->cmd_type = cmdType;
    head->cmd = cmd;
    head->version = 1;
    head->pkt_id = csg.pkt_index++;
    head->slot = slot;
}

/* 数据发送 */
void _csg_send_data(uint8_t cmd_type, uint8_t cmd, char *pkt, int32_t len, uint8_t slot)
{
    int32_t rv = 0;
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;

    /* 封装报文头. */
    _csg_head_init(pkt, sizeof(csg_pkt_head_t) + len, cmd_type, cmd, slot);

    rv = sendto(csg.skfd, pkt, head->len, 0, (struct sockaddr*)&csg.server, sizeof(csg.server));
    if (rv < 0)
    {
        DBG(DBG_M_PD_CSG_ERR, "Sendto return %s!\r\n", safe_strerror(errno));
    }
}

/* 与后台连接断开 */
void _csg_disconnect_set(const char *message)
{
    if (csg.is_connect) 
    {
        csg.is_connect = FALSE;
        log_warn(LOG_CSG, "[%s]CSG Connection lost!!!\n", message);
    }
}

/* 主动连接请求. */
int32_t _csg_connect_send(void)
{
    char *pkt = csg.buf_send;
    csg_contact_t *pinfo = (csg_contact_t *)(pkt + sizeof(csg_pkt_head_t));
    
    pinfo->type_m = device_info.type_m;
    pinfo->type_s = device_info.type_s;
    pinfo->dev_id = device_info.dev_id;
    strncpy(pinfo->hostname, host.name, sizeof(pinfo->hostname)-1);
    pinfo->factory_date = device_info.factory_date;
    pinfo->deployment_date = device_info.deployment_date;
    snprintf((char*)pinfo->app_version, DEV_VERSION_STR_LEN, "%s", device_info.app_version);
    snprintf((char*)pinfo->app_compile_time, DEV_VERSION_STR_LEN, "%s", device_info.app_compile_time);
    snprintf((char*)pinfo->hardware_version, DEV_VERSION_STR_LEN, "%s", device_info.hardware_version);
    snprintf((char*)pinfo->FPGA_version, DEV_VERSION_STR_LEN, "%s", device_info.FPGA_version);
    pinfo->ip = device_info.ip;
    pinfo->mask = device_info.mask;
    pinfo->gw = device_info.gw;
    memcpy(pinfo->mac, device_info.mac, sizeof(pinfo->mac));
    pinfo->server_port = csg.server_port;
    pinfo->server_ipv4 = csg.server_ip;

    memset(pinfo->port, 0, sizeof(pinfo->port));
    memset(pinfo->port_type, 0, sizeof(pinfo->port_type));
    _csg_send_data(CSG_REPLY, CSG_C_CONTACT, pkt, sizeof(csg_contact_t), 0);

    return E_NONE;
}

/* 心跳包 */
int32_t _csg_heartbeat_send(void)
{
    char *pkt = csg.buf_send;
    csg_heartbeat_t *pinfo = (csg_heartbeat_t *)(pkt + sizeof(csg_pkt_head_t));
    uint16_t i = 0;

    for(i = 0; i < PD_DAU_SUM; i++)
    {
        //if (dau_is_valid(dau[i]))
        {
            //pinfo->dau_state[i] = dau[i]->is_connect;
            //pinfo->dau_port_num[i] = dau[i]->port_num;
        }
        //else
        {
            pinfo->dau_state[i] = 0;
            pinfo->dau_port_num[i] = 0;
        }
    }

    pinfo->freq = 50;
    pinfo->out_sync = 0;
    pinfo->pt_sync = 0;
    pinfo->in_sync = 0;
    if (pd_state.sync)
    {
        if (PD_SYNC_PT == pd_config.config.sync_mode)
        {
            pinfo->pt_sync = 1;
        }
        else if (PD_SYNC_INSIDE == pd_config.config.sync_mode)
        {
            pinfo->in_sync = 1;
        }
        else if (PD_SYNC_OUTSIDE == pd_config.config.sync_mode)
        {
            pinfo->out_sync = 1;
        }
    }

    for(i = 0; i < PD_PORT_SUM; i++)
    {
        pinfo->port_link_alarm[i] = 0;
    }
    
    _csg_send_data(CSG_REPLY, CSG_C_HEARTBEAT, pkt, sizeof(csg_heartbeat_t), 0);    
    return E_NONE;
}

/* 解析连接报文 */
void _csg_connect_recv(void)
{
    csg.is_connect = TRUE;
    log_warn(LOG_CSG, "CSG connection OK!");
}

void _csg_add_dau_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    uint8_t slot = head->slot;
    if (slot >= PD_SLOTS_MAX || slot == 0)
    {
        DBG(DBG_M_PD_CSG_ERR, "Add DAU %d error!\r\n", slot);
        return;
    }

    slot--;
    printf("_csg_add_dau_recv slot = %d\n", slot);
    daus[slot].state.is_insert = TRUE;
    modbus.card[slot].is_insert = TRUE;
}

/* 解析心跳报文. */
void _csg_heartbeat_recv(char *pkt)
{
    uint32_t server_time = *(uint32_t*)(pkt + sizeof(csg_pkt_head_t));

    //printf("server_time:%d now:%ld\n", server_time, time(NULL));
    if (abs(server_time - time(NULL)) > 3)
    {
        time_set(server_time); //北京时间
    }
}

/* 设备重启报文. */
int32_t _csg_reboot_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    int *data = NULL;
    uint8_t slot = head->slot - 1;
    if (slot > PD_SLOTS_MAX)
    {
        DBG(DBG_M_PD_CSG, "slot:%d error\r\n", head->slot);
        return E_BAD_PARAM;
    }

    data = XMALLOC(MTYPE_CSG, sizeof(int));
    if (!data)
    {
        DBG(DBG_M_PD_CSG_ERR, "XMALLOC ERROR!\r\n");
        return E_MEM;
    }
    
    if (_csg_msg_send_dau(slot, DAU_SEND_RESET, data) != E_NONE)
    {
        XFREE(MTYPE_CSG, data);
    }
    return E_NONE;
}

/* 厂家参数设置报文处理. 
  说明：修改本地ip 设备ID 服务器地址时需要重启。
*/
int32_t _csg_dev_info_set_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    csg_dev_info_t *param = (csg_dev_info_t*)(pkt + sizeof(csg_pkt_head_t));
    csg_dev_info_t *data = NULL;
    uint8_t slot = head->slot - 1;
    if (slot > PD_SLOTS_MAX)
    {
        DBG(DBG_M_PD_CSG, "slot:%d error\r\n", head->slot);
        return E_BAD_PARAM;
    }

    data = XMALLOC(MTYPE_CSG, sizeof(csg_dev_info_t));
    if (!data)
    {
        DBG(DBG_M_PD_CSG_ERR, "XMALLOC ERROR!\r\n");
        return E_MEM;
    }
    
    memcpy(data, param, sizeof(csg_dev_info_t));
    if (_csg_msg_send_dau(slot, DAU_SEND_INFO_SET, data) != E_NONE)
    {
        XFREE(MTYPE_CSG, data);
    }
    return E_NONE;
}

/* 厂家参数查询报文处理. */
int32_t _csg_dev_info_get_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    int *data = NULL;
    uint8_t slot = head->slot - 1;
    if (slot > PD_SLOTS_MAX)
    {
        DBG(DBG_M_PD_CSG, "slot:%d error\r\n", head->slot);
        return E_BAD_PARAM;
    }

    data = XMALLOC(MTYPE_CSG, sizeof(int));
    if (!data)
    {
        DBG(DBG_M_PD_CSG_ERR, "XMALLOC ERROR!\r\n");
        return E_MEM;
    }
    
    if (_csg_msg_send_dau(slot, DAU_SEND_INFO_GET, data) != E_NONE)
    {
        XFREE(MTYPE_CSG, data);
    }
    return E_NONE;
}

/* 配置用户参数报文报文处理. */
int32_t _csg_config_set_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    csg_config_global_t *param = (csg_config_global_t*)(pkt + sizeof(csg_pkt_head_t));
    csg_config_global_t *data = NULL;
    uint8_t slot = head->slot - 1;
    if (slot > PD_SLOTS_MAX)
    {
        DBG(DBG_M_PD_CSG, "slot:%d error\r\n", head->slot);
        return E_BAD_PARAM;
    }

    data = XMALLOC(MTYPE_CSG, sizeof(csg_config_global_t));
    if (!data)
    {
        DBG(DBG_M_PD_CSG_ERR, "XMALLOC ERROR!\r\n");
        return E_MEM;
    }
    
    memcpy(data, param, sizeof(csg_config_global_t));
    if (_csg_msg_send_dau(slot, DAU_SEND_CFG_SET, data) != E_NONE)
    {
        XFREE(MTYPE_CSG, data);
    }
    return E_NONE;
}

/* 查询用户参数查询报文处理. */
int32_t _csg_config_get_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    int *data = NULL;
    uint8_t slot = head->slot - 1;
    if (slot > PD_SLOTS_MAX)
    {
        DBG(DBG_M_PD_CSG, "slot:%d error\r\n", head->slot);
        return E_BAD_PARAM;
    }

    data = XMALLOC(MTYPE_CSG, sizeof(int));
    if (!data)
    {
        DBG(DBG_M_PD_CSG_ERR, "XMALLOC ERROR!\r\n");
        return E_MEM;
    }
    
    if (_csg_msg_send_dau(slot, DAU_SEND_CFG_GET, data) != E_NONE)
    {
        XFREE(MTYPE_CSG, data);
    }
    return E_NONE;
}

/* 通道提交端口参数设置. */
int32_t _csg_port_config_set_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    csg_config_port_t *param = (csg_config_port_t*)(pkt + sizeof(csg_pkt_head_t));
    csg_config_port_t *data = NULL;
    uint8_t slot = head->slot - 1;
    if (slot > PD_SLOTS_MAX)
    {
        DBG(DBG_M_PD_CSG, "slot:%d error\r\n", head->slot);
        return E_BAD_PARAM;
    }

    data = XMALLOC(MTYPE_CSG, sizeof(csg_config_port_t));
    if (!data)
    {
        DBG(DBG_M_PD_CSG_ERR, "XMALLOC ERROR!\r\n");
        return E_MEM;
    }
    
    memcpy(data, param, sizeof(csg_config_port_t));
    if (_csg_msg_send_dau(slot, DAU_SEND_PORT_SET, data) != E_NONE)
    {
        XFREE(MTYPE_CSG, data);
    }
    return E_NONE;
}

/* 按通道提交端口参数查询结果. */
int32_t _csg_port_config_get_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    uint8_t *param = (uint8_t *)(pkt + sizeof(csg_pkt_head_t));
    uint8_t *data = NULL;
    uint8_t slot = head->slot - 1;
    if (slot > PD_SLOTS_MAX)
    {
        DBG(DBG_M_PD_CSG, "slot:%d error\r\n", head->slot);
        return E_BAD_PARAM;
    }

    data = XMALLOC(MTYPE_CSG, sizeof(int));
    if (!data)
    {
        DBG(DBG_M_PD_CSG_ERR, "XMALLOC ERROR!\r\n");
        return E_MEM;
    }
    *data = *param;
    
    if (_csg_msg_send_dau(slot, DAU_SEND_PORT_GET, data) != E_NONE)
    {
        XFREE(MTYPE_CSG, data);
    }
    return E_NONE;
}

/* 解析HUF实时图谱召唤报文. */
int32_t _csg_real_image_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    csg_real_image_get_t *param = (csg_real_image_get_t*)(pkt + sizeof(csg_pkt_head_t));
    csg_real_image_get_t *data = NULL;
    uint8_t slot = head->slot - 1;
    if (slot > PD_SLOTS_MAX)
    {
        DBG(DBG_M_PD_CSG, "slot:%d error\r\n", head->slot);
        return E_BAD_PARAM;
    }

    data = XMALLOC(MTYPE_CSG, sizeof(csg_real_image_get_t));
    if (!data)
    {
        DBG(DBG_M_PD_CSG_ERR, "XMALLOC ERROR!\r\n");
        return E_MEM;
    }
    
    memcpy(data, param, sizeof(csg_real_image_get_t));
    if (_csg_msg_send_dau(slot, DAU_SEND_PRPS_GET, data) != E_NONE)
    {
        XFREE(MTYPE_CSG, data);
    }
    return E_NONE;
}

int32_t _csg_trend_get_recv(char *pkt)
{
    csg_trend_ack_t *ptrend = (csg_trend_ack_t*)(pkt + sizeof(csg_pkt_head_t));
    if (ptrend->boosterpacke)
    {
        if (csg.trend_booster_id == ptrend->index)
        {
            sem_post(&csg.trend_booster_sem);
        }
    }
    else
    {
        if (csg.trend_msg_id == ptrend->index)
        {
            sem_post(&csg.trend_sem);
        }
    }
    return E_NONE;
}

int32_t _csg_event_get_recv(char *pkt)
{
    csg_event_ack_t *pevent = (csg_event_ack_t*)(pkt + sizeof(csg_pkt_head_t));
    if (pevent->boosterpacke)
    {
        if (csg.event_booster_id == pevent->index)
        {
            sem_post(&csg.event_booster_sem);
        }
    }
    else
    {
        if (csg.event_msg_id == pevent->index)
        {
            sem_post(&csg.event_sem);
        }
    }
    return E_NONE;
}

/* 升级文件接收 */
int32_t _csg_upgrade_recv(char *pkt)
{
    static int fd = -1;
    static uint32_t fix_len = 0;
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    csg_upgrade_data_t *head_msg = (csg_upgrade_data_t*)(pkt + CSG_HEAD_LEN);
    char *pdata = pkt + CSG_HEAD_LEN + sizeof(csg_upgrade_data_t);
    upgrade_ack_t ack = {0};
    int32_t size = 0;
    int32_t len_wr = 0;
    int32_t rv = E_NONE;
    uint32_t offset = 0;

    uint8_t slot = head->slot - 1;
    if (slot > PD_SLOTS_MAX)
    {
        DBG(DBG_M_PD_CSG, "slot:%d error\r\n", head->slot);
        return E_BAD_PARAM;
    }
    
    /* 首保处理, 打开文件描述符, 初始化变量 */
    if (head_msg->index == 0)
    {
        if (fd > 0)
        {
            close(fd);
            fd = -1;
        }

        fd = open(PD_UPG_SOFTWARE, O_WRONLY | O_CREAT | O_TRUNC, 0777);
        if (fd < 0)
        {
            DBG(DBG_M_PD_CSG_ERR, "Open file " PD_UPG_SOFTWARE " error!\n");
            return E_SYS_CALL;
        }
        fix_len = head_msg->len;
        
        DBG(DBG_M_PD_CSG, "Receive upgrade file start.\n");
    }
    DBG(DBG_M_PD_CSG_ERR,"type=%d,sum=%d,index=%d,len=%d,fix_len=%d\n", head_msg->type, head_msg->sum, head_msg->index, head_msg->len, fix_len);

    /* 收包流程 */
    size = head_msg->len;
    offset = head_msg->index * fix_len;
    if (lseek(fd, offset, SEEK_SET) < 0)
    {
       DBG(DBG_M_PD_CSG_ERR, "lseek file " PD_UPG_SOFTWARE " error!\n"); 
       return E_SYS_CALL;
    }
    len_wr = write(fd, pdata, size);
    if (len_wr != size)
    {
        DBG(DBG_M_PD_CSG_ERR, "Write file " PD_UPG_SOFTWARE " error!\n");
        return E_SYS_CALL;
    }
    
    /* 最后一个报文处理 */
    if (head_msg->sum - 1 == head_msg->index)
    {
        close(fd);
        fd = -1;
        DBG(DBG_M_PD_CSG, "Receive upgrade file end.\n");
        csg_upate_msg_t *data;

        data = XMALLOC(MTYPE_CSG, sizeof(csg_upate_msg_t));
        if (!data)
        {
            DBG(DBG_M_PD_CSG_ERR, "XMALLOC ERROR!\r\n");
            rv = E_ERROR;
            goto UP_ERR;
         }
        
        data->type = head_msg->type;

        char buff[64] = {0};
        if (getcwd(buff, sizeof(buff)) == NULL) 
        {
            DBG(DBG_M_PD_CSG_ERR, "getcwd error!\n");
            rv = E_ERROR;
            goto UP_ERR;
        } 

        snprintf(data->name, 128, "%s/%s", buff, PD_UPG_SOFTWARE);
        
        if (_csg_msg_send_dau(slot, DAU_SEND_UPDATE, data) != E_NONE)
        {
            XFREE(MTYPE_CSG, data);
        }
    }

UP_ERR:    
    ack.index = head_msg->index;
    ack.result = TRUE;

    DBG(DBG_M_PD_CSG_ERR," send ack\n");

    /* 发送应答 */
    memcpy(pkt + sizeof(csg_pkt_head_t), (char *)&ack, sizeof(csg_ack_t));
    _csg_send_data(CSG_REPLY, head->cmd, pkt, sizeof(ack), 0);     

     /* 如果升级线程开启失败返回错误信息. */
    if (rv != E_NONE)
    {
        printf("Upgrade start failed.\n");
        csg_upgrade_result_send(0, "Upgrade start failed.");
    }

    return E_NONE;
}

int32_t _csg_recv_process(char *pkt, uint32_t len)
{
    csg_pkt_head_t *head = (csg_pkt_head_t *)pkt;
  
    /* 报文头和 CRC 校验. */
    LD_E_RETURN(DBG_M_PD_CSG_ERR, _csg_pkt_check(pkt));
    csg.heartbeat_timeout_cnt = 0;

    if (CSG_REQUEST == head->cmd_type)
    {
        switch (head->cmd)
        {
        case CSG_C_CONTACT:
            _csg_connect_recv();
            break;
        case CSG_C_ADD_DAU:
            _csg_add_dau_recv(pkt);
            break;
        case CSG_C_RESET:
            _csg_reboot_recv(pkt);
            break;
        case CSG_C_UPDATE:
            _csg_upgrade_recv(pkt);
            break;
        case CSG_C_HEARTBEAT:
            _csg_heartbeat_recv(pkt);
            break;
        case CSG_C_DEV_INFO_SET:
            _csg_dev_info_set_recv(pkt);
            break;
        case CSG_C_DEV_INFO_GET:
            _csg_dev_info_get_recv(pkt);
            break;
        default:
            break;
        }
    }
    else if (CSG_PRV_REQUEST == head->cmd_type)
    {
        switch (head->cmd)
        {
        case CSG_PRV_CONFIG_GLOBAL_SET:
            _csg_config_set_recv(pkt);
            break;
        case CSG_PRV_CONFIG_GLOBAL_GET:
            _csg_config_get_recv(pkt);
            break;
        case CSG_PRV_CONFIG_PORT_SET:
            _csg_port_config_set_recv(pkt);
            break;
        case CSG_PRV_CONFIG_PORT_GET:
            _csg_port_config_get_recv(pkt);
            break;
        case CSG_PRV_CONFIG_REAL_WAVE:
            _csg_real_image_recv(pkt);
            break;
        case CSG_PRV_TREND:
            _csg_trend_get_recv(pkt);
            break;
        case CSG_PRV_REAL_PRPS:
            break;
        case CSG_PRV_EVENT:
            _csg_event_get_recv(pkt);
            break;
        default:
            break;
        }
    }
    return E_NONE;
}

int32_t _csg_send_add_dau(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL == data)
    {
        return E_BAD_PARAM;
    }
    memcpy(pkt + sizeof(csg_pkt_head_t), data, sizeof(csg_contact_t));
    _csg_send_data(CSG_REPLY, CSG_C_ADD_DAU, pkt, sizeof(csg_contact_t), slot + 1);
    return E_NONE;
}

int32_t _csg_send_reset(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    csg_ack_t ack = {0};
    ack.result = TRUE;
    memcpy(pkt + sizeof(csg_pkt_head_t), (char *)&ack, sizeof(csg_ack_t));
    _csg_send_data(CSG_REPLY, CSG_C_RESET, pkt, sizeof(csg_ack_t), slot + 1);
    return E_NONE;
}

int32_t _csg_send_info_set(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    csg_ack_t ack = {0};
    ack.result = TRUE;
    memcpy(pkt + sizeof(csg_pkt_head_t), (char *)&ack, sizeof(csg_ack_t));
    _csg_send_data(CSG_REPLY, CSG_C_DEV_INFO_SET, pkt, sizeof(csg_ack_t), slot + 1);
    return E_NONE;
}

int32_t _csg_send_info_get(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL == data)
    {
        return E_BAD_PARAM;
    }
    
    memcpy(pkt + sizeof(csg_pkt_head_t), data, sizeof(csg_dev_info_t));
    _csg_send_data(CSG_REPLY, CSG_C_DEV_INFO_GET, pkt, sizeof(csg_dev_info_t), slot + 1);
    return E_NONE;
}

int32_t _csg_send_updata_rv(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL == data)
    {
        return E_BAD_PARAM;
    }
    
    memcpy(pkt + sizeof(csg_pkt_head_t), data, sizeof(upgrade_res_t));
    _csg_send_data(CSG_REPLY, CSG_C_UPDATE_RESULT, pkt, sizeof(upgrade_res_t), slot + 1);
    return E_NONE;
}

int32_t _csg_send_heartbeat(uint8_t slot, void *data)
{
    csg_timestamp_t *param = NULL;
    struct timespec t;

    param = XMALLOC(MTYPE_CSG, sizeof(csg_timestamp_t));
    if (!param)
    {
        DBG(DBG_M_PD_CSG_ERR, "XMALLOC ERROR!\r\n");
        return E_MEM;
    }
    clock_gettime(CLOCK_MONOTONIC, &t);
    param->sec = time(NULL);
    param->msec = t.tv_nsec / 1000000;
    printf("sec=%u msec=%u\n", param->sec, param->msec);

    if (_csg_msg_send_dau(slot, DAU_SEND_HEARTBEAT, param) != E_NONE)
    {
        XFREE(MTYPE_CSG, param);
    }
    return E_NONE;
}

int32_t _csg_send_cfg_set(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    csg_ack_t ack = {0};
    ack.result = TRUE;
    memcpy(pkt + sizeof(csg_pkt_head_t), (char *)&ack, sizeof(csg_ack_t));
    _csg_send_data(CSG_PRV_REPLY, CSG_PRV_CONFIG_GLOBAL_SET, pkt, sizeof(csg_ack_t), slot + 1);
    return E_NONE;
}

int32_t _csg_send_cfg_get(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL == data)
    {
        return E_BAD_PARAM;
    }
    
    memcpy(pkt + sizeof(csg_pkt_head_t), data, sizeof(csg_config_global_t));
    _csg_send_data(CSG_PRV_REPLY, CSG_PRV_CONFIG_GLOBAL_GET, pkt, sizeof(csg_config_global_t), slot + 1);
    return E_NONE;
}

int32_t _csg_send_cfg_set_port(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    csg_config_port_ack_t ack = {0};
    hf_cfg_port_ack_t *psrc = (hf_cfg_port_ack_t *)data;
    ack.result = TRUE;
    ack.vport = psrc->vport;
    memcpy(pkt + sizeof(csg_pkt_head_t), (char *)&ack, sizeof(csg_config_port_ack_t));
    _csg_send_data(CSG_PRV_REPLY, CSG_PRV_CONFIG_PORT_SET, pkt, sizeof(csg_config_port_ack_t), slot + 1);
    return E_NONE;
}

int32_t _csg_send_cfg_get_port(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL == data)
    {
        return E_BAD_PARAM;
    }
    
    memcpy(pkt + sizeof(csg_pkt_head_t), data, sizeof(csg_config_port_t));
    _csg_send_data(CSG_PRV_REPLY, CSG_PRV_CONFIG_PORT_GET, pkt, sizeof(csg_config_port_t), slot + 1);
    return E_NONE;
}

int32_t _csg_send_prps_get(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    csg_ack_t ack = {0};
    ack.result = TRUE;
    memcpy(pkt + sizeof(csg_pkt_head_t), (char *)&ack, sizeof(csg_ack_t));
    _csg_send_data(CSG_PRV_REPLY, CSG_PRV_CONFIG_REAL_WAVE, pkt, sizeof(csg_ack_t), slot + 1);
    return E_NONE;
}

int32_t _csg_send_prps_data(uint8_t slot, void *data)
{
    char *pkt = csg.real_buf;
    csg_real_image_t *head = (csg_real_image_t *)(pkt + sizeof(csg_pkt_head_t));
    char *pdata = pkt + sizeof(csg_pkt_head_t) + sizeof(csg_real_image_t);
    hf_prps_t *real = (hf_prps_t *)data;
    
    uint8_t i = 0;
    uint32_t sum = 0;
    uint32_t len = 0;
    uint32_t last_pack_len = 0;
    uint32_t port_len = 0;  

    if (NULL == data)
    {
        return E_BAD_PARAM;
    }

    head->index = real->head.index;
    head->fre_cnt = real->head.fre_cnt;
    head->vport = real->head.vport;
    head->max = real->head.max;
    head->avg = real->head.avg;
    head->cnt = real->head.cnt;
    head->utc = real->head.utc;
    head->freq = real->head.freq;

    /* 每个工频周期 256 个点, 每个点占 2byte */
    port_len  = real->head.fre_cnt << 9;
    sum = port_len / CSG_PKT_LEN;
    last_pack_len = port_len % CSG_PKT_LEN;
    if (last_pack_len != 0)
    {
        sum += 1;
    }
    head->pkt_sum = sum;
    
    DBG(DBG_M_PD_CSG, "max = %d avg = %d cnt=%d fre_cnt=%d freq=%f port_len=%d\r\n", head->max, head->avg, head->cnt, head->fre_cnt, head->freq, port_len);
    for (i = 0; i < sum; i++)
    {
        len = (i == sum - 1) ? last_pack_len : CSG_PKT_LEN;
        head->pkt_index = i;
        head->len = len;
        memcpy(pdata, (char *)real->prps + CSG_PKT_LEN * i, len);
        
        /* 这里延迟 500us, 是因为当使用 100M 光转电模块时, 因为我们的光口是千兆的所以会丢包 */
        usleep(500);
        _csg_send_data(CSG_PRV_REPLY, CSG_PRV_REAL_PRPS, pkt, sizeof(csg_real_image_t) + len, slot + 1);    
    }    
    return E_NONE;
}

int32_t _csg_send_event_data(uint8_t slot, void *data)
{
    char *pkt = csg.event_buf;
    csg_event_head_t *head = (csg_event_head_t *)(pkt + sizeof(csg_pkt_head_t));
    char *pdata = pkt + sizeof(csg_pkt_head_t) + sizeof(csg_event_head_t);
    hf_event_t *event = (hf_event_t *)data;
    struct timespec ts;
    uint8_t err_cnt = 0;
    uint32_t index = 0;
    uint32_t event_len = 0;
    uint32_t sum = 0;

    if (NULL == event)
    {
        return E_BAD_PARAM;
    }
    
    if (PD_EVENT_TYPE_NONE == event->head.type)
    {
        return E_NONE;
    }

    index = 0;
    event_len = event->head.point_cnt * sizeof(pd_data_point_t);
    sum = event_len / CSG_PKT_LEN;
    if (event_len % CSG_PKT_LEN)
    {
        sum += 1;
    }

    head->sum = sum;
    head->vport = event->head.vport;
    head->boosterpack = 0;
    head->power_fre = event->head.power_fre;
    head->type = event->head.type;
    head->max = event->head.max;
    head->identifier = event->head.index;
    head->utc = event->head.utc;
    head->cnt = event->head.cnt;
    head->avg_o = event->head.avg_o;
    head->avg = event->head.avg;
    head->point_cnt = event->head.point_cnt;
            
    while (index < sum)
    {
        csg.event_msg_id = index;
        head->len = event_len > CSG_PKT_LEN ? CSG_PKT_LEN : event_len;
        head->index = index;
        
        memcpy(pdata, event->point + index * CSG_PKT_LEN, head->len);
        _csg_send_data(CSG_PRV_REPLY, CSG_PRV_EVENT, pkt, head->len + sizeof(csg_event_head_t), slot + 1);

        clock_gettime(CLOCK_REALTIME, &ts);
        ts.tv_sec += CSG_SEND_TIMEOUT;  // 设置 3 秒超时
        if (sem_timedwait(&csg.event_sem, &ts) != 0)
        {
            err_cnt++;
            if (err_cnt >= CSG_SEND_ERR_CNT)
            {
                _csg_disconnect_set(__FUNCTION__);
                return E_TIMEOUT;
            }
            DBG(DBG_M_PD_CSG_ERR, "event sem error:%s\r\n", strerror(errno));
            continue;
        }
        
        event_len -= head->len;
        index++;
        err_cnt = 0;
    }
    return E_NONE;
}

int32_t _csg_event_file_send(void)
{
    struct timespec ts;
    csg_event_head_t *head = NULL;
    char *pdata = NULL;
    char *pkt = csg.event_booster_buf;
    csg_event_head_t *pack = (csg_event_head_t *)(pkt + sizeof(csg_pkt_head_t));
    char *data = pkt + sizeof(csg_pkt_head_t) + sizeof(csg_event_head_t);
    uint8_t err_cnt = 0;
    uint32_t index = 0;
    uint32_t event_len = 0;
    uint32_t sum = 0;
    uint8_t slot = 0;
    
    head = &_csg_event.head;
    pdata = (char *)_csg_event.point;
    event_len = head->point_cnt * sizeof(pd_data_point_t);
    sum = event_len / CSG_PKT_LEN;
    if (event_len % CSG_PKT_LEN)
    {
        sum += 1;
    }

    slot = head->slot;
    pack->sum = sum;
    pack->vport = head->vport;
    pack->boosterpack = 1;
    pack->power_fre = head->power_fre;
    pack->type = head->type;
    pack->max = head->max;
    pack->identifier = head->index;
    pack->utc = head->utc;
    pack->cnt = head->cnt;
    pack->avg_o = head->avg_o;
    pack->avg = head->avg;
    pack->point_cnt = head->point_cnt;
    
    while (index < sum)
    {
        csg.event_booster_id = index;
        pack->len = event_len > CSG_PKT_LEN ? CSG_PKT_LEN : event_len;
        pack->index = index;
        
        memcpy(data, pdata + index * CSG_PKT_LEN, pack->len);
        _csg_send_data(CSG_PRV_REPLY, CSG_PRV_EVENT, pkt, pack->len + sizeof(csg_event_head_t), slot + 1);

        clock_gettime(CLOCK_REALTIME, &ts);
        ts.tv_sec += CSG_SEND_TIMEOUT;  // 设置 3 秒超时
        if (sem_timedwait(&csg.event_booster_sem, &ts) != 0)
        {
            err_cnt++;
            if (err_cnt >= CSG_SEND_ERR_CNT)
            {
                _csg_disconnect_set(__FUNCTION__);
                return E_TIMEOUT;
            }
            DBG(DBG_M_PD_CSG_ERR, "event sem error:%s\r\n", strerror(errno));
            continue;
        }
        
        event_len -= pack->len;
        index++;
        err_cnt = 0;
    }
    
    return E_NONE;
}

/* 趋势文件发送 */
int32_t _csg_trend_file_send(char *filename)
{
    struct timespec ts;
    FILE *file = NULL;
    csg_trend_file_t filehead = {0};
    char *pkt = csg.trend_booster_buf;
    csg_trend_t *pack = (csg_trend_t *)(pkt + sizeof(csg_pkt_head_t));
    char *data = pkt + sizeof(csg_pkt_head_t) + sizeof(csg_trend_t);
    uint8_t err_cnt = 0;
    uint32_t index = 0;
    uint32_t prps_len = 0;
    uint32_t sum = 0;
    uint32_t len = 0;
    uint32_t readbyte = 0;
    int offset = 0;
    uint8_t slot; 

    if (csg.trend_file.index_min == csg.trend_file.index_max)
    {
        return E_EMPTY;
    }
    
    file = fopen(filename, "r");
    if (!file) 
    {
        DBG(DBG_M_PD_CSG_ERR, "Open failed!\r\n");
        return E_NONE;
    }

    while(1)
    {
        fseek(file, offset, SEEK_SET);
        readbyte = fread(&filehead, 1, sizeof(csg_trend_file_t), file);
        if (readbyte != sizeof(csg_trend_file_t))
        {
            if (readbyte != 0)
            {
                DBG(DBG_M_PD_CSG_ERR,"read %d != %d\r\n", len, readbyte, strerror(errno));
            }
            fclose(file);
            return E_NONE;
        }

        printf("#1 %d %d %d %d %d\r\n", filehead.vport, filehead.type, filehead.len, filehead.identifier, filehead.utc);
        pack->type = filehead.type;
        pack->vport = filehead.vport;
        pack->boosterpack = 1;
        pack->identifier = filehead.identifier;
        pack->utc = filehead.utc;
        slot = filehead.slot;
        prps_len = filehead.len;
        sum = prps_len / CSG_PKT_LEN;
        if (prps_len % CSG_PKT_LEN)
        {
            sum += 1;
        }
        pack->sum = sum;

        offset += sizeof(csg_trend_file_t);
        index = 0;
        while (index < sum)
        {
            fseek(file, offset, SEEK_SET);
            csg.trend_booster_id = index;
            len = prps_len > CSG_PKT_LEN ? CSG_PKT_LEN : prps_len;
            
            pack->len = len;
            pack->index = index;

            readbyte = fread(data, 1, len, file);
            if (readbyte != len)
            {
                DBG(DBG_M_PD_CSG_ERR,"read %d != %d err:%s\r\n", len, readbyte, strerror(errno));
                fclose(file);
                return E_NONE;
            }
            
            _csg_send_data(CSG_PRV_REPLY, CSG_PRV_TREND, pkt, len + sizeof(csg_trend_t), slot + 1);

            clock_gettime(CLOCK_REALTIME, &ts);
            ts.tv_sec += CSG_SEND_TIMEOUT;  // 设置 3 秒超时
            if (sem_timedwait(&csg.trend_booster_sem, &ts) != 0)
            {
                err_cnt++;
                if (err_cnt >= CSG_SEND_ERR_CNT)
                {
                    fclose(file);
                    _csg_disconnect_set(__FUNCTION__);
                    return E_TIMEOUT;
                }
                DBG(DBG_M_PD_CSG_ERR, "trend sem error:%s\r\n", strerror(errno));
                continue;
            }
            
            prps_len -= len;
            index++;
            err_cnt = 0;
            offset += len;
        }
    }
    
    fclose(file);
    return E_NONE;
}

int32_t _csg_send_trend_prps_data(uint8_t slot, void *data)
{
    struct timespec ts;
    char *pkt = csg.trend_buf;
    csg_trend_t *head = (csg_trend_t *)(pkt + sizeof(csg_pkt_head_t));
    char *pdata = pkt + sizeof(csg_pkt_head_t) + sizeof(csg_trend_t);
    
    dau_t *dau = &daus[slot];
    hf_data_t *hf_data = (hf_data_t*)dau->private_data;
    hf_trend_prps_t *prps = &hf_data->trend.prps;
    hf_trend_col_t *col = &hf_data->trend.col;

    uint8_t err_cnt = 0;
    uint32_t index = 0;
    uint32_t prps_len = 0;
    uint32_t sum = 0;
    uint32_t len = 0;
    
    uint8_t port = *(uint32_t *)data;
    if (port >= 8)
        return E_BAD_PARAM;
    
    head->type = CSG_TREND_TYPE_PRPS;
    head->identifier = col->index;
    head->utc = col->utc;
    
    index = 0;
    prps_len = prps->port[port].point_cnt * sizeof(pd_data_point_t);
    sum = prps_len / CSG_PKT_LEN;
    if (prps_len % CSG_PKT_LEN)
    {
        sum += 1;
    }

    head->vport = port + 1;
    head->sum = sum;
            
    while (index < sum)
    {
        csg.trend_msg_id = index;
        len = prps_len > CSG_PKT_LEN ? CSG_PKT_LEN : prps_len;
        
        head->len = len;
        head->index = index;
        
        memcpy(pdata, (char *)prps->port[port].point + index * CSG_PKT_LEN, len);
        _csg_send_data(CSG_PRV_REPLY, CSG_PRV_TREND, pkt, len + sizeof(csg_trend_t), slot + 1);

        clock_gettime(CLOCK_REALTIME, &ts);
        ts.tv_sec += CSG_SEND_TIMEOUT;  // 设置 3 秒超时
        if (sem_timedwait(&csg.trend_sem, &ts) != 0)
        {
            err_cnt++;
            if (err_cnt >= CSG_SEND_ERR_CNT)
            {
                _csg_disconnect_set(__FUNCTION__);
                return E_TIMEOUT;
            }
            DBG(DBG_M_PD_CSG_ERR, "trend sem error:%s\r\n", strerror(errno));
            continue;
        }
        
        prps_len -= len;
        index++;
        err_cnt = 0;
    }
    return E_NONE;
}

int32_t _csg_send_trend_prpd_data(uint8_t slot, void *data)
{
    struct timespec ts;
    char *pkt = csg.trend_buf;
    csg_trend_t *head = (csg_trend_t *)(pkt + sizeof(csg_pkt_head_t));
    char *pdata = pkt + sizeof(csg_pkt_head_t) + sizeof(csg_trend_t);
    
    dau_t *dau = &daus[slot];
    hf_data_t *hf_data = (hf_data_t*)dau->private_data;
    hf_trend_col_t *col = &hf_data->trend.col;
    hf_trend_prpd_t *prpd = &hf_data->trend.prpd;

    uint8_t err_cnt = 0;
    uint32_t index = 0;
    uint32_t prps_len = 0;
    uint32_t sum = 0;
    uint32_t len = 0;
    
    uint8_t port = *(uint32_t *)data;
    if (port >= 8)
        return E_BAD_PARAM;
    
    head->type = CSG_TREND_TYPE_PRPD;
    head->identifier = col->index;
    head->utc = col->utc;
    
    index = 0;
    prps_len = CSG_TREND_PRPD_PORT_LEN;
    sum = prps_len / CSG_PKT_LEN;
    if (prps_len % CSG_PKT_LEN)
    {
        sum += 1;
    }

    head->vport = port + 1;
    head->sum = sum;
            
    while (index < sum)
    {
        csg.trend_msg_id = index;
        len = prps_len > CSG_PKT_LEN ? CSG_PKT_LEN : prps_len;
        
        head->len = len;
        head->index = index;
        
        memcpy(pdata, (char *)prpd->port[port].data + index * CSG_PKT_LEN, len);
        _csg_send_data(CSG_PRV_REPLY, CSG_PRV_TREND, pkt, len + sizeof(csg_trend_t), slot + 1);

        clock_gettime(CLOCK_REALTIME, &ts);
        ts.tv_sec += CSG_SEND_TIMEOUT;  // 设置 3 秒超时
        if (sem_timedwait(&csg.trend_sem, &ts) != 0)
        {
            err_cnt++;
            if (err_cnt >= CSG_SEND_ERR_CNT)
            {
                _csg_disconnect_set(__FUNCTION__);
                return E_TIMEOUT;
            }
            DBG(DBG_M_PD_CSG_ERR, "trend sem error:%s\r\n", strerror(errno));
            continue;
        }
        
        prps_len -= len;
        index++;
        err_cnt = 0;
    }
    return E_NONE;
}

int32_t _csg_send_trend_statistics_data(uint8_t slot, void *data)
{
    struct timespec ts;
    char *pkt = csg.trend_buf;
    csg_trend_t *head = (csg_trend_t *)(pkt + sizeof(csg_pkt_head_t));
    csg_trend_stat *pdata = (csg_trend_stat*)(pkt + sizeof(csg_pkt_head_t) + sizeof(csg_trend_t));
    
    dau_t *dau = &daus[slot];
    hf_data_t *hf_data = (hf_data_t*)dau->private_data;
    hf_trend_col_t *col = &hf_data->trend.col;

    uint8_t err_cnt = 0;
    uint32_t index = 0;
    uint32_t prps_len = 0;
    uint32_t sum = 0;
    uint32_t len = 0;
    
    uint8_t port = *(uint32_t *)data;
    if (port >= 8)
        return E_BAD_PARAM;
    
    head->type = CSG_TREND_TYPE_STAT;
    head->identifier = col->index;
    head->utc = col->utc;
    
    index = 0;
    prps_len = sizeof(csg_trend_stat);
    sum = prps_len / CSG_PKT_LEN;
    if (prps_len % CSG_PKT_LEN)
    {
        sum += 1;
    }

    head->vport = port + 1;
    head->sum = sum;
            
    while (index < sum)
    {
        csg.trend_msg_id = index;
        len = prps_len > CSG_PKT_LEN ? CSG_PKT_LEN : prps_len;
        
        head->len = len;
        head->index = index;

        pdata->data_cnt = col->port[port].data_cnt;
        pdata->max = col->port[port].max;
        pdata->avg = col->port[port].avg;
        pdata->cnt = col->port[port].cnt;
        pdata->phase = col->port[port].phase;
        pdata->noise = col->port[port].noise;
        pdata->event_cnt = col->port[port].event_cnt;
        
        _csg_send_data(CSG_PRV_REPLY, CSG_PRV_TREND, pkt, len + sizeof(csg_trend_t), slot + 1);

        clock_gettime(CLOCK_REALTIME, &ts);
        ts.tv_sec += CSG_SEND_TIMEOUT;  // 设置 3 秒超时
        if (sem_timedwait(&csg.trend_sem, &ts) != 0)
        {
            err_cnt++;
            if (err_cnt >= CSG_SEND_ERR_CNT)
            {
                _csg_disconnect_set(__FUNCTION__);
                return E_TIMEOUT;
            }
            DBG(DBG_M_PD_CSG_ERR, "trend sem error:%s\r\n", strerror(errno));
            continue;
        }
        
        prps_len -= len;
        index++;
        err_cnt = 0;
    }
    return E_NONE;
}

int32_t _csg_send_trend_data(uint8_t slot, void *data)
{
    _csg_send_trend_prps_data(slot, data);
    _csg_send_trend_prpd_data(slot, data);
    _csg_send_trend_statistics_data(slot, data);
    return E_NONE;
}

int32_t _csg_send_modbus_data(uint8_t slot, int32_t len, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL == data)
    {
        return E_BAD_PARAM;
    }
    memcpy(pkt + sizeof(csg_pkt_head_t), data, len);
    _csg_send_data(CSG_PRV_REPLY, CSG_PRV_MODBUS, pkt, len, slot + 1);
    return E_NONE;
}

/* csg向后台发包处理函数 */
int32_t _csg_send_process(uint8_t type, uint8_t slot, void *data)
{
    //csg_pkt_head_t *head = (csg_pkt_head_t*)data;
    /* 报文头和 CRC 校验. */    
    //LD_E_RETURN_N(_hf_pkt_check(slot, pkt));
    if (_csg_send_command[type])
    {
        _csg_send_command[type](slot, data);
    }

#if 0
    if (CSG_REQUEST == head->cmd_type)
    {
        _csg_send_command[type](slot, data);
    }
    else if (CSG_PRV_REQUEST == head->cmd_type)
    {
        _csg_send_prv_command[type](slot, data);
    }
#endif
    
    return E_NONE;
}

/* 创建目录 */
int32_t _csg_create_dir(char *dir)
{
    uint32_t i = 0;
    uint32_t len = 0;
    
    len = strlen(dir);

    /* 循环创建前级目录 */
    for(i = 0; i < len; i++)
    {
        if(dir[i] == '/')
        {
            dir[i] = '\0';
            if(access(dir, 0) != 0)
            {
                mkdir(dir, 744);
            }
            dir[i] ='/';
        }
    }

    /* 创建最后级目录 */
    if(len > 0 && access(dir, 0) != 0)
    {
        mkdir(dir, 744);
    }
    
    return E_NONE;
}

/* 在文件末尾写入数据 */
int32_t _csg_write_file(char *filename, char *filehead, int headsize, char *filedata, int datasize)
{
    /* 写入文件 */
    FILE *file = fopen(filename, "ab");
    if (!file) 
    {
        DBG(DBG_M_PD_CSG_ERR, "Open error.\r\n");
        return E_SYS_CALL;
    }

    fseek(file, 0, SEEK_END);   // 移动到文件末尾

    size_t bytes_written = fwrite(filehead, 1, headsize, file);
    if (bytes_written != headsize) 
    {
        DBG(DBG_M_PD_CSG_ERR, "Write error.\r\n");
        fclose(file);
        unlink(filename);       // 删除不完整文件
        return E_SYS_CALL;
    }

    bytes_written = fwrite(filedata, 1, datasize, file);
    if (bytes_written != datasize) 
    {
        DBG(DBG_M_PD_CSG_ERR, "Write error.\r\n");
        fclose(file);
        unlink(filename);       // 删除不完整文件
        return E_SYS_CALL;
    }

    /* 确保数据落盘 */
    fsync(fileno(file));
    fclose(file);
    return E_NONE;
}

/* 趋势 prps 写入文件 */
int32_t _csg_trend_prps_write_file(uint8_t slot, uint8_t port, char *filepath)
{
    hf_trend_prps_port_t *data_port = NULL;
    csg_trend_file_t file_head = {0};

    dau_t *dau = &daus[slot];
    hf_data_t *hf_data = (hf_data_t*)dau->private_data;
    hf_trend_prps_t *prps = &hf_data->trend.prps;
    hf_trend_col_t *col = &hf_data->trend.col;
    
    data_port = &prps->port[port];

    file_head.type  = CSG_TREND_TYPE_PRPS;
    file_head.vport = port + 1;
    file_head.port_num = 1;
    file_head.slot = slot;
    file_head.identifier = col->index;
    file_head.utc = col->utc;
    file_head.len = data_port->point_cnt * sizeof(pd_data_point_t);

    LD_E_RETURN(DBG_M_PD_CSG_ERR, _csg_write_file(filepath, (char *)&file_head, sizeof(file_head), (char*)data_port->point, file_head.len));

    return E_NONE;
}

/* 趋势 prpd 写入文件 */
int32_t _csg_trend_prpd_write_file(uint8_t slot, uint8_t port, char *filepath)
{
    pd_trend_prpd_port_t *data_port = NULL;
    csg_trend_file_t file_head = {0};

    dau_t *dau = &daus[slot];
    hf_data_t *hf_data = (hf_data_t*)dau->private_data;
    hf_trend_prpd_t *prpd = &hf_data->trend.prpd;
    hf_trend_col_t *col = &hf_data->trend.col;

    data_port = &prpd->port[port];

    file_head.type  = CSG_TREND_TYPE_PRPD;
    file_head.vport = port + 1;
    file_head.port_num = 1;
    file_head.slot = slot;
    file_head.identifier = col->index;
    file_head.utc = col->utc;
    file_head.len = sizeof(pd_trend_prpd_port_t);

    LD_E_RETURN(DBG_M_PD_CSG_ERR, _csg_write_file(filepath, (char *)&file_head, sizeof(file_head), (char*)data_port, file_head.len));
    
    return E_NONE;
}

/* 趋势统计数据写入文件 */
int32_t _csg_trend_statistics_write_file(uint8_t slot, uint8_t port, char *filepath)
{
    hf_trend_stat *data_port = NULL;
    csg_trend_file_t file_head = {0};
    csg_trend_stat stat = {0};

    dau_t *dau = &daus[slot];
    hf_data_t *hf_data = (hf_data_t*)dau->private_data;
    hf_trend_col_t *col = &hf_data->trend.col;
    
    data_port = &col->port[port];

    file_head.type  = CSG_TREND_TYPE_STAT;
    file_head.vport = port + 1;
    file_head.port_num = 1;
    file_head.slot = slot;
    file_head.identifier = col->index;
    file_head.utc = col->utc;
    file_head.len = sizeof(csg_trend_stat);

    stat.data_cnt = data_port->data_cnt;
    stat.max = data_port->max;
    stat.avg = data_port->avg;
    stat.cnt = data_port->cnt;
    stat.phase = data_port->phase;
    stat.noise = data_port->noise;
    stat.event_cnt = data_port->event_cnt;

    LD_E_RETURN(DBG_M_PD_CSG_ERR, _csg_write_file(filepath, (char *)&file_head, sizeof(file_head), (char *)&stat, file_head.len));

    
    return E_NONE;
}

int32_t _csg_event_write_file(uint8_t slot, void *data)
{
    int32_t len = 0;
    hf_event_t *event = (hf_event_t *)data;
    if (NULL == event)
    {
        return E_BAD_PARAM;
    }

    csg_event_head_t *head = (csg_event_head_t *)data;
    head->slot = slot;  

    len = sizeof(event->head) + event->head.point_cnt * sizeof(hf_data_point_t);
    file_fifo_write(&csg.event_file, (char*)event, len);
    return E_NONE;
}

/* 趋势数据写入文件 */
int32_t _csg_trend_write_file(uint8_t slot, void *data)
{
    int ret = E_NONE;
    char filepath[CSG_FILE_FIFO_PATH_LEN] = {0};

    uint8_t port = *(uint32_t *)data;
    if (port >= 8)
        return E_BAD_PARAM;
    
    if (csg.trend_file.index_max - csg.trend_file.index_min >= csg.trend_file.files_max)
    {
        return E_FULL;
    }
    
    snprintf(filepath, 128, "%s/%ld", CSG_TREND_NAME, csg.trend_file.index_max);

    ret |= _csg_trend_prps_write_file(slot, port, filepath);
    ret |= _csg_trend_prpd_write_file(slot, port, filepath);
    ret |= _csg_trend_statistics_write_file(slot, port, filepath);
    if (E_NONE == ret)
    {
        csg.trend_file.index_max++;
    }
    
    return ret;
}


/* 初始化最大最小索引 */
int32_t _csg_file_fifo_init(csg_file_fifo_t *config)
{
    DIR *dir = opendir(config->dir);
    struct dirent *entry;
    uint8_t first = 1;
    int64_t index = 0;

    /* 初始化最大最小 index */
    config->index_max = 0;
    config->index_min = 0;
    
    if (!dir)
    {
        _csg_create_dir(config->dir);
    }
    else
    {
        while ((entry = readdir(dir)) != NULL)
        {
            if (entry->d_type != DT_REG)
            {
                continue;
            }

            if (!isdigit(entry->d_name[0]))
            {
                continue;
            }
            //printf("filename:%s\n", entry->d_name);

            index = strtoll(entry->d_name, NULL, 10);
            if (first)
            {
                config->index_max = config->index_min = index;
                first = 0;
            }
            else
            {
                if (index > config->index_max)
                {
                    config->index_max = index;
                }

                if (index < config->index_min)
                {
                    config->index_min = index;
                }
            }
        }

        /* 非空目录最大索引要加一 */
        if (!first)
        {
            config->index_max++;
        }
        
        closedir(dir);
    }

    return E_NONE;
}

/* 删除最小索引文件 */
void _csg_file_fifo_delete_by_min_index(csg_file_fifo_t *config)
{
    char filename[512];
    
    snprintf(filename, sizeof(filename), "%s/%ld", config->dir, config->index_min);
    remove(filename);
    
    /* 更新最小索引 */
    config->index_min++;
}

/* 普通命令发送平台线程 */
void *_csg_send_cmd_handle(void *arg)
{
    dau_send_msg_t *recv_msg = NULL;

    /* 等待初始化完成 */
    while(!is_system_init)
    {
        sleep(1);
    }

    while(1)
    {
        if (fifo_read(csg.fifo_cmd_id, (void**)&recv_msg) != 0)
        {
            DBG(DBG_M_PD_CSG_ERR, "ERROR at fifo %d read!\r\n", csg.fifo_cmd_id);
            continue;
        }

        /* 响应其他报文 */
        if (csg.send_cb)
        {
            csg.send_cb(recv_msg->type, recv_msg->slot, recv_msg->data);
        }

        /* 释放数据内存, 注意一定要在 fifo_push 之前调用, 因为 fifo_push 后 recv_msg 已被释放. */
        XFREE(MTYPE_CSG, recv_msg->data);
        fifo_push(csg.fifo_cmd_id);
    }

    return NULL;
}

/* prps 发送平台线程 */
void *_csg_send_prps_handle(void *arg)
{
    dau_send_msg_t *recv_msg = NULL;

    /* 等待初始化完成 */
    while(!is_system_init)
    {
        sleep(1);
    }

    while(1)
    {
        if (fifo_read(csg.fifo_prps_id, (void**)&recv_msg) != 0)
        {
            DBG(DBG_M_PD_DAU_ERR, "ERROR at fifo %d read!\r\n", csg.fifo_prps_id);
            continue;
        }

        /* 响应其他报文 */
        if (csg.send_cb)
        {
            csg.send_cb(recv_msg->type, recv_msg->slot, recv_msg->data);
        }

        /* 释放数据内存, 注意一定要在 fifo_push 之前调用, 因为 fifo_push 后 recv_msg 已被释放. */
        XFREE(MTYPE_CSG, recv_msg->data);
        fifo_push(csg.fifo_prps_id);
    }

    return NULL;
}

/* trend 发送平台线程 */
void *_csg_send_trend_handle(void *arg)
{
    dau_send_msg_t *recv_msg = NULL;

    /* 等待初始化完成 */
    while(!is_system_init)
    {
        sleep(1);
    }

    while(1)
    {
        if (fifo_read(csg.fifo_trend_id, (void**)&recv_msg) != 0)
        {
            DBG(DBG_M_PD_DAU_ERR, "ERROR at fifo %d read!\r\n", csg.fifo_trend_id);
            continue;
        }

        /* 响应其他报文 */
        if (csg.is_connect)
        {
            if (csg.send_cb)
            {
                csg.send_cb(recv_msg->type, recv_msg->slot, recv_msg->data);
            }
        }
        else
        {
            csg.trend_write_file_cb(recv_msg->slot, recv_msg->data);
        }

        /* 释放数据内存, 注意一定要在 fifo_push 之前调用, 因为 fifo_push 后 recv_msg 已被释放. */
        XFREE(MTYPE_CSG, recv_msg->data);
        fifo_push(csg.fifo_trend_id);
    }

    return NULL;
}

/* event 发送平台线程 */
void *_csg_send_event_handle(void *arg)
{
    dau_send_msg_t *recv_msg = NULL;

    /* 等待初始化完成 */
    while(!is_system_init)
    {
        sleep(1);
    }

    while(1)
    {
        if (fifo_read(csg.fifo_event_id, (void**)&recv_msg) != 0)
        {
            DBG(DBG_M_PD_DAU_ERR, "ERROR at fifo %d read!\r\n", csg.fifo_event_id);
            continue;
        }

        /* 响应其他报文 */
        if (csg.is_connect)
        {
            if (csg.send_cb)
            {
                csg.send_cb(recv_msg->type, recv_msg->slot, recv_msg->data);
            }
        }
        else
        {
            csg.event_write_file_cb(recv_msg->slot, recv_msg->data);
        }

        /* 释放数据内存, 注意一定要在 fifo_push 之前调用, 因为 fifo_push 后 recv_msg 已被释放. */
        XFREE(MTYPE_CSG, recv_msg->data);
        fifo_push(csg.fifo_event_id);
    }

    return NULL;
}

void *_csg_boosterpack_handle(void *arg)
{
    char filepath[260];
    int32_t len = 0;

    /* 等待初始化完成 */
    while(!is_system_init)
    {
        sleep(1);
    }
    
    while (1)
    {
        sleep(1);
        
        if (!csg.is_connect)
        {
            continue;
        }

        /* 事件重发 */
        len = sizeof(_csg_event);
        if (E_NONE == file_fifo_read(&csg.event_file, (char*)(&_csg_event), &len))
        {
            if (E_NONE == _csg_event_file_send())
            {
                file_fifo_delete_by_min_index(&csg.event_file);
            }
            continue;
        }

        /* 趋势重发 */
        snprintf(filepath, sizeof(filepath), "%s/%ld", csg.trend_file.dir, csg.trend_file.index_min);
        if (E_NONE == _csg_trend_file_send(filepath))
        {
            _csg_file_fifo_delete_by_min_index(&csg.trend_file);
        }
    }
    return NULL;
}


/* 后台接收处理函数. */
void *_csg_recv_handle(void *arg)
{
    struct sockaddr_in server;
    socklen_t server_len;
    int32_t addr = 0;
    uint16_t data_len = 0;

    /* 等待初始化完成 */
    while(!is_system_init)
    {
        sleep(1);
    }

    prctl(PR_SET_NAME, "CSG_RCVE", 0, 0, 0);

    while(1)
    {
        /* 读取数据. */
        data_len = recvfrom(csg.skfd, csg.buf_recv, CSG_PKT_LEN, 0, (struct sockaddr*)&server, &server_len);
        if (data_len <= 0)
        {
            DBG(DBG_M_PD_CSG_ERR, "Recvfrom return ERROR %s!\r\n", safe_strerror(errno));
            continue;
        }
        
        addr = server.sin_addr.s_addr;
        if (addr != csg.server_ip)
        {
            continue;
        }

        _csg_recv_process(csg.buf_recv, data_len);
    }

    return NULL;
}

void _csg_report_modbus()
{
    csg_modbus_iron_t iron;
    csg_modbus_magneto_t magneto;
    for (int i = 0; i < PD_SLOTS_MAX; i++)
    {
        if (modbus.card[i].is_connect == TRUE
                && modbus.card[i].is_insert == TRUE
                && modbus.card[i].type_m == DEVICE_M_TYPE)
        {
            switch (modbus.card[i].type_s)
            {
            case DEVICE_IR_TYPE:
                iron.current = modbus.card[i].current;
                iron.alarm = modbus.card[i].alarm;
                _csg_send_modbus_data(i, sizeof(iron), &iron);
                break;
            case DEVICE_DC_TYPE:
                magneto.current = modbus.card[i].current;
                _csg_send_modbus_data(i, sizeof(magneto), &magneto);
                break;
            default:
                break;
            }
        }
    }
}


/* 心跳和连接处理函数. */
void *_csg_heartbeat_handle(void *arg)
{
    time_t now = 0;
    time_t t_connect = 0;
    time_t t_heartbeat = 0;
    int count = 0;

    /* 等待初始化完成 */
    while(!is_system_init)
    {
        sleep(1);
    }
    
    while(1)
    {
        sleep(1);
        now = time(NULL);
        
        /* 发送连接报文. */
        if (!csg.is_connect)
        {
            if (abs(now - t_connect) >= 3)
            {
                _csg_connect_send();
                t_connect = now;
            }

            continue;
        }

        /* 发送心跳包. */
        if (abs(now - t_heartbeat) >= pd_config.config.heartbeat_period * 60)
        {
            _csg_heartbeat_send();
            t_heartbeat = now;
            csg.heartbeat_timeout_cnt++;
            /* 等待回复报文后再进行连接判断 */
            sleep(3);
            if (csg.heartbeat_timeout_cnt > 3)
            {
                csg.heartbeat_timeout_cnt = 0;
                _csg_disconnect_set(__FUNCTION__);
            }
        }

        for (int i = 0; i < PD_SLOTS_MAX; i++)
        {
            if (daus[i].state.is_connect == TRUE 
                && daus[i].state.is_insert != TRUE)
            {
                _csg_send_add_dau(i, &daus[i].info);
            }
            else if (modbus.card[i].is_connect == TRUE
                && modbus.card[i].is_insert != TRUE)
            {
                csg_contact_t info = {0};
                info.type_m = modbus.card[i].type_m;
                info.type_s = modbus.card[i].type_s;
                _csg_send_add_dau(i, &info);
            }
        }

        if (count++ > 60)
        {
            count = 0;
            _csg_report_modbus();
        }
    }

    return NULL;
}

/* 配置保存函数. */
int _csg_config_save(vty_t* vty)
{
    int16_t i = 0;
    struct in_addr addr;

    addr.s_addr = csg.server_ip;
    vty_out(vty, "csg server %s %d%s", inet_ntoa(addr), csg.server_port, VTY_NEWLINE);
    i++;

    return i;
}

/* Interface functions -------------------------------------------------------*/
/* 后台通讯模块预初始化. */
int32_t csg_handle_init(void)
{
    int32_t rv = 0;

    memset(&csg, 0, sizeof(csg_t));
    csg.trend_msg_id = -1;
    csg.event_msg_id = -1;
    csg.fifo_prps_id = E_MEM;
    csg.fifo_event_id = E_MEM;
    csg.fifo_trend_id = E_MEM;
    
    /* 发送数据. */
    csg.server_ip = inet_addr("192.168.1.161");
    csg.server_port = 1885;
    bzero(&csg.server, sizeof(csg.server));
    csg.server.sin_family = AF_INET;
    csg.server.sin_addr.s_addr = csg.server_ip;
    csg.server.sin_port = htons(csg.server_port);

    cmd_install_element(CONFIG_NODE, &csg_server_set_cmd);
    cmd_install_element(COMMON_NODE, &csg_show_cmd);

    /* 注册配置保存函数 */
    rv = cmd_config_node_config_register(CONFIG_PRI_CSG, _csg_config_save);
    if (rv != E_NONE)
    {
        log_err(LOG_CSG, "Command save register ERROR %d!", rv);
        return rv;
    }

    return E_NONE;
}

/* 后台通讯模块初始化. */
int32_t csg_handle_init_after(void)
{
    struct sockaddr_in server;
    int fd = 0;
    thread_param_t param = {0};
    
    if (pd_config.config.protocol_type != PD_PROTOCOL_LAND)
    {
        return E_NONE;
    }

    /* 创建协议 socket. */
    if (0 == csg.skfd)
    {
        /* 创建socket */
        fd = socket(AF_INET, SOCK_DGRAM, 0);
        if (fd < 0)
        {
            log_err(LOG_CSG, "ERROR at socket create return %s!", safe_strerror(errno));
            return E_SYS_CALL;
        }

        /* 绑定端口 */
        bzero(&server, sizeof(server));
        server.sin_family = AF_INET;
        server.sin_addr.s_addr = htonl(INADDR_ANY);
        server.sin_port = htons(7777);
        if(bind(fd, (struct sockaddr*)&server, sizeof(server)) < 0)
        {
            log_err(LOG_CSG, "ERROR at socket bind return %s!", safe_strerror(errno));
            close(fd);
            return E_SYS_CALL;
        }

        /* 保存数据. */
        csg.skfd = fd;
    }

    csg.fifo_cmd_id = fifo_create(CSG_FIFO_CMD, CSG_CMD_FIFO_NUM);
    if (csg.fifo_cmd_id < 0)
    {
        log_err(LOG_CSG, "Open fifo " CSG_FIFO_CMD " error!");
        return E_ERROR;
    }

    csg.fifo_prps_id = fifo_create(CSG_FIFO_PRPS, CSG_PRPS_FIFO_NUM);
    if (csg.fifo_prps_id < 0)
    {
        log_err(LOG_CSG, "Open fifo " CSG_FIFO_PRPS " error!");
        return E_ERROR;
    }

    csg.fifo_event_id = fifo_create(CSG_FIFO_EVENT, CSG_EVENT_FIFO_NUM);
    if (csg.fifo_event_id < 0)
    {
        log_err(LOG_CSG, "Open fifo " CSG_FIFO_EVENT " error!");
        return E_ERROR;
    }

    csg.fifo_trend_id = fifo_create(CSG_FIFO_TREND, CSG_TREND_FIFO_NUM);
    if (csg.fifo_trend_id < 0)
    {
        log_err(LOG_CSG, "Open fifo " CSG_FIFO_TREND " error!");
        return E_ERROR;
    }

    sem_init(&csg.event_sem, 0, 0);
    sem_init(&csg.trend_sem, 0, 0);
    sem_init(&csg.event_booster_sem, 0, 0);
    sem_init(&csg.trend_booster_sem, 0, 0);

    snprintf(csg.event_file.dir, FILE_FIFO_PATH_LEN, "%s", CSG_EVENT_NAME);
    csg.event_file.files_max = pd_config.config.event_storage;
    file_fifo_init(&csg.event_file);

    snprintf(csg.trend_file.dir, CSG_FILE_FIFO_PATH_LEN, "%s", CSG_TREND_NAME);
    csg.trend_file.files_max = pd_config.config.trend_storage;
    _csg_file_fifo_init(&csg.trend_file);

    csg.send_cb = _csg_send_process;
    csg.event_write_file_cb = _csg_event_write_file;
    csg.trend_write_file_cb = _csg_trend_write_file;
    
    param.arg = NULL;
    param.log_module = LOG_CSG;
    
    param.priority = 70;
    snprintf(param.thread_name, THREAD_NAME_LEN, "CSG_RCVE");
    create_thread(_csg_recv_handle, &param);

    param.priority = 55;
    snprintf(param.thread_name, THREAD_NAME_LEN, "CSG_HEARTBEAT");
    create_thread(_csg_heartbeat_handle, &param);

    param.priority = 60;
    snprintf(param.thread_name, THREAD_NAME_LEN, "CSG_SEND_CMD");
    create_thread(_csg_send_cmd_handle, &param);

    param.priority = 80;
    snprintf(param.thread_name, THREAD_NAME_LEN, "CSG_SEND_PRPS");
    create_thread(_csg_send_prps_handle, &param);

    param.priority = 70;
    snprintf(param.thread_name, THREAD_NAME_LEN, "CSG_SEND_TREND");
    create_thread(_csg_send_trend_handle, &param);

    param.priority = 75;
    snprintf(param.thread_name, THREAD_NAME_LEN, "CSG_SEND_EVENT");
    create_thread(_csg_send_event_handle, &param);

    return E_NONE;
}

/* description: 远程升级结果返回回调函数
   param: rv    -- 返回结果
          buf   -- 描述字符串
   return: */
void csg_upgrade_result_send(int32_t rv, char *buf)
{
    upgrade_res_t ack = {0};
    char *pkt = csg.buf_send;

    ack.result = rv;
    strcpy(ack.context, buf);
    
    memcpy(pkt + sizeof(csg_pkt_head_t), (char *)&ack, sizeof(csg_ack_t));
    _csg_send_data(CSG_REPLY, CSG_C_UPDATE_RESULT, pkt, sizeof(upgrade_res_t), 0);    
}

command_handler _csg_get_table_handle(command_entry *ptable, command_type cmd)
{
    if (NULL == ptable)
    {
        return NULL;
    }

    for (;CMD_INVALID != ptable->cmd; ptable++)
    {
        if (cmd == ptable->cmd)
        {
            return ptable->handler;
        }
    }
    return ptable->handler;
}

#endif
/************************ (C) COPYRIGHT LandPower ***** END OF FILE ****************/