#include "uart.hpp" #include #include #include #include #include #include #include #include #include #include "mqttclient/mqtt_client.h" #include "minilzo/minilzo.h" #include "jsonparse/communication_cmd.hpp" extern zlog_category_t *zct; extern zlog_category_t *zbt; std::vector g_VecWaveDataX; std::vector g_VecWaveDataY; std::vector g_VecWaveDataZ; std::map g_mapCompress; std::map g_mapWaveChannel; unsigned char data[96000] = {0x00}; unsigned char outdata[96000] = {0x00}; unsigned char dealdata[96000] = {0x00}; char mqttData[512000] = {0}; void Uart::RecordBattery(std::string &strLongAddr, DataRecvStatic &dataStatic, std::string &nowTimetamp) { char insertSql[1024] = {0}; sprintf(insertSql, "'%s','%d','%f','%f','%f','%d','','','%s'", strLongAddr.c_str(), dataStatic.Dip, dataStatic.TemBot, dataStatic.nodeWorkTime, dataStatic.nodeSendTime, dataStatic.Voltage, nowTimetamp.c_str()); sqlite_db_ctrl::instance().InsertData(T_BATTERY_INFO(TNAME), insertSql); } void Uart::DataExtract(RecvData *p, int id, unsigned int &lowbit, float &n) { char buf[20] = {0}; sprintf(buf, "%02x%02x", p->Data[id+1], p->Data[id]); int temp = (unsigned int)strtol(buf, NULL, 16); unsigned char highbit = temp >> 14 & 0x3; lowbit = temp & 0x3fff; n = ScaleConvert(highbit); } void Uart::DealDataNodeFeature(const char *pData, int flag) { RecvData *pRecvData = (RecvData *)pData; char whereCon[1024] = {0}; char updateSql[1024] = {0}; char buf[20] = {0x00}; int nodeResend = 0, timing = 0; long staticIndex = 0; sprintf(buf, "%02x%02x", pRecvData->ShortAddr[0], pRecvData->ShortAddr[1]); if (flag == 1) { zlog_info(zct, "DealDataNodeFeature %02x%02x, %d", pRecvData->ShortAddr[0], pRecvData->ShortAddr[1], flag); } if (bSendTimeStamp) //波形处理中 return; std::string strShortAddr = std::string(buf); zlog_info(zct, "zigbeeShortAddr='%s'", strShortAddr.c_str()); char getLongAddr_sql[32] = {0}; //根据数据包中的传感器的短地址获取数据库中长地址(MAC),在下面判断该传感器是否存在,如果不存在则把数据包丢弃 sprintf(getLongAddr_sql, "zigbeeShortAddr='%s'", strShortAddr.c_str()); vec_t vecResult = sqlite_db_ctrl::instance().GetDataSingleLine(T_SENSOR_INFO(TNAME), " softVersion,dataNodeNo,MeasurementID,ProductNo ", getLongAddr_sql); if (vecResult.size() < 1) { zlog_warn(zct, "device info not found %02x%02x ", pRecvData->ShortAddr[0], pRecvData->ShortAddr[1]); return; } zlog_info(zct, "--------->the remote sensor short addr:%s strLongAddr=%s,softVersion = %s", buf, vecResult[1].c_str(), vecResult[0].c_str()); std::string strLongAddr = vecResult[1]; std::string strMeasurementID = vecResult[2]; std::string strProductNo = vecResult[3]; memset(whereCon, 0x00, sizeof(whereCon)); memset(updateSql, 0x00, sizeof(updateSql)); sprintf(whereCon, "zigbeeShortAddr='%s'", strShortAddr.c_str()); sprintf(updateSql, " StaticTime = StaticTime + 1"); sqlite_db_ctrl::instance().UpdateTableData(T_SENSOR_INFO(TNAME), updateSql, whereCon); if (vecResult[0] == "3.0" || vecResult[0] == "4.0") { return; } long nodetimestamp = BUILD_UINT32(pRecvData->Data[29], pRecvData->Data[28], pRecvData->Data[27], pRecvData->Data[26]); char localtimestamp[32] = {0}; sprintf(localtimestamp,"%ld",nodetimestamp); std::string nowTimetamp = std::string(localtimestamp); strTimetamp = nowTimetamp; int iTemp = 0; // unsigned char highbit = 0; unsigned int lowbit = 0; float n = 0; DataRecvStatic dataStatic; memset(buf, 0, sizeof(buf)); sprintf(buf, "%02x%02x", pRecvData->Data[5], pRecvData->Data[4]); iTemp = (int)strtol(buf, NULL, 16); dataStatic.Dip = iTemp; int fTemp = 0; memset(buf, 0, sizeof(buf)); sprintf(buf, "%02x%02x", pRecvData->Data[1], pRecvData->Data[0]); iTemp = (int)strtol(buf, NULL, 16); if (iTemp < 0x8000) { fTemp = iTemp; } else { fTemp = (((~iTemp) & 0xffff) + 1) * (-1); } dataStatic.TemBot = fTemp * 0.0625; //设备温度 fTemp = 0; memset(buf, 0, sizeof(buf)); sprintf(buf, "%02x%02x", pRecvData->Data[3], pRecvData->Data[2]); iTemp = (int)strtol(buf, NULL, 16); if (iTemp < 0x8000) { fTemp = iTemp; } else { fTemp = (((~iTemp) & 0xffff) + 1) * (-1); } dataStatic.TemTop = fTemp * 0.0625; //环境温度 memset(buf, 0, sizeof(buf)); sprintf(buf, "%02x%02x", pRecvData->Data[7], pRecvData->Data[6]); iTemp = (int)strtol(buf, NULL, 16); dataStatic.Voltage = iTemp; DataExtract(pRecvData, 30, lowbit, n); dataStatic.nodeWorkTime = lowbit * n; DataExtract(pRecvData, 32, lowbit, n); dataStatic.nodeSendTime = lowbit * n; dataStatic.nodeWorkTime = dataStatic.nodeWorkTime - dataStatic.nodeSendTime; RecordBattery(strMeasurementID, dataStatic, nowTimetamp); char szTableName[50] = {0x00}, szTableNameStatic[50] = {0x00}, szTableNameData[50] = {0x00}; sprintf(szTableName, "t_dataStatic_%s", strMeasurementID.c_str()); memcpy(szTableNameStatic, szTableName, sizeof(szTableNameStatic)); memset(whereCon, 0x00, sizeof(whereCon)); // sprintf(whereCon, "StaticIndex = %ld", staticIndex); // int count = sqlite_db_ctrl::instance().GetTableRows(szTableNameStatic, whereCon); //避免重复数据 // sprintf(szTableNameData, "t_data_%s", strMeasurementID.c_str()); // int count2 = sqlite_db_ctrl::instance().GetTableRows(szTableNameData, whereCon); // if (count > 0 || count2 > 0) { // char logInfo[100] = {0x00}; // sprintf(logInfo, "ShortAddr = %s,staticIndex = %ld,staticData = %d, data = %d", strShortAddr.c_str(), staticIndex, count, count2); // zlog_info(zct, logInfo); // return; // } memset(whereCon, 0x00, sizeof(whereCon)); sprintf(szTableNameData, "t_data_%s", strMeasurementID.c_str()); ///////////////////////////////////////////////////////////// for V2.0.3 upgrade to V3.0 ////////////////////////////////////////////////////////////更换电池判断 // sprintf(whereCon, " dataNodeNo = '%s' and StaticIndex > 0 order by StaticIndex desc LIMIT 0 , 1 ", strLongAddr.c_str()); // std::string strStaticIndex = sqlite_db_ctrl::instance().GetData(szTableNameStatic, "StaticIndex", whereCon); // if (atol(strStaticIndex.c_str()) - staticIndex > 100) { // sqlite_db_ctrl::instance().Deletetable(szTableNameStatic); // sqlite_db_ctrl::instance().Deletetable(szTableNameData); // zlog_info(zct, "staticIndexNOW = %ld,strStaticIndexLast = %s", staticIndex, strStaticIndex.c_str()); // } // zlog_info(zct, "NowstaticIndex = %ld,RecordStaticIndex = %ld", staticIndex, atol(strStaticIndex.c_str())); // if (staticIndex != atol(strStaticIndex.c_str() + 1) && strStaticIndex != "" && staticIndex < atol(strStaticIndex.c_str())) { // sprintf(whereCon, "StaticIndex = %ld order by StaticIndex desc LIMIT 0 , 1", atol(strStaticIndex.c_str())); // vec_t vecResult = sqlite_db_ctrl::instance().GetDataSingleLine(szTableNameStatic, "timeStamp,StaticIndex", whereCon); // if (vecResult.size() > 0) { // memset(whereCon, 0x00, sizeof(whereCon)); // sprintf(whereCon, "dataNodeNo = '%s'", strLongAddr.c_str()); // std::string staticInterval = sqlite_db_ctrl::instance().GetData(T_SENSOR_INFO(TNAME), "featureInterval", whereCon); // long nNowTimetamp = atol(vecResult[0].c_str()) - (atol(staticInterval.c_str()) * (atol(vecResult[1].c_str()) - staticIndex)) * 60; // char tmp[10] = {0x00}; // sprintf(tmp, "%ld", nNowTimetamp); // nowTimetamp = std::string(tmp); // nodeResend = 1; // } // } zlog_info(zct, "nowTimetamp = %s", nowTimetamp.c_str()); sprintf(whereCon, "channelID='%s' ", (strMeasurementID + "-S").c_str()); int Count = sqlite_db_ctrl::instance().GetTableRows(szTableName, whereCon); if (Count == -1) { sqlite_db_ctrl::instance().CreatedataStatictable(szTableName); sqlite_db_ctrl::instance().Createtable(szTableNameData); } std::string strTmp = ""; char sztmp[100] = {0x00}; strTmp = "name = '" + std::string(szTableNameStatic) + "' and sql LIKE '%nodeResend%' "; int row = sqlite_db_ctrl::instance().GetTableRows(" sqlite_master ", strTmp.c_str()); zlog_info(zct, "row1 = %d", row); if (row == 0) { memset(sztmp, 0x00, sizeof(sztmp)); sprintf(sztmp, "ALTER TABLE %s ADD COLUMN 'nodeResend'", szTableNameStatic); sqlite_db_ctrl::instance().CreateTable(sztmp); } strTmp = "name = '" + std::string(szTableNameData) + "' and sql LIKE '%nodeResend%' "; row = sqlite_db_ctrl::instance().GetTableRows(" sqlite_master ", strTmp.c_str()); zlog_info(zct, "row2 = %d", row); if (row == 0) { memset(sztmp, 0x00, sizeof(sztmp)); sprintf(sztmp, "ALTER TABLE %s ADD COLUMN 'nodeResend'", szTableNameData); sqlite_db_ctrl::instance().CreateTable(sztmp); } char selectCon[128] = {0}; sprintf(selectCon, "channelID='%s' ORDER BY timeStamp ASC LIMIT 0,1", (strMeasurementID + "-S").c_str()); std::string strTime = sqlite_db_ctrl::instance().GetData(szTableName, "timeStamp", selectCon); zlog_info(zct, "strLongAddr = %s,strTime = %s", strLongAddr.c_str(), strTime.c_str()); long lTime = atol(nowTimetamp.c_str()) - atol(strTime.c_str()); zlog_info(zct, "lTime = %ld,OneWeek = %d", lTime, OneWeek); zlog_info(zct, "dataStatic.TemTop : %f dataStatic.TemBot : %f dataStatic.Dip :%d dataStatic.Voltage : %d", dataStatic.TemTop, dataStatic.TemBot, dataStatic.Dip, dataStatic.Voltage); sprintf(updateSql, "temTop='%f',temBot='%f',dip='%d',voltage='%d',timeStamp='%s',StaticIndex = %ld, nodeResend = %d,zigbeeSignal = '',zigbeeSignalNode = '',statisticType = '%d',timing = '%d' ", dataStatic.TemTop, dataStatic.TemBot, dataStatic.Dip, dataStatic.Voltage, nowTimetamp.c_str(), staticIndex, nodeResend, flag, timing); sprintf(whereCon, "channelID='%s' ", (strMeasurementID + "-S").c_str()); if (/*0 == sqlite_db_ctrl::instance().GetTableRows(T_DATASTATIC_INFO(TNAME), whereCon)*/ (Count * 3 < SAVE_COUNT && lTime < OneWeek) || strTime.size() == 0) { zlog_info(zct, "insert static data to sql"); char insertSql[1024] = {0}; sprintf(insertSql, "'%s','%s','%f','%f','%d','%d','',%ld,'%s','1',%d,'','%d','%d'", strMeasurementID.c_str(), (strMeasurementID + "-S").c_str(), dataStatic.TemTop, dataStatic.TemBot, dataStatic.Dip, dataStatic.Voltage, staticIndex, nowTimetamp.c_str(), nodeResend, flag, timing); sqlite_db_ctrl::instance().InsertData(szTableName, insertSql); if (0 == sqlite_db_ctrl::instance().GetTableRows(T_DATASTATIC_INFO(TNAME), whereCon)) { // First Connect char insertSql[1024] = {0}; sprintf(insertSql, "'%s','%s','%f','%f','%d','%d','',%ld,'%s','1',%d", strMeasurementID.c_str(), (strMeasurementID + "-S").c_str(), dataStatic.TemTop, dataStatic.TemBot, dataStatic.Dip, dataStatic.Voltage, staticIndex, nowTimetamp.c_str(), nodeResend); sqlite_db_ctrl::instance().InsertData(T_DATASTATIC_INFO(TNAME), insertSql); sqlite_db_ctrl::instance().CalculateBattery(); } else { memset(updateSql, 0x00, sizeof(updateSql)); sprintf(updateSql, "temTop='%f',temBot='%f',dip='%d',voltage='%d',timeStamp='%s',StaticIndex = %ld ", dataStatic.TemTop, dataStatic.TemBot, dataStatic.Dip, dataStatic.Voltage, nowTimetamp.c_str(), staticIndex); sqlite_db_ctrl::instance().UpdateTableData(T_DATASTATIC_INFO(TNAME), updateSql, whereCon); } } else { memset(whereCon, 0x00, sizeof(whereCon)); sprintf(whereCon, "channelID='%s' and timeStamp = '%s'", (strMeasurementID + "-S").c_str(), strTime.c_str()); zlog_info(zct, "update static data to sql"); sqlite_db_ctrl::instance().UpdateTableData(szTableName, updateSql, whereCon); memset(whereCon, 0x00, sizeof(whereCon)); sprintf(whereCon, "channelID='%s' ", (strMeasurementID + "-S").c_str()); memset(updateSql, 0x00, sizeof(updateSql)); sprintf(updateSql, "temTop='%f',temBot='%f',dip='%d',voltage='%d',timeStamp='%s',StaticIndex = %ld ", dataStatic.TemTop, dataStatic.TemBot, dataStatic.Dip, dataStatic.Voltage, nowTimetamp.c_str(), staticIndex); sqlite_db_ctrl::instance().UpdateTableData(T_DATASTATIC_INFO(TNAME), updateSql, whereCon); } memset(szTableName, 0x00, sizeof(szTableName)); sprintf(szTableName, "t_data_%s", strMeasurementID.c_str()); DataRecvDym dataDymX; DataExtract(pRecvData, 8, lowbit, n); dataDymX.DiagnosisPk = lowbit * n; DataExtract(pRecvData, 10, lowbit, n); dataDymX.RmsValues = lowbit * n; DataExtract(pRecvData, 12, lowbit, n); dataDymX.IntegratPk = lowbit * n; DataExtract(pRecvData, 14, lowbit, n); dataDymX.IntegratRMS = lowbit * n; DataExtract(pRecvData, 16, lowbit, n); dataDymX.Amp1 = lowbit * n; DataExtract(pRecvData, 18, lowbit, n); dataDymX.Amp2 = lowbit * n; DataExtract(pRecvData, 20, lowbit, n); dataDymX.Amp3 = lowbit * n; DataExtract(pRecvData, 22, lowbit, n); dataDymX.Amp4 = lowbit * n; DataExtract(pRecvData, 24, lowbit, n); dataDymX.Amp5 = lowbit * n; memset(buf, 0, sizeof(buf)); dataDymX.EnvelopEnergy = 0; memset(buf, 0, sizeof(buf)); dataDymX.Phase1 = 0; memset(buf, 0, sizeof(buf)); dataDymX.Phase2 = 0; memset(buf, 0, sizeof(buf)); memset(buf, 0, 8); dataDymX.Phase3 = 0; memset(buf, 0, sizeof(buf)); sprintf(buf, "%02x%02x", pRecvData->Data[35], pRecvData->Data[34]); dataDymX.Phase4 = 0; memset(whereCon, 0, 1024); sprintf(whereCon, "channelID='%s' ", (strMeasurementID + "-X").c_str()); memset(updateSql, 0, 1024); sprintf(updateSql, "diagnosisPk='%f',integratPk='%f',integratRMS='%f',rmsValues='%f',envelopEnergy='%f',\ Amp1='%f',Amp2='%f',Amp3='%f',Amp4='%f',Amp5='%f',Phase1='%f',Phase2='%f',Phase3='%f',Phase4='%f',timeStamp='%s',StaticIndex = %ld,nodeResend = %d ", dataDymX.DiagnosisPk, dataDymX.IntegratPk, dataDymX.IntegratRMS, dataDymX.RmsValues, dataDymX.EnvelopEnergy, dataDymX.Amp1, dataDymX.Amp2, dataDymX.Amp3, dataDymX.Amp4, dataDymX.Amp5, dataDymX.Phase1, dataDymX.Phase2, dataDymX.Phase3, dataDymX.Phase4, nowTimetamp.c_str(), staticIndex, nodeResend); if ((Count * 3 < SAVE_COUNT && lTime < OneWeek) || strTime.size() == 0) { // 1 week char insertSql[1024] = {0}; memset(insertSql, 0x00, sizeof(insertSql)); sprintf(insertSql, "'%s','%s','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f',%ld,'%s','1',%d", strMeasurementID.c_str(), (strMeasurementID + "-X").c_str(), dataDymX.DiagnosisPk, dataDymX.IntegratPk, dataDymX.IntegratRMS, dataDymX.RmsValues, dataDymX.EnvelopEnergy, dataDymX.Amp1, dataDymX.Amp2, dataDymX.Amp3, dataDymX.Amp4, dataDymX.Amp5, dataDymX.Phase1, dataDymX.Phase2, dataDymX.Phase3, dataDymX.Phase4, staticIndex, nowTimetamp.c_str(), nodeResend); sqlite_db_ctrl::instance().InsertData(szTableName, insertSql); if (0 == sqlite_db_ctrl::instance().GetTableRows(T_DATA_INFO(TNAME), whereCon)) sqlite_db_ctrl::instance().InsertData(T_DATA_INFO(TNAME), insertSql); else sqlite_db_ctrl::instance().UpdateTableData(T_DATA_INFO(TNAME), updateSql, whereCon); } else { memset(whereCon, 0x00, sizeof(whereCon)); sprintf(whereCon, "channelID='%s' and timeStamp = '%s'", (strMeasurementID + "-X").c_str(), strTime.c_str()); sqlite_db_ctrl::instance().UpdateTableData(szTableName, updateSql, whereCon); memset(whereCon, 0x00, sizeof(whereCon)); sprintf(whereCon, "channelID='%s' ", (strMeasurementID + "-X").c_str()); sqlite_db_ctrl::instance().UpdateTableData(T_DATA_INFO(TNAME), updateSql, whereCon); } zlog_info(zct, "x:%s,%s,diagnosisPk=%f,integratPk=%f,integratRMS=%f,rmsValues=%f,envelopEnergy=%f,Amp1=%f,Amp2=%f,Amp3=%f,Amp4=%f,Amp5=%f,Phase1=%f,Phase2=%f,Phase3=%f,Phase4=%f,timeStamp=%s", strMeasurementID.c_str(), (strMeasurementID + "-X").c_str(), dataDymX.DiagnosisPk, dataDymX.IntegratPk, dataDymX.IntegratRMS, dataDymX.RmsValues, dataDymX.EnvelopEnergy, dataDymX.Amp1, dataDymX.Amp2, dataDymX.Amp3, dataDymX.Amp4, dataDymX.Amp5, dataDymX.Phase1, dataDymX.Phase2, dataDymX.Phase3, dataDymX.Phase4, nowTimetamp.c_str()); Json::Value valNodeData; Json::Value valNodeFeature; valNodeFeature["dataNodeNo"] = strMeasurementID; valNodeFeature["ChannelId"] = strMeasurementID + "-X"; valNodeFeature["diagnosisPk"] = dataDymX.DiagnosisPk; valNodeFeature["integratPk"] = dataDymX.IntegratPk; valNodeFeature["integratRMS"] = dataDymX.IntegratRMS; valNodeFeature["rmsValues"] = dataDymX.RmsValues; valNodeFeature["envelopEnergy"] = dataDymX.EnvelopEnergy; valNodeFeature["Amp1"] = dataDymX.Amp1; valNodeFeature["Amp2"] = dataDymX.Amp2; valNodeFeature["Amp3"] = dataDymX.Amp3; valNodeFeature["Amp4"] = dataDymX.Amp4; valNodeFeature["Amp5"] = dataDymX.Amp5; valNodeFeature["Phase1"] = dataDymX.Phase1; valNodeFeature["Phase2"] = dataDymX.Phase2; valNodeFeature["Phase3"] = dataDymX.Phase3; valNodeFeature["Phase4"] = dataDymX.Phase4; valNodeFeature["timeStamp"] = nowTimetamp; valNodeData.append(valNodeFeature); DataRecvDym dataDymY; DataExtract(pRecvData, 34, lowbit, n); dataDymY.DiagnosisPk = lowbit * n; DataExtract(pRecvData, 36, lowbit, n); dataDymY.RmsValues = lowbit * n; DataExtract(pRecvData, 38, lowbit, n); dataDymY.IntegratPk = lowbit * n; DataExtract(pRecvData, 40, lowbit, n); dataDymY.IntegratRMS = lowbit * n; DataExtract(pRecvData, 42, lowbit, n); dataDymY.Amp1 = lowbit * n; DataExtract(pRecvData, 44, lowbit, n); dataDymY.Amp2 = lowbit * n; DataExtract(pRecvData, 46, lowbit, n); dataDymY.Amp3 = lowbit * n; DataExtract(pRecvData, 48, lowbit, n); dataDymY.Amp4 = lowbit * n; DataExtract(pRecvData, 50, lowbit, n); dataDymY.Amp5 = lowbit * n; memset(buf, 0, sizeof(buf)); dataDymY.EnvelopEnergy = 0; memset(buf, 0, sizeof(buf)); dataDymY.Phase1 = 0; memset(buf, 0, sizeof(buf)); dataDymY.Phase2 = 0; memset(buf, 0, sizeof(buf)); dataDymY.Phase3 = 0; memset(buf, 0, sizeof(buf)); dataDymY.Phase4 = 0; memset(whereCon, 0, 1024); sprintf(whereCon, "channelID='%s' ", (strMeasurementID + "-Y").c_str()); memset(updateSql, 0, 1024); sprintf(updateSql, "diagnosisPk='%f',integratPk='%f',integratRMS='%f',rmsValues='%f',envelopEnergy='%f',\ Amp1='%f',Amp2='%f',Amp3='%f',Amp4='%f',Amp5='%f',Phase1='%f',Phase2='%f',Phase3='%f',Phase4='%f',timeStamp='%s',StaticIndex = %ld,nodeResend = %d ", dataDymY.DiagnosisPk, dataDymY.IntegratPk, dataDymY.IntegratRMS, dataDymY.RmsValues, dataDymY.EnvelopEnergy, dataDymY.Amp1, dataDymY.Amp2, dataDymY.Amp3, dataDymY.Amp4, dataDymY.Amp5, dataDymY.Phase1, dataDymY.Phase2, dataDymY.Phase3, dataDymY.Phase4, nowTimetamp.c_str(), staticIndex, nodeResend); if (/*0 == sqlite_db_ctrl::instance().GetTableRows(T_DATA_INFO(TNAME), whereCon)*/ (Count * 3 < SAVE_COUNT && lTime < OneWeek) || strTime.size() == 0) { char insertSql[1024] = {0}; memset(insertSql, 0x00, sizeof(insertSql)); sprintf(insertSql, "'%s','%s','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f',%ld,'%s','1',%d", strMeasurementID.c_str(), (strMeasurementID + "-Y").c_str(), dataDymY.DiagnosisPk, dataDymY.IntegratPk, dataDymY.IntegratRMS, dataDymY.RmsValues, dataDymY.EnvelopEnergy, dataDymY.Amp1, dataDymY.Amp2, dataDymY.Amp3, dataDymY.Amp4, dataDymY.Amp5, dataDymY.Phase1, dataDymY.Phase2, dataDymY.Phase3, dataDymY.Phase4, staticIndex, nowTimetamp.c_str(), nodeResend); sqlite_db_ctrl::instance().InsertData(szTableName, insertSql); if (0 == sqlite_db_ctrl::instance().GetTableRows(T_DATA_INFO(TNAME), whereCon)) sqlite_db_ctrl::instance().InsertData(T_DATA_INFO(TNAME), insertSql); else sqlite_db_ctrl::instance().UpdateTableData(T_DATA_INFO(TNAME), updateSql, whereCon); } else { memset(whereCon, 0x00, sizeof(whereCon)); sprintf(whereCon, "channelID='%s' and timeStamp = '%s'", (strMeasurementID + "-Y").c_str(), strTime.c_str()); sqlite_db_ctrl::instance().UpdateTableData(szTableName, updateSql, whereCon); memset(whereCon, 0x00, sizeof(whereCon)); sprintf(whereCon, "channelID='%s' ", (strMeasurementID + "-Y").c_str()); sqlite_db_ctrl::instance().UpdateTableData(T_DATA_INFO(TNAME), updateSql, whereCon); } zlog_info(zct, "y: %s,%s,diagnosisPk=%f,integratPk=%f,integratRMS=%f,rmsValues=%f,envelopEnergy=%f,Amp1=%f,Amp2=%f,Amp3=%f,Amp4=%f,Amp5=%f,Phase1=%f,Phase2=%f,Phase3=%f,Phase4=%f,timeStamp=%s", strMeasurementID.c_str(), (strMeasurementID + "-Y").c_str(), dataDymY.DiagnosisPk, dataDymY.IntegratPk, dataDymY.IntegratRMS, dataDymY.RmsValues, dataDymY.EnvelopEnergy, dataDymY.Amp1, dataDymY.Amp2, dataDymY.Amp3, dataDymY.Amp4, dataDymY.Amp5, dataDymY.Phase1, dataDymY.Phase2, dataDymY.Phase3, dataDymY.Phase4, nowTimetamp.c_str()); valNodeFeature["dataNodeNo"] = strMeasurementID; valNodeFeature["ChannelId"] = strMeasurementID + "-Y"; valNodeFeature["diagnosisPk"] = dataDymY.DiagnosisPk; valNodeFeature["integratPk"] = dataDymY.IntegratPk; valNodeFeature["integratRMS"] = dataDymY.IntegratRMS; valNodeFeature["rmsValues"] = dataDymY.RmsValues; valNodeFeature["envelopEnergy"] = dataDymY.EnvelopEnergy; valNodeFeature["Amp1"] = dataDymY.Amp1; valNodeFeature["Amp2"] = dataDymY.Amp2; valNodeFeature["Amp3"] = dataDymY.Amp3; valNodeFeature["Amp4"] = dataDymY.Amp4; valNodeFeature["Amp5"] = dataDymY.Amp5; valNodeFeature["Phase1"] = dataDymY.Phase1; valNodeFeature["Phase2"] = dataDymY.Phase2; valNodeFeature["Phase3"] = dataDymY.Phase3; valNodeFeature["Phase4"] = dataDymY.Phase4; valNodeFeature["timeStamp"] = nowTimetamp; valNodeData.append(valNodeFeature); DataRecvDym dataDymZ; DataExtract(pRecvData, 52, lowbit, n); dataDymZ.DiagnosisPk = lowbit * n; DataExtract(pRecvData, 54, lowbit, n); dataDymZ.RmsValues = lowbit * n; DataExtract(pRecvData, 56, lowbit, n); dataDymZ.IntegratPk = lowbit * n; DataExtract(pRecvData, 58, lowbit, n); dataDymZ.IntegratRMS = lowbit * n; DataExtract(pRecvData, 60, lowbit, n); dataDymZ.Amp1 = lowbit * n; DataExtract(pRecvData, 62, lowbit, n); dataDymZ.Amp2 = lowbit * n; DataExtract(pRecvData, 64, lowbit, n); dataDymZ.Amp3 = lowbit * n; DataExtract(pRecvData, 66, lowbit, n); dataDymZ.Amp4 = lowbit * n; DataExtract(pRecvData, 68, lowbit, n); dataDymZ.Amp5 = lowbit * n; DataExtract(pRecvData, 70, lowbit, n); dataDymZ.EnvelopEnergy = lowbit * n; DataExtract(pRecvData, 72, lowbit, n); dataDymZ.Phase1 = lowbit * n; DataExtract(pRecvData, 74, lowbit, n); dataDymZ.Phase2 = lowbit * n; DataExtract(pRecvData, 76, lowbit, n); dataDymZ.Phase3 = lowbit * n; DataExtract(pRecvData, 78, lowbit, n); dataDymZ.Phase4 = lowbit * n; memset(whereCon, 0, 1024); sprintf(whereCon, "channelID='%s' ", (strMeasurementID + "-Z").c_str()); memset(updateSql, 0, 1024); sprintf(updateSql, "diagnosisPk='%f',integratPk='%f',integratRMS='%f',rmsValues='%f',envelopEnergy='%f',\ Amp1='%f',Amp2='%f',Amp3='%f',Amp4='%f',Amp5='%f',Phase1='%f',Phase2='%f',Phase3='%f',Phase4='%f',timeStamp='%s',StaticIndex = %ld,nodeResend = %d ", dataDymZ.DiagnosisPk, dataDymZ.IntegratPk, dataDymZ.IntegratRMS, dataDymZ.RmsValues, dataDymZ.EnvelopEnergy, dataDymZ.Amp1, dataDymZ.Amp2, dataDymZ.Amp3, dataDymZ.Amp4, dataDymZ.Amp5, dataDymZ.Phase1, dataDymZ.Phase2, dataDymZ.Phase3, dataDymZ.Phase4, nowTimetamp.c_str(), staticIndex, nodeResend); if (/*0 == sqlite_db_ctrl::instance().GetTableRows(T_DATA_INFO(TNAME), whereCon)*/ Count * 3 < SAVE_COUNT && (lTime < OneWeek || strTime.size() == 0)) { char insertSql[1024] = {0}; memset(insertSql, 0x00, sizeof(insertSql)); sprintf(insertSql, "'%s','%s','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f','%f',%ld,'%s','1',%d", strMeasurementID.c_str(), (strMeasurementID + "-Z").c_str(), dataDymZ.DiagnosisPk, dataDymZ.IntegratPk, dataDymZ.IntegratRMS, dataDymZ.RmsValues, dataDymZ.EnvelopEnergy, dataDymZ.Amp1, dataDymZ.Amp2, dataDymZ.Amp3, dataDymZ.Amp4, dataDymZ.Amp5, dataDymZ.Phase1, dataDymZ.Phase2, dataDymZ.Phase3, dataDymZ.Phase4, staticIndex, nowTimetamp.c_str(), nodeResend); sqlite_db_ctrl::instance().InsertData(szTableName, insertSql); if (0 == sqlite_db_ctrl::instance().GetTableRows(T_DATA_INFO(TNAME), whereCon)) sqlite_db_ctrl::instance().InsertData(T_DATA_INFO(TNAME), insertSql); else sqlite_db_ctrl::instance().UpdateTableData(T_DATA_INFO(TNAME), updateSql, whereCon); } else { memset(whereCon, 0x00, sizeof(whereCon)); sprintf(whereCon, "channelID='%s' and timeStamp = '%s'", (strMeasurementID + "-Z").c_str(), strTime.c_str()); sqlite_db_ctrl::instance().UpdateTableData(szTableName, updateSql, whereCon); memset(whereCon, 0x00, sizeof(whereCon)); sprintf(whereCon, "channelID='%s' ", (strMeasurementID + "-Z").c_str()); sqlite_db_ctrl::instance().UpdateTableData(T_DATA_INFO(TNAME), updateSql, whereCon); } zlog_info(zct, "Z: %s,%s,diagnosisPk=%f,integratPk=%f,integratRMS=%f,rmsValues=%f,envelopEnergy=%f,Amp1=%f,Amp2=%f,Amp3=%f,Amp4=%f,Amp5=%f,Phase1=%f,Phase2=%f,Phase3=%f,Phase4=%f,timeStamp=%s", strMeasurementID.c_str(), (strMeasurementID + "-Z").c_str(), dataDymZ.DiagnosisPk, dataDymZ.IntegratPk, dataDymZ.IntegratRMS, dataDymZ.RmsValues, dataDymZ.EnvelopEnergy, dataDymZ.Amp1, dataDymZ.Amp2, dataDymZ.Amp3, dataDymZ.Amp4, dataDymZ.Amp5, dataDymZ.Phase1, dataDymZ.Phase2, dataDymZ.Phase3, dataDymZ.Phase4, nowTimetamp.c_str()); memset(whereCon, 0x00, sizeof(whereCon)); sprintf(whereCon, "MeasurementID='%s'", strMeasurementID.c_str()); sqlite_db_ctrl::instance().UpdateTableData(T_SENSOR_INFO(TNAME), "status='1'", whereCon); //无线传感器Z信息 valNodeFeature["dataNodeNo"] = strMeasurementID; valNodeFeature["ChannelId"] = strMeasurementID + "-Z"; valNodeFeature["diagnosisPk"] = dataDymZ.DiagnosisPk; valNodeFeature["integratPk"] = dataDymZ.IntegratPk; valNodeFeature["integratRMS"] = dataDymZ.IntegratRMS; valNodeFeature["rmsValues"] = dataDymZ.RmsValues; valNodeFeature["envelopEnergy"] = dataDymZ.EnvelopEnergy; valNodeFeature["Amp1"] = dataDymZ.Amp1; valNodeFeature["Amp2"] = dataDymZ.Amp2; valNodeFeature["Amp3"] = dataDymZ.Amp3; valNodeFeature["Amp4"] = dataDymZ.Amp4; valNodeFeature["Amp5"] = dataDymZ.Amp5; valNodeFeature["Phase1"] = dataDymZ.Phase1; valNodeFeature["Phase2"] = dataDymZ.Phase2; valNodeFeature["Phase3"] = dataDymZ.Phase3; valNodeFeature["Phase4"] = dataDymZ.Phase4; valNodeFeature["timeStamp"] = nowTimetamp; valNodeData.append(valNodeFeature); memset(whereCon, 0, 1024); sprintf(whereCon, "MeasurementID='%s'", strMeasurementID.c_str()); std::string strBattery = sqlite_db_ctrl::instance().GetData(T_SENSOR_INFO(TNAME), "batteryPower", whereCon); std::vector vBattery; vBattery.push_back("0"); vBattery.push_back("0"); if (strBattery.length() > 0) { boost::split(vBattery, strBattery, boost::is_any_of(","), boost::token_compress_on); } //无线传感器信息 Json::Value root; Json::Value valdatastatic; valdatastatic["TemperatureTop"] = dataStatic.TemTop; valdatastatic["TemperatureBot"] = dataStatic.TemBot; valdatastatic["WorkTime"] = dataStatic.nodeWorkTime; valdatastatic["SendTime"] = dataStatic.nodeSendTime; valdatastatic["Dip"] = dataStatic.Dip; valdatastatic["Voltage"] = dataStatic.Voltage; valdatastatic["ChannelType"] = "STATUS"; valdatastatic["ChannelId"] = strMeasurementID + "-S"; valdatastatic["TimeStamp"] = nowTimetamp; valdatastatic["bateryProportion"] = atof(vBattery[1].c_str()) / atof(vBattery[0].c_str()); valdatastatic["batteryRemainDay"] = atof(vBattery[1].c_str()); valdatastatic["dataNodeNo"] = strMeasurementID; valNodeData.append(valdatastatic); root["data"] = valNodeData; root["TimeStamp"] = nowTimetamp; root["dataNodeNo"] = strMeasurementID; root["dataNodeGatewayNo"] = GlobalConfig::MacAddr_G; Json::FastWriter featureValue; std::string strstatisticData = featureValue.write(root); int iRet = data_publish(strstatisticData.c_str(), GlobalConfig::Topic_G.mPubData.c_str()); zlog_info(zct, "dataNodeNo = '%s' and TimeStamp = '%s',MQTT ret = %d", strMeasurementID.c_str(), nowTimetamp.c_str(), iRet); if (iRet != 0) { char updateSql[1024] = {0}; memset(whereCon, 0, 1024); sprintf(whereCon, "dataNodeNo = '%s' and TimeStamp = '%s'", strMeasurementID.c_str(), nowTimetamp.c_str()); memcpy(updateSql, "sendMsg='0'", 11); sqlite_db_ctrl::instance().UpdateTableData(szTableNameStatic, updateSql, whereCon); sqlite_db_ctrl::instance().UpdateTableData(szTableNameData, updateSql, whereCon); } //综上代码,把静态数据,x y z轴的特征值存放到sql数据库中(如果数据原来不存在,则插入新数据;如果存在,则更新数据) zlog_info(zct, "Dip : %d TemBot : %f TemBot : %f Voltage : %d", dataStatic.Dip, dataStatic.TemBot, dataStatic.TemTop, dataStatic.Voltage); memset(selectCon, 0x00, sizeof(selectCon)); sprintf(selectCon, "zigbeeSignal <> '' ORDER BY timeStamp desc LIMIT 0,1"); strTime = sqlite_db_ctrl::instance().GetData(szTableNameStatic, "timeStamp", selectCon); JsonData jd; jd.JsonCmd_32(strMeasurementID,0,0,strMeasurementID,""); if (flag == 1) { zlog_info(zct, "DealDataNodeFeature end %02x%02x", pRecvData->ShortAddr[0], pRecvData->ShortAddr[1]); } } float Uart::ScaleConvert(int highbit) { return 0.0001f * pow(100.0f, highbit); } void Uart::DealDataNodeWave(const char *pData, int comand) { RecvData *pRecvData = (RecvData *)pData; if (wave_trans_) { if (comand == WAVE_X) { VecWaveDataX.push_back(*pRecvData); } else if (comand == WAVE_Y) { VecWaveDataY.push_back(*pRecvData); } else if (comand == WAVE_Z) { VecWaveDataZ.push_back(*pRecvData); } } else { if (comand == WAVE_X) { g_VecWaveDataX[m_waveCountX] = *pRecvData; m_waveCountX++; } else if (comand == WAVE_Y) { g_VecWaveDataY[m_waveCountY] = *pRecvData; m_waveCountY++; } else if (comand == WAVE_Z) { g_VecWaveDataZ[m_waveCountZ] = *pRecvData; m_waveCountZ++; } } char localtimestamp[32] = {0}; GetTimeNet(localtimestamp, 1); // 接收到原始数据信息,则更新时间戳,如果三秒种未收到原始数据,则重新从短地址 9999 切换回 短地址 8888 m_TimeStamp = strtol(localtimestamp, NULL, 10); } void Uart::DealWaveThread() { while (1) { DealWave(); sleep(1); } } std::vector Uart::DealData(int iChannel, float coe, unsigned int sampleRate, int ACCSampleTime, std::string strProduct) { memset(data,0,sizeof(data)); memset(dealdata,0,sizeof(dealdata)); memset(outdata,0,sizeof(outdata)); size_t waveCount = 0; long unsigned int new_len = 0, deallen = 0; int compress = 0; size_t count = 0; long iTemp = 0; char buf[8] = {0x00}; std::vector vecData; size_t j = 0; std::string strShortAddr = ""; if (iChannel == WAVE_X) { if (VecWaveDataX.size() > 0) { g_VecWaveDataX.assign(VecWaveDataX.begin(), VecWaveDataX.end()); waveCount = VecWaveDataX.size(); } waveCount = m_waveCountX; for (; j < waveCount; j++) { RecvData recvData = g_VecWaveDataX[j]; memcpy(data + j * 92, recvData.Data, 92); } memset(buf, 0x00, sizeof(buf)); sprintf(buf, "%02x%02x", g_VecWaveDataX[0].ShortAddr[0], g_VecWaveDataX[0].ShortAddr[1]); strShortAddr = std::string(buf); compress = g_mapCompress[strShortAddr].compressChannelX; count = g_mapCompress[strShortAddr].CountX; } if (iChannel == WAVE_Y) { if (VecWaveDataY.size() > 0) { g_VecWaveDataY.assign(VecWaveDataY.begin(), VecWaveDataY.end()); waveCount = VecWaveDataY.size(); } waveCount = m_waveCountY; for (; j < waveCount; j++) { RecvData recvData = g_VecWaveDataY[j]; memcpy(data + j * 92, recvData.Data, 92); } memset(buf, 0x00, sizeof(buf)); sprintf(buf, "%02x%02x", g_VecWaveDataY[0].ShortAddr[0], g_VecWaveDataY[0].ShortAddr[1]); strShortAddr = std::string(buf); compress = g_mapCompress[strShortAddr].compressChannelY; count = g_mapCompress[strShortAddr].CountY; } if (iChannel == WAVE_Z) { if (VecWaveDataZ.size() > 0) { g_VecWaveDataZ.assign(VecWaveDataZ.begin(), VecWaveDataZ.end()); waveCount = VecWaveDataZ.size(); } waveCount = m_waveCountZ; for (; j < waveCount; j++) { RecvData recvData = g_VecWaveDataZ[j]; memcpy(data + j * 92, recvData.Data, 92); } memset(buf, 0x00, sizeof(buf)); sprintf(buf, "%02x%02x", g_VecWaveDataZ[0].ShortAddr[0], g_VecWaveDataZ[0].ShortAddr[1]); strShortAddr = std::string(buf); compress = g_mapCompress[strShortAddr].compressChannelZ; count = g_mapCompress[strShortAddr].CountZ; } zlog_info(zct, "len = %d,data = %02x,iChannel = %d,compress = %d,count = %d", j, data[0], iChannel, compress, count); if (j * 92 < count) return vecData; if (compress) { zlog_info(zct, "iChannel = %d,compress = %d", iChannel, compress); int r = lzo1x_decompress(data, count, outdata, &new_len, NULL); zlog_info(zct, "lzo1x_decompress end"); if (r == LZO_E_OK) { zlog_warn(zct, "iChannel = %d ,ShortAddr = %s decompressed %lu bytes back into %lu bytes", iChannel, strShortAddr.c_str(), (unsigned long)j * 92, (unsigned long)new_len); } else { zlog_error(zct, "internal error - decompression failed: %d,channel = %d,ShortAddr = %s", r, iChannel, strShortAddr.c_str()); return vecData; } memcpy(dealdata, outdata, new_len); deallen = new_len; } else { memcpy(dealdata, data, j * 92); deallen = j * 92; } for (size_t i = 0; i < deallen; i++) { float fTemp = 0.0f; memset(buf, 0, 8); sprintf(buf, "%02x%02x", dealdata[2 * i + 1], dealdata[i * 2]); iTemp = strtol(buf, NULL, 16); if (iTemp < 0x8000) { fTemp = iTemp * coe * 9.8f; // convert to m/s2 } else { fTemp = (((~iTemp) & 0xffff) + 1) * -coe * 9.8f; // convert to m/s2 } vecData.push_back(fTemp); if (strProduct == "01") { if (vecData.size() == sampleRate * ACCSampleTime && iChannel == WAVE_X) { //过滤数据包结尾空数据 zlog_info(zct, "%d vecData.size() == %d,sampleRate * ACCSampleTime = %d", iChannel, vecData.size(), sampleRate * ACCSampleTime); break; } if (vecData.size() == sampleRate * ACCSampleTime && iChannel == WAVE_Y) { //过滤数据包结尾空数据 zlog_info(zct, "%d vecData.size() == %d,sampleRate * ACCSampleTime = %d", iChannel, vecData.size(), sampleRate * ACCSampleTime); break; } if (vecData.size() == sampleRate * ACCSampleTime && iChannel == WAVE_Z) { //过滤数据包结尾空数据 zlog_info(zct, "%d vecData.size() == %d,sampleRate * ACCSampleTime = %d", iChannel, vecData.size(), sampleRate * ACCSampleTime); break; } } else if (strProduct == "02") { if (vecData.size() == 8192 && iChannel == WAVE_X) { //过滤数据包结尾空数据 break; } if (vecData.size() == 8192 && iChannel == WAVE_Y) { //过滤数据包结尾空数据 break; } if (vecData.size() == sampleRate * ACCSampleTime && iChannel == WAVE_Z) { //过滤数据包结尾空数据 break; } } } return vecData; } void Uart::DealWave() { std::string strShortAddr = ""; std::string strShortAddrTemp; std::string strLongAddr = ""; std::string strMeasurementID = ""; std::string strFileName = ""; std::string strProduct = ""; std::vector vecData; if (wave_trans_) { //对每个传感器的每个通道进行遍历然后处理数据,例如:传感器1x轴的数据处理完后,再去处理y轴的。传感器1的所有数据处理完后,再处理传感器2的 char getzigbeeShortAddr[32] = {0}; sprintf(getzigbeeShortAddr, "zigbeeShortAddr='%02x%02x'", (wave_shortAddr >> 8) & 0xFF,wave_shortAddr & 0xFF); vec_t res = sqlite_db_ctrl::instance().GetDataSingleLine(T_SENSOR_INFO(TNAME), " * ", getzigbeeShortAddr); if (res.size() < 0) { wave_trans_ = false; return; } strLongAddr = res[0]; strMeasurementID = res[44]; if (0 == strLongAddr.length()) { sleep(1); wave_trans_ = false; return; } if (m_waveCountX > 0 || m_waveCountY > 0 || m_waveCountZ > 0){ char localtimestamp[32] = {0}; GetTimeNet(localtimestamp, 1); char insertSql[100] = {0x00}; char whereCon[50] = {0x00}; sprintf(whereCon, "MeasurementID='%s'",strMeasurementID.c_str()); memset(whereCon, 0x00, sizeof(whereCon)); sprintf(insertSql, "'%s','%02x%02x','%s',0,'1','%s' ", strMeasurementID.c_str(),(wave_shortAddr >> 8) & 0xFF,wave_shortAddr & 0xFF,localtimestamp,""); sqlite_db_ctrl::instance().InsertData(" receive_wave_status ", insertSql); } std::string ran = ""; int n = 0; int range = 0; float coe = 0; int sampleRate = 0, ACCSampleTime = 0; char getrange[32] = {0}; std::string str = "range"; sprintf(getrange, "zigbeeShortAddr='%s'", strShortAddr.c_str()); ran = res[25]; sampleRate = atoi(res[23].c_str()); ACCSampleTime = atoi(res[36].c_str()); strProduct = res[17]; memset(getrange, 0, 32); sprintf(getrange, "%s", ran.c_str()); n = (int)strtol(getrange, NULL, 32); if (m_waveCountX > 0 || VecWaveDataX.size() > 0) { zlog_info(zct, "m_waveCountX = %d,VecWaveData = %d", m_waveCountX, VecWaveDataX.size()); coe = Calcoe(n, WAVE_X, strProduct, range); vecData = DealData(WAVE_X, coe, sampleRate, ACCSampleTime, strProduct); WriteDatFile(sampleRate, strMeasurementID, WAVE_X, vecData); m_waveCountX = 0; g_VecWaveDataX.clear(); VecWaveDataX.clear(); } if (m_waveCountY > 0 || VecWaveDataY.size() > 0) { zlog_info(zct, "m_waveCountY = %d,VecWaveData = %d", m_waveCountY, VecWaveDataY.size()); coe = Calcoe(n, WAVE_Y, strProduct, range); vecData = DealData(WAVE_Y, coe, sampleRate, ACCSampleTime, strProduct); WriteDatFile(sampleRate, strMeasurementID, WAVE_Y, vecData); m_waveCountY = 0; g_VecWaveDataY.clear(); VecWaveDataY.clear(); } if (m_waveCountZ > 0 || VecWaveDataZ.size() > 0) { zlog_info(zct, "m_waveCountZ = %d,VecWaveDataZ = %d", m_waveCountZ, VecWaveDataZ.size()); coe = Calcoe(n, WAVE_Z, strProduct, range); vecData = DealData(WAVE_Z, coe, sampleRate, ACCSampleTime, strProduct); WriteDatFile(sampleRate, strMeasurementID, WAVE_Z, vecData); m_waveCountZ = 0; g_VecWaveDataZ.clear(); VecWaveDataZ.clear(); } wave_trans_ = false; } } float Uart::Calcoe(int ran, int iChannel, std::string &product, int range) { float coe = 0.0f; if (product == "01") { switch (ran) { case 0: { range = 8; coe = 8 * 1.0f / 32767; } break; case 1: { range = 16; coe = 16 * 1.0f / 32767; } break; case 2: { range = 32; coe = 32 * 1.0f / 32767; } break; case 3: { range = 64; coe = 64 * 1.0f / 32767; } break; } } else if (product == "02") { if (iChannel == WAVE_X || iChannel == WAVE_Y) { coe = 0.00048828125f; } if (iChannel == WAVE_Z) { coe = 0.00172607421875f; } } return coe; } void Uart::WriteDatFile(int sampleRate, std::string &strMeasurementID, int iChannel, std::vector &vecData) { if (vecData.size() <= 0) return; std::string strFileName = ""; char localtimestamp[32] = {0}; GetTimeNet(localtimestamp, 1); std::string nowTimetamp = std::string(localtimestamp); std::string strChannelID = ""; float mean = Calculation::mean(vecData); float frTemp; char buf[33] = {0x00}; std::string strWaveData = ""; WaveChannel wave_channel; memset(mqttData,0,sizeof(mqttData)); switch (iChannel) { case WAVE_X: { strFileName = "/opt/data/" + strMeasurementID + "-X.dat"; strChannelID = strMeasurementID + "-X"; } break; case WAVE_Y: { strFileName = "/opt/data/" + strMeasurementID + "-Y.dat"; strChannelID = strMeasurementID + "-Y"; } break; case WAVE_Z: { strFileName = "/opt/data/" + strMeasurementID + "-Z.dat"; strChannelID = strMeasurementID + "-Z"; } break; default: break; } FILE *fp = fopen(strFileName.c_str(), "w"); fwrite(localtimestamp,sizeof(localtimestamp),1,fp); zlog_info(zct, " vecData.size : %d,start ", vecData.size()); int id = 0; for (size_t i = 0; i < vecData.size(); i++) { frTemp = vecData[i] - mean; memset(buf, 0x00, sizeof(buf)); sprintf(buf, "%.2f", frTemp); fwrite(&frTemp,sizeof(float),1,fp); if (iChannel == WAVE_X){ wave_channel.WaveChannelX[i] = frTemp; }else if (iChannel == WAVE_Y){ wave_channel.WaveChannelY[i] = frTemp; }else if (iChannel == WAVE_Z){ wave_channel.WaveChannelZ[i] = frTemp; } if (i != vecData.size() -1){ strncpy(mqttData + id ,buf,strlen(buf)); id = id + strlen(buf); strncpy(mqttData + id,",",1); id = id + 1; }else{ strncpy(mqttData + id ,buf,strlen(buf)); } } fclose(fp); zlog_info(zct, "fopen FIle vecData.size : %d end ", vecData.size()); wave_channel.wave_timestamp = nowTimetamp; g_mapWaveChannel[strMeasurementID] = wave_channel; Json::Value valWaveData; valWaveData["number"] = sampleRate; valWaveData["channelId"] = strChannelID; valWaveData["dataNodeNo"] = strMeasurementID; valWaveData["dataNodeGatewayNo"] = GlobalConfig::MacAddr_G; valWaveData["SensorEngineeringUnit"] = ""; valWaveData["timeStamp"] = nowTimetamp; valWaveData["waveData"] = mqttData; valWaveData["mean"] = mean; Json::FastWriter WaveValue; std::string WaveData = WaveValue.write(valWaveData); char selectCon[128] = {0}; sprintf(selectCon, "channelID='%s' ORDER BY timeStamp ASC LIMIT 0,1", strChannelID.c_str()); std::string strTime = sqlite_db_ctrl::instance().GetData("t_data_waveSend", "timeStamp", selectCon); long lTime = atol(nowTimetamp.c_str()) - atol(strTime.c_str()); int Count = sqlite_db_ctrl::instance().GetTableRows("t_data_waveSend", NULL); std::string strFileName_Record = strFileName + "_" + nowTimetamp; if ((Count * 3 < SAVE_COUNT && lTime < OneWeek) || strTime.size() == 0) { char insertSql[128] = {0x00}; sprintf(insertSql, "'%s','%s','%s',1,0,0", strChannelID.c_str(), strFileName_Record.c_str(), nowTimetamp.c_str()); sqlite_db_ctrl::instance().InsertData("t_data_waveSend", insertSql); } else { char updateSql[128] = {0}, whereCon[128] = {0}; sprintf(updateSql, "waveName='%s',timeStamp='%s'", strFileName_Record.c_str(), nowTimetamp.c_str()); memset(whereCon, 0x00, sizeof(whereCon)); sprintf(whereCon, "channelID='%s' and timeStamp = '%s'", strChannelID.c_str(), strTime.c_str()); zlog_info(zct, "update static data to sql"); sqlite_db_ctrl::instance().UpdateTableData("t_data_waveSend", updateSql, whereCon); } //传感器发来的数据包中的表示设备信息的数据转化为json格式后,通过调用data_publish将数据传给mqttclient : Topic:wireless/cmd/60294D203717 int iRet = data_publish(WaveData.c_str(), GlobalConfig::Topic_G.mPubWaveData.c_str()); if (iRet != 0) { char whereCon[1024] = {0x00}; char updateSql[1024] = {0x00}; char tmpWhere[128] = {0x00}; sprintf(tmpWhere, "channelID = '%s' and sendMsg = 0 ", strChannelID.c_str()); int count = sqlite_db_ctrl::instance().GetTableRows("t_data_waveSend", tmpWhere); zlog_info(zct, "save channlID %s dat count = %d", strChannelID.c_str(), count); zlog_warn(zct, "send failed,filename %s,iRet = %d", strFileName.c_str(), iRet); if (count <= 3) { sprintf(whereCon, "channelID='%s' and timeStamp = '%s' ", strChannelID.c_str(), nowTimetamp.c_str()); sprintf(updateSql, "SendMsg = 0 , error_code = %d",iRet); sqlite_db_ctrl::instance().UpdateTableData("t_data_waveSend", updateSql, whereCon); std::string strFileName_failed = strFileName + "_" + nowTimetamp; char tmpCmd[128] = {0x00}; sprintf(tmpCmd, "cp %s %s", strFileName.c_str(), strFileName_failed.c_str()); system(tmpCmd); } else { memset(tmpWhere, 0x00, sizeof(tmpWhere)); memset(updateSql, 0x00, sizeof(updateSql)); sprintf(tmpWhere, " sendMsg = 0 and channelID='%s' ORDER BY timeStamp ASC LIMIT 0,1", strChannelID.c_str()); vec_t vecRet = sqlite_db_ctrl::instance().GetDataSingleLine("t_data_waveSend", "*", tmpWhere); memset(tmpWhere, 0x00, sizeof(tmpWhere)); sprintf(tmpWhere, " sendMsg = 0 and timeStamp = '%s' and channelID = '%s' ", vecRet[2].c_str(), vecRet[0].c_str()); sprintf(updateSql, " sendMsg = 3 "); sqlite_db_ctrl::instance().UpdateTableData("t_data_waveSend", updateSql, tmpWhere); memset(tmpWhere, 0x00, sizeof(tmpWhere)); memset(updateSql, 0x00, sizeof(updateSql)); sprintf(whereCon, " channelID='%s' and timeStamp = '%s' ", strChannelID.c_str(), nowTimetamp.c_str()); sprintf(updateSql, " sendMsg = 0 , error_code = %d ",iRet); sqlite_db_ctrl::instance().UpdateTableData("t_data_waveSend", updateSql, whereCon); std::string strFileName_failed = strFileName + "_" + nowTimetamp; char tmpCmd[128] = {0x00}; sprintf(tmpCmd, "cp %s %s", strFileName.c_str(), strFileName_failed.c_str()); system(tmpCmd); zlog_info(zct, "cp dat file %s \n", tmpCmd); memset(tmpWhere, 0x00, sizeof(tmpWhere)); sprintf(tmpWhere, " channelID = '%s' and sendMsg = 0 ", strChannelID.c_str()); sqlite_db_ctrl::instance().GetTableRows("t_data_waveSend", tmpWhere); memset(tmpCmd, 0x00, sizeof(tmpCmd)); sprintf(tmpCmd, "rm %s ", vecRet[1].c_str()); system(tmpCmd); zlog_info(zct, "rm dat file %s \n", tmpCmd); } JsonData jd; jd.JsonCmd_32(strMeasurementID,1,1,strChannelID,"send failed"); } else { zlog_warn(zct, "send data , filename %s,size = %d\n", strFileName.c_str(), vecData.size()); } JsonData jd; jd.JsonCmd_32(strMeasurementID,0,1,strChannelID,""); zlog_info(zct, "write data to filename %s", strFileName.c_str()); std::vector().swap(vecData); sleep(1); }