Cable_v3.2_MQTT_TJ/app/lib/a_process/ca_db.c

/*****************************************************************************
 * file    lib/process/ca_db.c  
 * author  YuLiang
 * version 1.0.0
 * date    29-Nov-2023
 * brief   This file provides all the database related operation functions.
 ******************************************************************************
 * Attention
 *
 * <h2><center>&copy; COPYRIGHT(c) 2023 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_CA
/* 标准 C 库头文件 */
#include <stdarg.h>
#include <sys/types.h>
#include <sys/stat.h>
#include <execinfo.h>

/* 私有头文件 */
#include <sqlite3.h>

#include "cmd.h"
#include "fifo.h"
#include "ca_db.h"
#include "ca_collect.h"

/* Private define ------------------------------------------------------------*/
#define CA_DB_NAME_LEN 64
#define CA_DB_FIFO "CA_DB_FIFO"
#define CA_DB_BUF_LEN 2048
#define CA_DB_WR_CNT 500                // 写 500 次数据库进行一次数据库清理

#define CA_DB_LOCK pthread_mutex_lock(&ca_db_ctrl.mutex)
#define CA_DB_UNLOCK pthread_mutex_unlock(&ca_db_ctrl.mutex)

/* Private typedef -----------------------------------------------------------*/
/* fifo 数据结构 */
typedef struct
{
    uint32_t type;                      // 数据类型
    void *data;                         // 数据
} ca_db_msg_t;

/* 数据库模块属性结构体 */
typedef struct
{
    char filename[CA_DB_NAME_LEN];      // 数据库名字
    sqlite3 *db;                        // 数据库指针
    int32_t fifo_id;                    // fifo id
    int32_t cnt;                        // 写入计数
    char str[CA_DB_BUF_LEN];            // 操作 buf
    pthread_mutex_t mutex;              // 数据库读写互斥锁
} ca_db_ctrl_t;

/* Private function prototypes -----------------------------------------------*/

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

/* Private variables ---------------------------------------------------------*/
ca_db_ctrl_t ca_db_ctrl;                // 数据库全局控制结构体

/* Internal functions --------------------------------------------------------*/
/* description: 显示历史环流数据
   param:
   return: CMD_XXX */
CMD(ca_db_show_data,
    ca_db_show_data_cmd,
    "show db offset <1-5000> num <1-5000>",
    "Show\n"
    "Database\n"
    "Offset\n"
    "Offset index\n"
    "Number\n"
    "Number\n")
{
    uint16_t index = strtol(argv[0], NULL, 10) - 1;
    uint16_t num = strtol(argv[1], NULL, 10);
    
    ca_db_show_data_by_index(index, num);
    
    return CMD_SUCCESS;
}

/* description: 将数据按照格式串口打印
   param: argv - 数据内容
   return: E_XXX */
int _ca_db_show_data(void *data, int argc, char **argv, char **azColName)
{
    ca_coll_dev_data_t d;
    float temp = 0;
    uint8_t i = 0;
    uint8_t k = 0;

    k = 2;
    d.vin = strtol(argv[k++], NULL, 10);
    d.vbat = strtol(argv[k++], NULL, 10);
    d.vsc = strtol(argv[k++], NULL, 10);
    d.temperature = strtol(argv[k++], NULL, 10);
    d.run_time = strtol(argv[k++], NULL, 10);
    d.energy_mode = strtol(argv[k++], NULL, 10);
    d.CSQ = strtol(argv[k++], NULL, 10);

    for(i = 0; i < CA_COLL_ADC_CH_SUM; i++)
    {
        d.elec[i] = strtol(argv[k++], NULL, 10);
    }

    d.sen_short = strtol(argv[k++], NULL, 10);

    for(i = 0; i < CA_COLL_SENSOR_SUM; i++)
    {
        d.sen_temp[i] = strtol(argv[k++], NULL, 10);
        d.sen_x[i] = strtol(argv[k++], NULL, 10);
        d.sen_y[i] = strtol(argv[k++], NULL, 10);
        d.sen_z[i] = strtol(argv[k++], NULL, 10);
    }
    
    printh("ID %s UTC %s\r\n", argv[0], argv[1]);

    printh("VCC %-0.3fv VBAT %-0.3fv VSC %-0.3fv ", d.vin / 1000.0, d.vbat / 1000.0, d.vsc / 1000.0);
    printh("Temperature %-0.1f℃ Run %ds Mode %d ", d.temperature / 10.0, d.run_time,d.energy_mode);
    printh("Short %s\r\n", d.sen_short ? "TRUE" : "FALSE");
    
    printh("Electricity(v): ");
    for(i = 0; i < CA_COLL_ADC_CH_SUM - 2; i++)
    {
        printh("%-8.3f ", d.elec[i] / 1000.0);
    }
    printh("\r\n");
    
    printh("Sensor temperature(℃): ");
    for(i = 0; i < CA_COLL_SENSOR_SUM; i++)
    {
        temp = d.sen_temp[i] / 10.0;
        printh("%-5.1f ", temp);
    }
    printh("\r\n");

    printh("Sensor x: ");
    for(i = 0; i < CA_COLL_SENSOR_SUM; i++)
    {
        printh("%-5d ", d.sen_x[i]);
    }
    printh("\r\n");

    printh("Sensor y: ");
    for(i = 0; i < CA_COLL_SENSOR_SUM; i++)
    {
        printh("%-5d ", d.sen_y[i]);
    }
    printh("\r\n");

    printh("Sensor z: ");
    for(i = 0; i < CA_COLL_SENSOR_SUM; i++)
    {
        printh("%-5d ", d.sen_z[i]);
    }
    printh("\r\n\n");
    
    return 0;
}

