/******************************************************************************
 * 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"

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

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

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

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

/* Private function prototypes -----------------------------------------------*/
extern void _csg_server_set(int32_t ip, uint16_t port);
void _csg_show();

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

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

/* 校验收到包的包头, 长度, 校验码. */
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)
{
    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++;
}

/* 数据发送 */
void _csg_send_data(uint8_t cmd_type, uint8_t cmd, char *pkt, int32_t len)
{
    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);

    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));
    uint8_t unit = 0;
    uint8_t port = 0;
    
    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;
    strncpy((char *)pinfo->app_version, host.version, sizeof(pinfo->app_version)-1);
    strncpy((char *)pinfo->app_compile_time, host.compile, sizeof(pinfo->app_compile_time)-1);
    strncpy((char *)pinfo->hardware_version, host.hardversion, 31);
    strncpy((char *)pinfo->FPGA_version, host.FPGAversion, 31);
    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, 0, sizeof(pinfo->port_type));
    for(unit = 0; unit < PD_DAU_SUM; unit++)
    {
        //if (!dau_is_valid(dau[unit]))
        {
            continue;
        }
        
        //for(port = 0; port < dau[unit]->port_num; port++)
        {
            pinfo->port[port] = pd_config.config_port[unit][port].vport;
            pinfo->port_type[port] = pd_config.config_port[unit][port].port_type;;
        }
    }

    _csg_send_data(CSG_REPLY, CSG_C_CONTACT, pkt, sizeof(csg_contact_t));

    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));    
    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;
    int slot = head->slot;

    printf("_csg_add_dau_recv slot = %d\n", head->slot);
    head->dev_id = head->sdev_id;
    head->cmd_type = CSG_REQUEST;
    head->cmd = CSG_C_CONTACT;
    head->len = CSG_HEAD_LEN + 4;
    //(uint32_t *)(pkt + CSG_HEAD_LEN) = time(NULL);
    uint32_t *timestamp = (uint32_t *)(pkt + CSG_HEAD_LEN);
    *timestamp = time(NULL);

    //_dau_response(slot, pkt, head->len);
    _dau_response(slot, pkt, head->len);
}

/* 解析心跳报文. */
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); //北京时间
    }
}

/* 设备重启报文. */
void _csg_reboot_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    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, head->cmd, pkt, sizeof(csg_ack_t));

    sleep(3);
    reboot_system(LOG_CSG, REBOOT_REMOTE_RESET);
}

/* 厂家参数设置报文处理. 
  说明：修改本地ip 设备ID 服务器地址时需要重启。
*/
int32_t _csg_dev_info_set_recv(char *pkt)
{
    int change_ip = 0;
    REBOOT_MSG boottype = REBOOT_NONE;
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    csg_dev_info_t *pinfo = (csg_dev_info_t *)(pkt + CSG_HEAD_LEN);
    device_info.dev_id = pinfo->dev_id;
    device_info.mask = pinfo->mask;
    device_info.gw = pinfo->gw;

    if (strncmp((char *)(pinfo->hostname), device_info.hostname, sizeof(device_info.hostname)))
    {
        snprintf((char*)device_info.hostname, PD_DEV_NUM_LEN, "%s", pinfo->hostname);
        boottype = REBOOT_REMOTE_HOST_NAME_CHANGE;
    }

    if (device_info.ip != pinfo->ip)
    {
        device_info.ip = pinfo->ip;  
        change_ip++;
        boottype = REBOOT_REMOTE_IP_CHANGE;
    }
    
    if (csg.server_ip != pinfo->server_ipv4)
    {
        csg.server_ip = pinfo->server_ipv4;
        boottype = REBOOT_REMOTE_SERVER_IP_CHANGE;
    }
    
    if (csg.server_port != pinfo->server_port)
    {
        csg.server_port = pinfo->server_port;
        boottype = REBOOT_REMOTE_SERVER_IP_CHANGE;
    }

    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, head->cmd, pkt, sizeof(csg_ack_t));

    vtysh_config_save();
    if (change_ip)
    {
        uint8_t mac[MAC_ADDR_LEN] = {0};
        mac_generate_from_ip(device_info.ip, mac);
        memcpy(device_info.mac, mac, MAC_ADDR_LEN);
        vtysh_eth0_save();
    }
    vtysh_device_save();
    
    if (boottype)
    {
        reboot_system(LOG_CSG, boottype);
    }
    
    return 0;
}

