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"

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

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

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

/* Private variables ---------------------------------------------------------*/
csg_t csg;

/* Private function prototypes -----------------------------------------------*/
void _csg_server_set(int32_t ip, uint16_t port);
void _csg_show();
void _csg_send_add_dau(uint8_t slot, void *data);
void _csg_send_reset(uint8_t, void *data);
void _csg_send_info_set(uint8_t, void *data);
void _csg_send_info_get(uint8_t, void *data);
void _csg_send_heartbeat(uint8_t slot, void *data);
void _csg_send_cfg_set(uint8_t, void *data);
void _csg_send_cfg_get(uint8_t, void *data);
void _csg_send_cfg_set_port(uint8_t, void *data);
void _csg_send_cfg_get_port(uint8_t, void *data);
void _csg_send_prps_get(uint8_t, void *data);
void _csg_send_trend_data(uint8_t, void *data);
void _csg_send_prps_data(uint8_t, void *data);
void _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
    NULL,                           // 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
    NULL,                           // 13
    _csg_send_prps_data,            // DAU_SEND_PRPS            14
    NULL,                           // 15  
};

/* 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;
}

/* 解析心跳报文. */
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_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;
    
    /* 首保处理, 打开文件描述符, 初始化变量 */
    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");
        rv = pd_upg_start(PD_UPG_FROM_CSG, head_msg->type);
    }
    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_event_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    //int slot = head->slot;

    printf("_csg_event_recv slot = %d\n", head->slot);
    head->dev_id = head->sdev_id;
    head->cmd = CSG_PRV_EVENT;

    //_dau_response(slot, pkt, head->len); 
    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:
            break;
        case CSG_PRV_REAL_PRPS:
            break;
        case CSG_PRV_EVENT:
            _csg_event_recv(pkt);
            break;
        default:
            break;
        }
    }
    return E_NONE;
}

void _csg_send_add_dau(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL == data)
    {
        return;
    }
    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);
}

void _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);
}

void _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);
}

void _csg_send_info_get(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL == data)
    {
        return;
    }
    
    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);
}

void _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;
    }
    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);
    }
}

void _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);
}

void _csg_send_cfg_get(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL == data)
    {
        return;
    }
    
    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);
}

void _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);
}

void _csg_send_cfg_get_port(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL == data)
    {
        return;
    }
    
    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);
}

void _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);
}

void _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;
    }

    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);    
    }    
}

void _csg_send_trend(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL != data)
    {
        return;
    }
    memcpy(pkt + sizeof(csg_pkt_head_t), data, sizeof(csg_ack_t));
    _csg_send_data(CSG_PRV_REPLY, CSG_PRV_TREND, pkt, sizeof(csg_ack_t), slot + 1);
}

void _csg_send_event(uint8_t slot, void *data)
{
    char *pkt = csg.buf_send;
    if (NULL != data)
    {
        return;
    }
    memcpy(pkt + sizeof(csg_pkt_head_t), data, sizeof(csg_ack_t));
    _csg_send_data(CSG_PRV_REPLY, CSG_PRV_EVENT, pkt, sizeof(csg_ack_t), slot + 1);
}

/* 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;
}

/* 普通命令发送平台线程 */
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.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_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.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_event_id);
    }

    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_heartbeat_handle(void *arg)
{
    time_t now = 0;
    time_t t_connect = 0;
    time_t t_heartbeat = 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);
            }
        }
    }

    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.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;
    }

    csg.send_cb = _csg_send_process;
    
    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 ****************/