/* description: 删除多余的数据保留 5000 条
   param: argv - 数据内容
   return: E_XXX */
int32_t _ca_db_clean(void)
{
    ca_db_ctrl_t *log = &ca_db_ctrl;
    char *zErrMsg = NULL;

    /* 删除小于比最大 id 小 5000 的数据, 即保留 id 最大的 5000 条数据 */
    snprintf(log->str, CA_DB_BUF_LEN, "delete from cable_data where id<(select Max(id) from cable_data)-5000;");
    if (sqlite3_exec(log->db, log->str, 0, 0, &zErrMsg) != SQLITE_OK)
    {
        DBG(DBG_M_CA_DB_ERR, "SQL DELETE ERR: %s\r\n", zErrMsg);
        sqlite3_free(zErrMsg);
        return E_SYS_CALL;
    }

    return E_NONE;
}

/* description: 将历史数据写入数据库
   param: data - 数据内容
   return: E_XXX */
int32_t _ca_db_data_write(ca_coll_dev_data_t *data)
{
    struct tm *day = NULL;
    char *zErrMsg = NULL;
    char *str = ca_db_ctrl.str;
    char time_str[TIME_STR_LEN] = {0};
    uint16_t len = 0;
    time_t utc = 0;

    utc = time(NULL);
    day = localtime(&utc);
    strftime(time_str, TIME_STR_LEN, "%Y/%m/%d %H:%M:%S", day);

    CA_DB_LOCK;

    len = snprintf(str, CA_DB_BUF_LEN, "INSERT INTO cable_data VALUES (NULL, \"%s\", %d, %d, %d, %d, %d, %d, %d, ", \
        time_str, data->vin, data->vbat, data->vsc, data->temperature, data->run_time, data->energy_mode, data->CSQ);
    len += snprintf(str + len, CA_DB_BUF_LEN - len, "%d, %d, %d, %d, %d, %d, %d, %d, ", \
        data->elec[0], data->elec[1], data->elec[2], data->elec[3], data->elec[4], data->elec[5], data->elec[6], data->elec[7]);
    len += snprintf(str + len, CA_DB_BUF_LEN - len, "%d, %d, %d, %d, %d, %d, %d, %d, %d, ", \
        data->sen_short, data->sen_temp[0], data->sen_x[0], data->sen_y[0], data->sen_z[0], data->sen_temp[1], data->sen_x[1], data->sen_y[1], data->sen_z[1]);
    len += snprintf(str + len, CA_DB_BUF_LEN - len, "%d, %d, %d, %d, %d, %d, %d, %d, ", \
        data->sen_temp[2], data->sen_x[2], data->sen_y[2], data->sen_z[2], data->sen_temp[3], data->sen_x[3], data->sen_y[3], data->sen_z[3]);
    len += snprintf(str + len, CA_DB_BUF_LEN - len, "%d, %d, %d, %d, %d, %d, %d, %d, ", \
        data->sen_temp[4], data->sen_x[4], data->sen_y[4], data->sen_z[4], data->sen_temp[5], data->sen_x[5], data->sen_y[5], data->sen_z[5]);
    len += snprintf(str + len, CA_DB_BUF_LEN - len, "%d, %d, %d, %d, %d, %d, %d, %d);", \
        data->sen_temp[6], data->sen_x[6], data->sen_y[6], data->sen_z[6], data->sen_temp[7], data->sen_x[7], data->sen_y[7], data->sen_z[7]);
    
    if (sqlite3_exec(ca_db_ctrl.db, str, 0, 0, &zErrMsg ) != SQLITE_OK)
    {
        DBG(DBG_M_CA_DB_ERR, "SQL INSERT error: %s\r\n", zErrMsg);
        sqlite3_free(zErrMsg);
        return E_SYS_CALL;
    }
        
    /* 保证数据库大小不超过 5000 */
    ca_db_ctrl.cnt++;
    if (ca_db_ctrl.cnt >= CA_DB_WR_CNT)
    {
        _ca_db_clean();
        ca_db_ctrl.cnt = 0;
    }

    CA_DB_UNLOCK;

    return E_NONE;
}