/* 厂家参数查询报文处理. */
int32_t _csg_dev_info_get_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    csg_dev_info_t *pinfo = (csg_dev_info_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;
    strcpy(pinfo->hostname, device_info.hostname);
    pinfo->factory_date = device_info.factory_date;
    pinfo->deployment_date = device_info.deployment_date;
    strncpy((char *)pinfo->app_compile_time, host.compile, 31);
    strncpy((char *)pinfo->app_version, host.version, 31);
    strncpy((char *)pinfo->hardware_version, host.hardversion, 31);
    strncpy((char *)pinfo->FPGA_version, host.FPGAversion, 31);
    pinfo->ip = device_info.ip;
    pinfo->mask = device_info.mask;
    pinfo->gw = device_info.gw;
    memcpy(pinfo->mac, device_info.mac, sizeof(pinfo->mac));
    //info.server_port = device_info.server_port;
    //info.server_ipv4 = device_info.server_ipv4;
    pinfo->server_port = csg.server_port;
    pinfo->server_ipv4 = csg.server_ip;

    _csg_send_data(CSG_REPLY, head->cmd, pkt, sizeof(csg_dev_info_t));

    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 *pnet = (csg_config_global_t *)(pkt + CSG_HEAD_LEN);
    
    pd_config.config.power_frequency = pnet->power_frequency;
    pd_config.config.trend_period = pnet->trend_period * 60;
    pd_config.config.sync_mode = pnet->sync_mode;
    pd_config.config.heartbeat_period = pnet->heartbeat_period;
    pd_config.config.pps_mode = pnet->pps_mode;
    pd_config.config.protocol_type = pnet->protocol_type;
    pd_config.config.event_storage = pnet->event_storage;
    pd_config.config.trend_storage = pnet->trend_storage;
        
    vtysh_config_save();

    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, head->cmd, pkt, sizeof(csg_ack_t));
    return E_NONE;
}

/* 查询用户参数查询报文处理. */
int32_t _csg_config_get_recv(char *pkt)
{
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    csg_config_global_t config = {0};
    config.power_frequency = pd_config.config.power_frequency;
    config.sync_mode = pd_config.config.sync_mode;
    config.heartbeat_period = pd_config.config.heartbeat_period;
    config.pps_mode = pd_config.config.pps_mode;
    config.protocol_type = pd_config.config.protocol_type;
    config.trend_period = pd_config.config.trend_period / 60;
    config.trend_storage = pd_config.config.trend_storage;
    config.event_storage = pd_config.config.event_storage;
    memcpy(pkt + sizeof(csg_pkt_head_t), (char *)&config, sizeof(csg_config_global_t));
    _csg_send_data(CSG_PRV_REPLY, head->cmd, pkt, sizeof(csg_config_global_t));

    return E_NONE;
}

/* 通道提交端口参数设置. */
int32_t _csg_port_config_set_recv(char *pkt)
{
    uint8_t vport = *(uint8_t*)(pkt + CSG_HEAD_LEN);
    uint8_t unit = 0;
    uint8_t port = 0;
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    csg_config_port_t *pnet = (csg_config_port_t *)(pkt + CSG_HEAD_LEN);
    
    pd_config.config_port[unit][port].vport = pnet->vport;
    pd_config.config_port[unit][port].port_type = pnet->port_type;
    pd_config.config_port[unit][port].filter = pnet->filter;
    pd_config.config_port[unit][port].sensor_type = pnet->sensor_type;
    pd_config.config_port[unit][port].is_auto_noise = pnet->is_auto_noise;
    pd_config.config_port[unit][port].denoise_type = pnet->denoise_type;
    pd_config.config_port[unit][port].denoise_variance = pnet->denoise_variance;
    
    pd_config.config_port[unit][port].event_counter_h = pnet->event_counter_h;
    pd_config.config_port[unit][port].event_sec_h = pnet->event_sec_h;
    pd_config.config_port[unit][port].event_thr_h = pnet->event_thr_h;
    pd_config.config_port[unit][port].event_counter_thr_h = pnet->event_counter_thr_h;
    
    pd_config.config_port[unit][port].event_counter_l = pnet->event_counter_l;
    pd_config.config_port[unit][port].event_sec_l = pnet->event_sec_l;
    pd_config.config_port[unit][port].event_thr_l = pnet->event_thr_l;
    pd_config.config_port[unit][port].event_counter_thr_l = pnet->event_counter_thr_l;
    
    pd_config.config_port[unit][port].burst_time = pnet->burst_time;
    pd_config.config_port[unit][port].burst_thr = pnet->burst_thr;
    pd_config.config_port[unit][port].denoise_manual = pnet->denoise_manual;
    pd_config.config_port[unit][port].denoise_auto = pnet->denoise_auto;

    pd_config.config_real[unit][port].filter_cfg = pd_config.config_port[unit][port].filter;
    //dau_port_filter_set(unit, port);

    vtysh_config_save();

    //csg_config_port_ack_t ack = {0};
    //ack.vport = pnet->vport;
    //ack.result = TRUE;
    //memcpy(pkt + sizeof(csg_pkt_head_t), (char *)&ack, sizeof(csg_config_port_ack_t));
    //_csg_send_data(CSG_PRV_REPLY, head->cmd, pkt, sizeof(csg_config_port_ack_t));
    return E_NONE;
}

/* 按通道提交端口参数查询结果. */
int32_t _csg_port_config_get_recv(char *pkt)
{
    uint8_t vport = *(uint8_t*)(pkt + CSG_HEAD_LEN);
    uint8_t unit = 0;
    uint8_t port = 0;
    csg_pkt_head_t *head = (csg_pkt_head_t*)pkt;
    csg_config_port_t *pnet = (csg_config_port_t *)(pkt + CSG_HEAD_LEN);

    //if (dau_vport_to_port(pnet->vport, &unit, &port) != E_NONE)
    {
        DBG(DBG_M_PD_CSG_ERR, "Pkt port %d error!\r\n", vport);
        return E_ERROR;
    }
    csg_config_port_t config = {0};
    config.vport = pd_config.config_port[unit][port].vport;
    config.port_type = pd_config.config_port[unit][port].port_type;
    config.filter = pd_config.config_port[unit][port].filter;
    config.sensor_type = pd_config.config_port[unit][port].sensor_type;
    config.is_auto_noise = pd_config.config_port[unit][port].is_auto_noise;
    config.denoise_type = pd_config.config_port[unit][port].denoise_type;
    config.denoise_variance = pd_config.config_port[unit][port].denoise_variance;
    
    config.event_counter_h = pd_config.config_port[unit][port].event_counter_h;
    config.event_sec_h = pd_config.config_port[unit][port].event_sec_h;
    config.event_thr_h = pd_config.config_port[unit][port].event_thr_h;
    config.event_counter_thr_h = pd_config.config_port[unit][port].event_counter_thr_h;
    
    config.event_counter_l = pd_config.config_port[unit][port].event_counter_l;
    config.event_sec_l = pd_config.config_port[unit][port].event_sec_l;
    config.event_thr_l = pd_config.config_port[unit][port].event_thr_l;
    config.event_counter_thr_l = pd_config.config_port[unit][port].event_counter_thr_l;
    
    config.burst_time = pd_config.config_port[unit][port].burst_time;
    config.burst_thr = pd_config.config_port[unit][port].burst_thr;
    config.denoise_manual = pd_config.config_port[unit][port].denoise_manual;
    config.denoise_auto = pd_config.config_port[unit][port].denoise_auto;

    memcpy(pkt + sizeof(csg_pkt_head_t), (char *)&config, sizeof(csg_config_port_t));

    _csg_send_data(CSG_PRV_REPLY, head->cmd, pkt, sizeof(pd_config_port_t));

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

    /* 如果升级线程开启失败返回错误信息. */
    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:
            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_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)
    {
        /* 读取数据. */
        memset(csg.buf_recv, 0, sizeof(csg.buf_recv));
        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__);
            }
        }
    }

    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;
    }
    
    param.arg = NULL;
    param.log_module = LOG_CSG;
    
    param.priority = 45;
    snprintf(param.thread_name, THREAD_NAME_LEN, "CSG_RCVE");
    create_thread(_csg_recv_handle, &param);

    param.priority = 45;
    snprintf(param.thread_name, THREAD_NAME_LEN, "CSG_HEARTBEAT");
    create_thread(_csg_heartbeat_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));    
}

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