/* description: 写数据库线程
   param:
   return: */
void *_ca_db_handle_write(void *arg)
{
    ca_db_msg_t *recv_msg = NULL;

    while(1)
    {
        /* 等待数据 */
        if (fifo_read(ca_db_ctrl.fifo_id, (void **)&recv_msg) != 0)
        {
            DBG(DBG_M_CA_DB_ERR, "ERROR at fifo read!\r\n");
            continue;
        }

        /* 处理数据 */
        switch(recv_msg->type)
        {
        case CA_DB_T_DATA:
            _ca_db_data_write((ca_coll_dev_data_t*)recv_msg->data);
            break;
        default:
            break;
        }

        /* 释放数据 */
        XFREE(MTYPE_CA_DB, recv_msg->data);
        fifo_push(ca_db_ctrl.fifo_id);
    }
    return NULL;
}

/* Interface functions -------------------------------------------------------*/
/* description: 数据库处理预初始化.
   param:
   return: E_XXX */
int32_t ca_db_init(void)
{
    ca_db_ctrl_t *log = &ca_db_ctrl;
    int32_t rv = 0;
    char *zErrMsg = NULL;

    /* 打开数据库 */
    snprintf(log->filename, CA_DB_NAME_LEN, CA_DB_FILE);
    rv = sqlite3_open(log->filename, &log->db);
    if (rv)
    {
        log->db = NULL;
        log_err(LOG_CA_DB, "Can't open database!");
        return E_SYS_CALL;
    }

    /* 创建历史数据表 */
    snprintf(log->str, CA_DB_BUF_LEN, \
        "CREATE TABLE IF NOT EXISTS cable_data(ID INTEGER PRIMARY KEY AUTOINCREMENT," \
        "UTC TEXT, VIN INTEGER, VBAT INTEGER, VSC INTEGER, TEMP INTEGER, RUN INTEGER, MODE INTEGER, CSQ INTEGER," \
        "ELEC1 INTEGER, ELEC2 INTEGER, ELEC3 INTEGER, ELEC4 INTEGER, ELEC5 INTEGER, ELEC6 INTEGER, ELEC7 INTEGER, ELEC8 INTEGER," \
        "SEN_S INTEGER, SEN1_T INTEGER, SEN1_X INTEGER, SEN1_Y INTEGER, SEN1_Z INTEGER, SEN2_T INTEGER, SEN2_X INTEGER, SEN2_Y INTEGER, SEN2_Z INTEGER," \
        "SEN3_T INTEGER, SEN3_X INTEGER, SEN3_Y INTEGER, SEN3_Z INTEGER, SEN4_T INTEGER, SEN4_X INTEGER, SEN4_Y INTEGER, SEN4_Z INTEGER," \
        "SEN5_T INTEGER, SEN5_X INTEGER, SEN5_Y INTEGER, SEN5_Z INTEGER, SEN6_T INTEGER, SEN6_X INTEGER, SEN6_Y INTEGER, SEN6_Z INTEGER," \
        "SEN7_T INTEGER, SEN7_X INTEGER, SEN7_Y INTEGER, SEN7_Z INTEGER, SEN8_T INTEGER, SEN8_X INTEGER, SEN8_Y INTEGER, SEN8_Z INTEGER);");
    if (sqlite3_exec(log->db, log->str, 0, 0, &zErrMsg) != SQLITE_OK)
    {
        log_err(LOG_CA_DB, "SQL create error: %s\r\n", zErrMsg);
        sqlite3_free(zErrMsg);
        return E_SYS_CALL;
    }

    /* 维持数据库大小 */
    _ca_db_clean();

    /* 添加命令行 */
    cmd_install_element(COMMON_NODE, &ca_db_show_data_cmd);
    
    return E_NONE;
}

/* description: 数据库处理初始化.
   param:
   return: E_XXX */
int32_t ca_db_init_after(void)
{
    struct sched_param param;
    pthread_attr_t attr;
    pthread_t pid;
    ca_db_ctrl_t *db = &ca_db_ctrl;
    int rv = 0;

    /* 舒适化线程锁, 保证只有数据库单线程操作 */
    pthread_mutex_init(&db->mutex, NULL);

    /* 初始化 fifo */
    db->fifo_id = fifo_create(CA_DB_FIFO, 32);
    if (db->fifo_id < 0)
    {
        log_err(LOG_CA_DB, "Open fifo " CA_DB_FIFO " error.");
        return E_ERROR;
    }

    /* 创建数据库造作线程, 配置线程RR调度，优先级25 */
    pthread_attr_init(&attr);
    param.sched_priority = 25;
    pthread_attr_setschedpolicy(&attr, SCHED_RR);
    pthread_attr_setschedparam(&attr, &param);
    pthread_attr_setinheritsched(&attr, PTHREAD_EXPLICIT_SCHED);
    rv = pthread_create(&pid, &attr, _ca_db_handle_write, NULL);
    if (rv != 0)
    {
        log_err(LOG_CA_DB, "Can't create cable db pthread %d.", rv);
        return E_SYS_CALL;
    }
    else
    {
        thread_m_add("CA_DB_THREAD", pid);
    }
    pthread_attr_destroy(&attr);

    return E_NONE;
}

/* description: 写数据库
   param: type - 操作类型
          data - 操作数据指针
   return: E_XXX */
int32_t ca_db_write(uint32_t type, void *data)
{
    ca_db_msg_t msg;
    void *pkt = NULL;

    /* 申请报文空间 */
    if (CA_DB_T_DATA == type)
    {
        pkt = XMALLOC_Q(MTYPE_CA_DB, sizeof(ca_coll_dev_data_t));
        if (!pkt)
        {
            DBG(DBG_M_CA_DB_ERR, "Malloc ERR!\r\n");
            return E_MEM;
        }

        /* 封装消息. */
        memcpy(pkt, data, sizeof(ca_coll_dev_data_t));
    }

    msg.type = type;
    msg.data = pkt;

    if (fifo_write(ca_db_ctrl.fifo_id, (void*)(&msg), sizeof(ca_db_msg_t)) != sizeof(ca_db_msg_t))
    {
        DBG(DBG_M_CA_DB_ERR, "Fifo write ERROR\r\n");
        XFREE(MTYPE_CA_DB, pkt);
        return E_ERROR;
    }

    return E_NONE;
}

/* description: 显示据库数据
   param: index - 偏移
          num - 条目
   return: */
void ca_db_show_data_by_index(uint16_t index, uint16_t num)
{
    ca_db_ctrl_t *log = &ca_db_ctrl;
    char *zErrMsg = NULL;

    CA_DB_LOCK;

    /* 按最新数据排序(id 递减), 读取偏移 index, 数量 num 条数据 */
    snprintf(log->str, CA_DB_BUF_LEN, "SELECT * FROM cable_data ORDER BY id DESC LIMIT %d OFFSET %d;", num, index);
    
    if (sqlite3_exec(log->db, log->str, _ca_db_show_data, 0, &zErrMsg) != SQLITE_OK)
    {
        DBG(DBG_M_CA_DB_ERR, "SQL show error: %s\r\n", zErrMsg);
        sqlite3_free(zErrMsg);
    }

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