#include "rate.h" #include #include #include #include #include struct rate_dev_cfg rate_cfg_head; //描述设备的报文配置信息 列表表头 extern void channel_lock(channel_t* channel); extern void channel_unlock(channel_t* channel); extern unsigned short Reverse16(unsigned short sVal); // 高低反转 extern unsigned int Reverse32(unsigned int iVal); // 完全反转 extern unsigned int Reverse16with32(unsigned int iVal); // 双字节反转 long milliseconds; int rate_send(void * args); extern void GEN_Num2ASIC(unsigned char datas, unsigned char *asic); static void rate_GEN_ARRHEX2ASIC(unsigned char *hexdata, unsigned char *asicdata, unsigned short hexlength,char spilite) { unsigned short i=0; unsigned char asic[2]; if(spilite == 0) { for(i=0;istring; int type = 0; if (strncmp(key, "send_", 5) == 0 ) { type = 0; } else if (strncmp(key, "read_", 5) == 0) { type = 1; } else { ems_syslog(LOG_WARNING, "Unknown key in message object: %s", key); continue; } if (!cJSON_IsArray(item)) { ems_syslog(LOG_WARNING, "Expected array for message key: %s", key); continue; } struct rate_dev_message *message = (struct rate_dev_message *)calloc(1, sizeof(struct rate_dev_message)); message->type = type; list_add_tail(&message->list, &dev_cfg->list); INIT_LIST_HEAD(&message->data.list); int cmd_count = cJSON_GetArraySize(item); for (int i = 0; i < cmd_count; i++) { cJSON *cmd = cJSON_GetArrayItem(item, i); if (!cJSON_IsString(cmd)) { ems_syslog(LOG_WARNING, "Expected string for command: %d", i); continue; } struct rate_message_node *Node = (struct rate_message_node *)calloc(1, sizeof(struct rate_message_node)); Node->type = RATE_MSG_NODE_NON; const char *cmd_str = cmd->valuestring; list_add_tail(&Node->list, &message->data.list); if (strstr(cmd_str, "zone") == cmd_str) // 尖峰平谷配置 { char suffix[5] = {0}; if(sscanf(cmd_str, "zone%d_%4s", &Node->data.time[0], suffix) == 2) { Node->data.time[1] = 0; if(strcmp(suffix, "MON") == 0) Node->type = RATE_MSG_NODE_TZ_MON; else if(strcmp(suffix, "DAY") == 0) Node->type = RATE_MSG_NODE_TZ_DAY; else { ems_syslog(LOG_WARNING, "Invalid zone format: %s", cmd_str); continue; } } else { ems_syslog(LOG_WARNING, "Invalid zone format: %s", cmd_str); continue; } } else if (strstr(cmd_str, "row") == cmd_str && strstr(cmd_str, "col")) // 矩阵配置 { char suffix[5] = ""; if (sscanf(cmd_str, "row%d_col%d_%4s", &Node->data.time[0], &Node->data.time[1], suffix) == 3) { ems_syslog(LOG_DEBUG, " Matrix node: row=%d, col=%d, suffix=%s", Node->data.time[0], Node->data.time[1], suffix); if (strcmp(suffix, "MIN") == 0) Node->type = RATE_MSG_NODE_TF_MIN; else if (strcmp(suffix, "HOUR") == 0) Node->type = RATE_MSG_NODE_TF_HOUR; else if (strcmp(suffix, "MODE") == 0) Node->type = RATE_MSG_NODE_TF_MODE; else { ems_syslog(LOG_WARNING, "Invalid matrix format: %s", cmd_str); continue; } ems_syslog(LOG_DEBUG, " Node type: %d", Node->type); } else { ems_syslog(LOG_WARNING, "Invalid matrix format: %s", cmd_str); continue; } } else if (strcmp(cmd_str, "PASSWD") == 0) // 密码配置 { Node->type = RATE_MSG_NODE_PASSWD; Node->data.passwd = 0; ems_syslog(LOG_DEBUG, " Found PASSWD node"); } else if (strcmp(cmd_str, "645_CR") == 0) // 密码配置 { Node->type = RATE_MSG_NODE_645_CR; ems_syslog(LOG_DEBUG, " Found PASSWD node"); } else if (strstr(cmd_str, "tab645_") == cmd_str) // 密码配置 { Node->type = RATE_MSG_NODE_645_TAB; if(sscanf(cmd_str, "tab645_%d", &Node->data.time[0]) == 1) { ems_syslog(LOG_DEBUG, " Matrix node: row=%d", Node->data.time[0]); } ems_syslog(LOG_DEBUG, " Found PASSWD node"); } else if (strstr(cmd_str, "md_rtu_crc") == cmd_str) // 密码配置 { Node->type = RATE_MSG_NODE_RTU_CRC; ems_syslog(LOG_DEBUG, " Found md_rtu_crc node"); } else // 无法识别特殊标识的全部视为hex文本来记录 { ems_syslog(LOG_DEBUG, " Text node: %s", cmd_str); Node->type = RATE_MSG_NODE_TEXT; Node->data.message = strdup(cmd_str); } } } } DATA_RATE_ORDER get_RATE_ORDER_by_string(char *typeString); char *rate_order_string[] = { "A", "B", "AB", "BA", "ABCD", "DCBA", "BADC", "CDAB", "ABCDEFGH", "GHEFCDAB", "BADCFEHG", "HGFEDCBA" }; DATA_RATE_ORDER get_RATE_ORDER_by_string(char *typeString) { for (DATA_RATE_ORDER i = RATE_ORDER_A; i < RATE_ORDER_MAX; i++) { if(strcmp(typeString,rate_order_string[i])==0) return i; } proto_syslog(LOG_WARNING, "type %s ORDER_ERROR",typeString); return RATE_ORDER_ERROR; } static char hex_char_to_value(char c) { if (c >= '0' && c <= '9') return c - '0'; else if (c >= 'A' && c <= 'F') return c - 'A' + 10; else if (c >= 'a' && c <= 'f') return c - 'a' + 10; return 0; } static void calculate_crc_645(const char *str, char crc[3] ) { size_t len = strlen(str); unsigned char sum = 0; // 每两个字符解析为一个字节并累加 for (size_t i = 0; i < len; i += 2) { unsigned char high = hex_char_to_value(str[i]); // 高4位 unsigned char low = hex_char_to_value(str[i + 1]); // 低4位 unsigned char byte = (high << 4) | low; // 组合成一个字节 sum += byte; // 累加到sum } // 将累加和转换为两位十六进制字符串 const char hex_chars[] = "0123456789ABCDEF"; crc[0] = hex_chars[(sum >> 4) & 0x0F]; // 高4位 crc[1] = hex_chars[sum & 0x0F]; // 低4位 crc[2] = '\0'; // 字符串结束符 } /** * @description: 获取rate.json的配置注册到rate_cfg_head * @return {*} */ int get_rate_cfg() { INIT_LIST_HEAD(&rate_cfg_head.list); // 链表初始化放在最前防止后续遍历接口问题 const char *cfg_str = read_file_data(RATECFG_PATH); if (!cfg_str) { ems_syslog(LOG_ERR, "Failed to read rate config file: %s", RATECFG_PATH); return -1; } cJSON *root = cJSON_Parse(cfg_str); free((char *)cfg_str); if (!root) { ems_syslog(LOG_ERR, "Failed to parse rate config JSON"); return -1; } if (!cJSON_IsArray(root)) { ems_syslog(LOG_ERR, "Rate config JSON root is not an array"); cJSON_Delete(root); return -1; } int device_count = cJSON_GetArraySize(root); ems_syslog(LOG_INFO, "Found %d device configurations", device_count); int dev_index = 0; cJSON *device_item = NULL; cJSON_ArrayForEach(device_item, root) { // 解析基础信息 cJSON *vendor = cJSON_GetObjectItem(device_item, "vendor"); cJSON *mode = cJSON_GetObjectItem(device_item, "mode"); cJSON *agreement = cJSON_GetObjectItem(device_item, "agreement"); cJSON *adapter = cJSON_GetObjectItem(device_item, "adapter"); cJSON *read_en = cJSON_GetObjectItem(device_item, "read_en"); cJSON *write_en = cJSON_GetObjectItem(device_item, "write_en"); cJSON *delay_time = cJSON_GetObjectItem(device_item, "delay_time"); cJSON *protocol = cJSON_GetObjectItem(device_item, "protocol"); if (!vendor || !mode || !adapter || !read_en || !write_en) { ems_syslog(LOG_ERR, "Missing required fields in device config"); continue; } struct rate_dev_cfg *dev_cfgs = calloc(1, sizeof(struct rate_dev_cfg)); if (!dev_cfgs) { ems_syslog(LOG_ERR, "Failed to allocate memory for device configs"); cJSON_Delete(root); return -1; } // 填充设备配置 dev_cfgs->vendor = strdup(vendor->valuestring); dev_cfgs->model = strdup(mode->valuestring); if(agreement) { dev_cfgs->agreement = strdup(agreement->valuestring); } else { dev_cfgs->agreement = strdup(""); } // 处理read_en字段 if (cJSON_IsString(read_en)) { dev_cfgs->read_en = (strcasecmp(read_en->valuestring, "true") == 0) ? 1 : 0; } else { dev_cfgs->read_en = cJSON_IsTrue(read_en) ? 1 : 0; } // 处理write_en字段 if (cJSON_IsString(write_en)) { dev_cfgs->write_en = (strcasecmp(write_en->valuestring, "true") == 0) ? 1 : 0; } else { dev_cfgs->write_en = cJSON_IsTrue(write_en) ? 1 : 0; } dev_cfgs->delay_time = delay_time->valueint; dev_cfgs->protocol = get_RATE_ORDER_by_string(protocol->valuestring); // 解析adapter数组 if (cJSON_IsArray(adapter)) { int adapter_size = cJSON_GetArraySize(adapter); for (int i = 0; i < adapter_size && i < 6; i++) { cJSON *item = cJSON_GetArrayItem(adapter, i); if (cJSON_IsNumber(item)) { dev_cfgs->adapter[i] = item->valueint; } } } ems_syslog(LOG_INFO, "Loaded device config: %s-%s", dev_cfgs->vendor, dev_cfgs->model); // 处理message对象 cJSON *message = cJSON_GetObjectItem(device_item, "write"); if (message && cJSON_IsObject(message)) { INIT_LIST_HEAD(&dev_cfgs->write.list); process_message_object(message, &dev_cfgs->write); } // 处理Callfortesting对象 cJSON *callfortesting = cJSON_GetObjectItem(device_item, "read"); if (callfortesting && cJSON_IsObject(callfortesting)) { INIT_LIST_HEAD(&dev_cfgs->read.list); process_message_object(callfortesting, &dev_cfgs->read); } list_add_tail(&dev_cfgs->list, &rate_cfg_head.list); dev_index++; } cJSON_Delete(root); ems_syslog(LOG_INFO, "Successfully loaded %d device configurations", dev_index); return 0; } static void get_data_short(unsigned short *data_s, int point_info, int value) { switch (point_info) { case RATE_ORDER_A: // 8位 data_s[0] = value; /* fall through */ case RATE_ORDER_B: // 8位 反转 data_s[0] = ((value & 0x00F0) >> 4) | ((value & 0x0F) << 4) ; /* fall through */ case RATE_ORDER_AB: // 16位 小端序 (AB) data_s[0] = value; break; case RATE_ORDER_BA: // 16位 大端序 (BA) data_s[0] = Reverse16(value); break; case RATE_ORDER_ABCD: // 32位 小端序 (ABCD) data_s[0] = value & 0xFFFF; // 低16位 (AB) data_s[1] = (value >> 16) & 0xFFFF; // 高16位 (CD) break; case RATE_ORDER_BADC: // 32位 混合序 (BADC) data_s[0] = Reverse16(value & 0xFFFF); // AB -> BA data_s[1] = Reverse16((value >> 16) & 0xFFFF); // CD -> DC break; case RATE_ORDER_CDAB: // 32位 中端序 (CDAB) data_s[0] = (value >> 16) & 0xFFFF; // 高16位 (CD) data_s[1] = value & 0xFFFF; // 低16位 (AB) break; case RATE_ORDER_DCBA: // 32位 大端序 (DCBA) { unsigned int reversed = Reverse32(value); data_s[0] = reversed >> 16; // 高16位 (DC) data_s[1] = reversed & 0xFFFF; // 低16位 (BA) } break; default: ems_syslog(LOG_WARNING, "Unsupported byte RATE_ORDER: %d", point_info); break; } } static void get_data_short_645(unsigned short *data_s, int point_info, int value) { char buf[10] = {0}; snprintf(buf, 10, "%d", value); sscanf(buf, "%hx", &data_s[0]); data_s[0] += 0x33; } // 反转16位整数的字节序 // static unsigned short Reverse16(unsigned short value) { // return (value >> 8) | (value << 8); // } // // 反转32位整数的字节序 // static unsigned int Reverse32(unsigned int value) { // return ((value >> 24) & 0xFF) | // ((value >> 8) & 0xFF00) | // ((value << 8) & 0xFF0000) | // ((value << 24) & 0xFF000000); // } // 从data_s数组中提取整数值 static int get_data_short_reverse(unsigned short *data_s, int point_info) { switch (point_info) { case RATE_ORDER_A: return (int)(data_s[0] & 0x00FF); case RATE_ORDER_B: return (int)(((data_s[0] & 0x00F0) >> 4) | ((data_s[0] & 0x0F) << 4)); case RATE_ORDER_AB: // 小端序16位 (AB) return (int)data_s[0]; case RATE_ORDER_BA: // 大端序16位 (BA) return (int)Reverse16(data_s[0]); case RATE_ORDER_ABCD: // 小端序32位 (ABCD) return (int)((data_s[1] << 16) | data_s[0]); case RATE_ORDER_BADC: // 混合序32位 (BADC) return (int)((Reverse16(data_s[1]) << 16) | Reverse16(data_s[0])); case RATE_ORDER_CDAB: // 中端序32位 (CDAB) return (int)((data_s[0] << 16) | data_s[1]); case RATE_ORDER_DCBA: // 大端序32位 (DCBA) return (int)Reverse32((data_s[1] << 16) | data_s[0]); default: return 0; // 未知类型返回0 } } int value_mode_transition(struct lnxall_buff *buf, int value, DATA_RATE_ORDER protocol) { unsigned short bytes[10] = {0}; get_data_short(bytes, protocol, value); if(protocol < RATE_ORDER_AB) { lbuff_sprintf(buf, "%.2hx", bytes[0]); } else if(protocol < RATE_ORDER_ABCD) { lbuff_sprintf(buf, "%.4hx", bytes[0]); } else { lbuff_sprintf(buf, "%.4hx%.4hx", bytes[0], bytes[1]); } return 0; } int value_mode_transition_645(struct lnxall_buff *buf, int value, DATA_RATE_ORDER protocol) { unsigned short bytes[10] = {0}; get_data_short_645(bytes, protocol, value); if(protocol < RATE_ORDER_AB) { lbuff_sprintf(buf, "%.2hx", bytes[0]); } else if(protocol < RATE_ORDER_ABCD) { lbuff_sprintf(buf, "%.4hx", bytes[0]); } else { lbuff_sprintf(buf, "%.4hx%.4hx", bytes[0], bytes[1]); } return 0; } int hex_to_value(char *buf, int *value, int protocol) { unsigned short bytes[10] = {0}; int ret = 0; if(protocol < RATE_ORDER_AB) { sscanf(buf, "%2hx", &bytes[0]); ret = 2; } else if(protocol < RATE_ORDER_ABCD) { sscanf(buf, "%4hx", &bytes[0]); ret = 4; } else /*更长的字段后续追加*/ { sscanf(buf, "%4hx%4hx", &bytes[0], &bytes[1]); ret = 8; } *value = get_data_short_reverse(bytes, protocol); return ret; } int hex_to_value_645(char *buf, int *value, int protocol) { unsigned short bytes[10] = {0}; char buff[10] = {0}; memcpy(buff, buf, 2); sscanf(buff, "%hx", &bytes[0]); bytes[0] -= 0x33; memset(buff, 0, 10); sprintf(buff, "%hx", bytes[0]); sscanf(buff, "%d", value); return 2; } unsigned short modbus_crc16(const unsigned char *buf, int len) { unsigned short crc = 0xFFFF; for (int pos = 0; pos < len; pos++) { crc ^= (unsigned short)buf[pos]; for (int i = 0; i < 8; i++) { if (crc & 0x0001) { crc >>= 1; crc ^= 0xA001; } else { crc >>= 1; } } } return crc; } // 将十六进制字符串转为字节数组 int hexstr_to_bytes(const char *hexstr, unsigned char *out) { int len = strlen(hexstr); if (len % 2 != 0) return -1; // 必须是偶数长度 int out_len = len / 2; for (int i = 0; i < out_len; i++) { sscanf(hexstr + 2*i, "%2hhx", &out[i]); } return out_len; } // 输出 CRC 字符串，低字节在前 void get_rtu_crc_string(const char *buf_str, char crc[5]) { unsigned char bytes[256]; int len = strlen(buf_str) / 2; for (int i = 0; i < len; i++) { sscanf(buf_str + 2*i, "%2hhx", &bytes[i]); } unsigned short crc_val = modbus_crc16(bytes, len); unsigned char lo = crc_val & 0xFF; unsigned char hi = (crc_val >> 8) & 0xFF; // 注意顺序：低字节在前，高字节在后 snprintf(crc, 5, "%02x%02x", lo, hi); } /** * @description: 获取各行各列数据 返回节点数据 若访问指针超出范围则返回 NULL * @param {int} row * @param {int} col * @param {tags_matrix} *matrix * @return {*} */ struct tags_matrix_Node * get_matrix_Node(int row, int col, struct tags_matrix *matrix) { if(row < matrix->row_num && col < matrix->rows[row].Node_num) { return &matrix->rows[row].Node[col]; } return NULL; } /** * @description: 写入配置 * @param {char} *dev_no * @param {tags_matrix} *matrix：需写入的配置将按规则预填充在矩阵中 * @return {*} */ int rate_cfg_write(char *dev_no, struct tags_matrix *matrix) { int timeout; int ret = 0; proto_forward_t *pro_pr = get_proto_forward_var(); channel_t *chan_pr=NULL; device_t *dev_pr=NULL; struct rate_dev_message *write_item = NULL; int list_count = 0; // current if(find_dev_channel_pr_by_dev_no(pro_pr,dev_no,&chan_pr,&dev_pr) != 0) { ems_syslog(LOG_ERR, "dev_no[%s] not found!", dev_no); return -1; } struct list_head *pos, *lists; struct rate_dev_cfg *current = NULL, *tmp_dev_cfg; // 遍历链表查找匹配的设备配置 list_for_each_safe(pos, lists, &rate_cfg_head.list) { tmp_dev_cfg = list_entry(pos, struct rate_dev_cfg, list); ems_syslog(LOG_ERR, "Device config found: vendor=%s, model=%s", tmp_dev_cfg->vendor, tmp_dev_cfg->model); // struct rate_message_node *message_item = NULL; struct list_head *pos_item, *list_item; list_for_each_safe(pos_item, list_item, &tmp_dev_cfg->list) { // message_item = list_entry(pos_item, struct rate_message_node, list); } if(strcmp(tmp_dev_cfg->vendor, dev_pr->use_template->vendor) == 0 && strcmp(tmp_dev_cfg->model, dev_pr->use_template->model) == 0) { current = tmp_dev_cfg; break; } } if(NULL == current) { ems_syslog(LOG_ERR, "No matching device config found"); return -2; } if(current->write_en == 0) { ems_syslog(LOG_ERR, "Device config write is disabled"); return -3; } list_count = 0; list_for_each_safe(pos, lists, ¤t->write.list ) { list_count++; // 获取报文个数 } struct lnxall_buff *buff = (struct lnxall_buff *)calloc(list_count, sizeof(struct lnxall_buff)); if(!buff) { ems_syslog(LOG_ERR, "Failed to allocate memory for lnxall_buff"); return -3; } // 遍历rate_cfg_head链表 int cnt = 0; struct tags_matrix_Node *rate_node = NULL; int send_hex_len = 0; timeout = current->delay_time; /*报文组装循环*/ if(0 == strcmp(current->agreement, "DLT645")) { list_for_each_safe(pos, lists, ¤t->write.list ) { write_item = list_entry(pos, struct rate_dev_message, list); ems_syslog(LOG_DEBUG, "Processing device config: type=%d", write_item->type); lbuff_init(&buff[cnt], 256); struct rate_message_node *message_item = NULL; struct list_head *pos_item, *list_item; list_for_each_safe(pos_item, list_item, &write_item->data.list) { message_item = list_entry(pos_item, struct rate_message_node, list); switch (message_item->type) { case RATE_MSG_NODE_NON: ems_syslog(LOG_ERR, "Unknown message type"); break; case RATE_MSG_NODE_TEXT: lbuff_append(&buff[cnt], message_item->data.message, (int)strlen(message_item->data.message)); break; case RATE_MSG_NODE_TZ_MON: if(message_item->data.time[0] >= matrix->row_num) { value_mode_transition_645(&buff[cnt], 0, current->protocol); } else { value_mode_transition_645(&buff[cnt], matrix->rows[message_item->data.time[0]].mon, current->protocol); } break; case RATE_MSG_NODE_TZ_DAY: if(message_item->data.time[0] >= matrix->row_num) { value_mode_transition_645(&buff[cnt], 0, current->protocol); } else { value_mode_transition_645(&buff[cnt], matrix->rows[message_item->data.time[0]].day, current->protocol); } break; case RATE_MSG_NODE_TF_HOUR: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { value_mode_transition_645(&buff[cnt], 0, current->protocol); } else { value_mode_transition_645(&buff[cnt], rate_node->hour, current->protocol); } break; case RATE_MSG_NODE_TF_MIN: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { value_mode_transition_645(&buff[cnt], 0, current->protocol); } else { value_mode_transition_645(&buff[cnt], rate_node->min, current->protocol); } break; case RATE_MSG_NODE_TF_MODE: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { value_mode_transition_645(&buff[cnt], 0, current->protocol); } else { value_mode_transition_645(&buff[cnt], (int)rate_node->type, current->protocol); } break; case RATE_MSG_NODE_PASSWD: // 密码先不做，用户的645电表若有密码，则控制器必须要得知，若控制器得知，可在rate_fig.json中同步填写 { } break; case RATE_MSG_NODE_645_CR: { char *data_start = strstr(buff[cnt].bufptr, "68"); char crc[3] = ""; if(NULL == data_start) break; calculate_crc_645(data_start, crc); lbuff_append(&buff[cnt], crc, 2); } break; case RATE_MSG_NODE_645_TAB: break; case RATE_MSG_NODE_RTU_CRC: break; } } ems_syslog(LOG_ERR, "write_hex: [%s]", buff[cnt].bufptr); cnt++; } } else { // list_for_each_entry(write_item, &rate_cfg_head.write.list, list) { list_for_each_safe(pos, lists, ¤t->write.list ) { write_item = list_entry(pos, struct rate_dev_message, list); ems_syslog(LOG_DEBUG, "Processing device config: type=%d", write_item->type); lbuff_init(&buff[cnt], 256); struct rate_message_node *message_item = NULL; struct list_head *pos_item, *list_item; list_for_each_safe(pos_item, list_item, &write_item->data.list) { message_item = list_entry(pos_item, struct rate_message_node, list); switch (message_item->type) { case RATE_MSG_NODE_NON: ems_syslog(LOG_ERR, "Unknown message type"); break; case RATE_MSG_NODE_TEXT: lbuff_append(&buff[cnt], message_item->data.message, (int)strlen(message_item->data.message)); break; case RATE_MSG_NODE_TZ_MON: if(message_item->data.time[0] > matrix->row_num) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto ERR; } value_mode_transition(&buff[cnt], matrix->rows[message_item->data.time[0]].mon, current->protocol); break; case RATE_MSG_NODE_TZ_DAY: if(message_item->data.time[0] > matrix->row_num) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto ERR; } value_mode_transition(&buff[cnt], matrix->rows[message_item->data.time[0]].day, current->protocol); break; case RATE_MSG_NODE_TF_HOUR: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto ERR; } value_mode_transition(&buff[cnt], rate_node->hour, current->protocol); break; case RATE_MSG_NODE_TF_MIN: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto ERR; } value_mode_transition(&buff[cnt], rate_node->min, current->protocol); break; case RATE_MSG_NODE_TF_MODE: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto ERR; } value_mode_transition(&buff[cnt], rate_node->type, current->protocol); break; case RATE_MSG_NODE_PASSWD: /*TODO 密码目前无协议样本暂时不予配置实现*/ break; case RATE_MSG_NODE_RTU_CRC: { char *data_start = buff[cnt].bufptr; char crc[5] = ""; if(NULL == data_start) break; get_rtu_crc_string(data_start, crc); lbuff_append(&buff[cnt], crc, 4); } break; case RATE_MSG_NODE_645_CR: break; case RATE_MSG_NODE_645_TAB: break; } } cnt++; } } /*报文组装后下发*/ cnt = 0; channel_lock(chan_pr); list_for_each_safe(pos, lists, ¤t->write.list ) { write_item = list_entry(pos, struct rate_dev_message, list); void *hex_data = malloc(strlen(buff[cnt].bufptr) * sizeof(unsigned char) + 2); if(!hex_data) { channel_unlock(chan_pr); ems_syslog(LOG_ERR, "Failed to allocate memory for hex data"); ret = 1; goto ERR; } send_hex_len = GEN_ARRASIC2HEX((unsigned char *)buff[cnt].bufptr, hex_data, strlen(buff[cnt].bufptr)); if(send_hex_len<=0) { channel_unlock(chan_pr); ems_syslog(LOG_ERR, "Failed to convert ASCII to HEX"); ret = 2; free(hex_data); goto ERR; } if(!(chan_pr->s > 0)) { channel_unlock(chan_pr); ems_syslog(LOG_ERR, "dev_no[%s] channel is not connected!", dev_no); free(hex_data); goto ERR; } if(write_item->type == 0) /* 0:send 1:read */ { ems_syslog(LOG_ERR, "dev_no[%s] write data: [%s]", dev_no, buff[cnt].bufptr); if (write(chan_pr->s, hex_data, send_hex_len) < 0) { channel_unlock(chan_pr); ems_syslog(LOG_ERR, "dev_no[%s] write data error!: [%s]", dev_no, buff[cnt].bufptr); ret = 3; free(hex_data); goto ERR; } } else { int len = read_data_timeout(chan_pr->s,hex_data,send_hex_len, timeout); unsigned char *hex_str = calloc(strlen(buff[cnt].bufptr) , sizeof(unsigned char) + 2); rate_GEN_ARRHEX2ASIC(hex_data, hex_str, send_hex_len, 0); ems_syslog(LOG_ERR, "dev_no[%s] read data: [%s]", dev_no, hex_str); if(len <= 0 || memcmp(hex_str, buff[cnt].bufptr, strlen(buff[cnt].bufptr))) { channel_unlock(chan_pr); ems_syslog(LOG_ERR, "dev_no[%s] read data error!: [%s]", dev_no, buff[cnt].bufptr); ret = 4; free(hex_data); free(hex_str); goto ERR; } free(hex_str); } free(hex_data); cnt++; } channel_unlock(chan_pr); ERR: cnt = 0; list_for_each_entry(write_item, ¤t->write.list, list) { lbuff_free(&buff[cnt]); cnt++; } free(buff); return ret; } int rate_zhuzhuang_mqtt_message(int seq, struct tags_matrix *matrix) { if (!matrix) { ems_syslog(LOG_ERR, "[zxf]bulid_get_multi_rate_info invalid input parameters"); return -1; } // 创建info对象 cJSON *info = cJSON_CreateObject(); if (!info) { ems_syslog(LOG_ERR, "[zxf]Failed to create info object"); return -1; } // 添加dev_no字段 cJSON_AddStringToObject(info, "dev_no", matrix->dev_no); // 创建tags数组 cJSON *tags = cJSON_CreateArray(); if (!tags) { ems_syslog(LOG_ERR, "[zxf]Failed to create tags array"); cJSON_Delete(info); return -1; } cJSON_AddItemToObject(info, "tags", tags); // 遍历所有行 for (int i = 0; i < matrix->row_num; i++) { struct tags_matrix_row *row = &matrix->rows[i]; // 创建行数组 cJSON *row_array = cJSON_CreateArray(); if (!row_array) { ems_syslog(LOG_ERR, "[zxf]Failed to create row array"); continue; } cJSON_AddItemToArray(tags, row_array); // 添加月份和日期 [mon, day] cJSON *date_array = cJSON_CreateArray(); cJSON_AddItemToArray(date_array, cJSON_CreateNumber(row->mon)); cJSON_AddItemToArray(date_array, cJSON_CreateNumber(row->day)); cJSON_AddItemToArray(row_array, date_array); // 遍历所有节点 for (int j = 0; j < row->Node_num; j++) { struct tags_matrix_Node *node = &row->Node[j]; // 创建时间类型数组 ["HH:MM", type] cJSON *node_array = cJSON_CreateArray(); char time_str[6]; snprintf(time_str, sizeof(time_str), "%02d:%02d", node->hour, node->min); cJSON_AddItemToArray(node_array, cJSON_CreateString(time_str)); cJSON_AddItemToArray(node_array, cJSON_CreateNumber(node->type)); cJSON_AddItemToArray(row_array, node_array); } } // 将info对象转换为字符串 char *info_str = cJSON_PrintUnformatted(info); cJSON_Delete(info); if (!info_str) { ems_syslog(LOG_ERR, "[zxf]Failed to print info JSON"); return -1; } // 构建完整报文 struct lnxall_buff ex_data; lbuff_init(&ex_data, 512); lbuff_sprintf(&ex_data, "\"info\":%s", info_str); free(info_str); struct lnxall_buff data_all; lbuff_init(&data_all,(int)strlen(ex_data.bufptr)); char topic_sn[64] = "2188310003E6"; lbuff_sprintf(&data_all,"{\"funcId\":\"%s\",\"lcSN\":\"%s\",\"seq\": %d,\"time\":%ld,%s}",FUNCTION_RATE_GET,topic_sn, seq,time(NULL), ex_data.bufptr); ems_syslog(LOG_ERR, "[zxf]Built multi-rate info message: %s", data_all.bufptr); char pub_topic[512] = {0}; sprintf(pub_topic,"emms2/"TYPE_GET_RESP"/%s/%s", topic_sn,FUNCTION_RATE_GET); ems_syslog(LOG_ERR,"报文内容: pub topic [%s] data:[%s]",pub_topic,data_all.bufptr); return 0; } /** * @description: 读取配置 * @param {char} *dev_no * @param {tags_matrix} *matrix: 返回的数据将按规则填充在矩阵中 * @return {*} */ int rate_cfg_read(char *dev_no, struct tags_matrix *matrix) { struct rate_dev_cfg *current = NULL; int timeout; int ret = 0; struct list_head *pos, *lists; proto_forward_t *pro_pr = get_proto_forward_var(); channel_t *chan_pr=NULL; device_t *dev_pr=NULL; struct rate_dev_message *read_item = NULL; int list_count = 0; // current if(find_dev_channel_pr_by_dev_no(pro_pr,dev_no,&chan_pr,&dev_pr) != 0) { ems_syslog(LOG_ERR, "dev_no[%s] not found!", dev_no); return -1; } if(check_rate_support(dev_pr->use_template->vendor, dev_pr->use_template->model) < 1 ) { ems_syslog(LOG_ERR, "No matching device config found"); return -2; } ems_syslog(LOG_DEBUG, "Starting rate configuration write"); struct rate_dev_cfg *tmp_dev_cfg; // 遍历链表查找匹配的设备配置 list_for_each_safe(pos, lists, &rate_cfg_head.list) { tmp_dev_cfg = list_entry(pos, struct rate_dev_cfg, list); if(strcmp(tmp_dev_cfg->vendor, dev_pr->use_template->vendor) == 0 && strcmp(tmp_dev_cfg->model, dev_pr->use_template->model) == 0) { current = tmp_dev_cfg; break; } } if(NULL == current) { ems_syslog(LOG_ERR, "No matching device config found"); return -2; } if(current->read_en == 0) { ems_syslog(LOG_ERR, "Device config write is disabled"); return -3; } list_count = 0; list_for_each_safe(pos, lists, ¤t->read.list ) { list_count++; // 获取报文个数 } struct lnxall_buff *buff = (struct lnxall_buff *)calloc(list_count, sizeof(struct lnxall_buff)); if(!buff) { ems_syslog(LOG_ERR, "Failed to allocate memory for lnxall_buff"); return -3; } // 遍历rate_cfg_head链表 int cnt = 0; struct tags_matrix_Node *rate_node = NULL; int send_hex_len = 0, read_hex_len = 0; timeout = current->delay_time; /*报文组装循环 在读数据中目的是为了测量返回报文的长度*/ // list_for_each_entry(write_item, &rate_cfg_head.read.list, list) { if(0 == strcmp(current->agreement, "DLT645")) { list_for_each_safe(pos, lists, ¤t->read.list ) { read_item = list_entry(pos, struct rate_dev_message, list); ems_syslog(LOG_DEBUG, "Processing device config: type=%d", read_item->type); struct rate_message_node *message_item = NULL; lbuff_init(&buff[cnt], 256); struct list_head *pos_item, *list_item; list_for_each_safe(pos_item, list_item, &read_item->data.list) { message_item = list_entry(pos_item, struct rate_message_node, list); ems_syslog(LOG_DEBUG, "Processing message: type=%d", message_item->type); switch (message_item->type) { case RATE_MSG_NODE_NON: ems_syslog(LOG_ERR, "Unknown message type"); break; case RATE_MSG_NODE_TEXT: lbuff_append(&buff[cnt], message_item->data.message, strlen(message_item->data.message)); break; case RATE_MSG_NODE_TZ_MON: value_mode_transition_645(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_TZ_DAY: value_mode_transition_645(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_TF_HOUR: value_mode_transition_645(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_TF_MIN: value_mode_transition_645(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_TF_MODE: value_mode_transition_645(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_645_CR: value_mode_transition_645(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_645_TAB: value_mode_transition_645(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_PASSWD: break; case RATE_MSG_NODE_RTU_CRC: break; } } cnt++; } } else { list_for_each_safe(pos, lists, ¤t->read.list ) { read_item = list_entry(pos, struct rate_dev_message, list); ems_syslog(LOG_DEBUG, "Processing device config: type=%d", read_item->type); struct rate_message_node *message_item = NULL; lbuff_init(&buff[cnt], 256); struct list_head *pos_item, *list_item; list_for_each_safe(pos_item, list_item, &read_item->data.list) { message_item = list_entry(pos_item, struct rate_message_node, list); ems_syslog(LOG_DEBUG, "Processing message: type=%d", message_item->type); switch (message_item->type) { case RATE_MSG_NODE_NON: ems_syslog(LOG_ERR, "Unknown message type"); break; case RATE_MSG_NODE_TEXT: lbuff_append(&buff[cnt], message_item->data.message, strlen(message_item->data.message)); break; case RATE_MSG_NODE_TZ_MON: value_mode_transition(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_TZ_DAY: value_mode_transition(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_TF_HOUR: value_mode_transition(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_TF_MIN: value_mode_transition(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_TF_MODE: value_mode_transition(&buff[cnt], 0, current->protocol); break; case RATE_MSG_NODE_PASSWD: /*TODO 密码目前无协议样本暂时不予配置实现*/ break; case RATE_MSG_NODE_645_CR: break; case RATE_MSG_NODE_645_TAB: break; case RATE_MSG_NODE_RTU_CRC: break; } } cnt++; } } /*报文组装后下发*/ cnt = 0; // list_for_each_entry(read_item, &rate_cfg_head.read.list, list) // { channel_lock(chan_pr); list_for_each_safe(pos, lists, ¤t->read.list ) { read_item = list_entry(pos, struct rate_dev_message, list); void *hex_data = malloc(strlen(buff[cnt].bufptr) * sizeof(unsigned char) + 2); if(!hex_data) { ems_syslog(LOG_ERR, "Failed to allocate memory for hex data"); ret = 1; channel_unlock(chan_pr); goto FREE_BUFF_ALL; } if(!(chan_pr->s > 0)) { ems_syslog(LOG_ERR, "dev_no[%s] channel is not connected!", dev_no); ret = 1; free(hex_data); channel_unlock(chan_pr); goto FREE_BUFF_ALL; } if(read_item->type == 0) /* 0:send 1:read */ { ems_syslog(LOG_ERR, "dev_no[%s] read data:[%s]", dev_no, buff[cnt].bufptr); send_hex_len = GEN_ARRASIC2HEX((unsigned char *)buff[cnt].bufptr, hex_data, strlen(buff[cnt].bufptr)); if(send_hex_len<=0) { ems_syslog(LOG_ERR, "Failed to convert ASCII to HEX"); ret = 2; free(hex_data); channel_unlock(chan_pr); goto FREE_BUFF_ALL; } if (write(chan_pr->s, hex_data, send_hex_len) < 0) { ems_syslog(LOG_ERR, "dev_no[%s] write data error!: [%s]", dev_no, buff[cnt].bufptr); ret = 3; free(hex_data); channel_unlock(chan_pr); goto FREE_BUFF_ALL; } } else { read_hex_len = GEN_ARRASIC2HEX((unsigned char *)buff[cnt].bufptr, hex_data, strlen(buff[cnt].bufptr)); int len = read_data_timeout(chan_pr->s,hex_data,read_hex_len, timeout); if(len <= 0 ) { ems_syslog(LOG_ERR, "dev_no[%s] read data error!: [%s]", dev_no, buff[cnt].bufptr); ret = 4; free(hex_data); channel_unlock(chan_pr); goto FREE_BUFF_ALL; } memset(buff[cnt].bufptr, 0, buff[cnt].curlen); rate_GEN_ARRHEX2ASIC(hex_data,(unsigned char *)buff[cnt].bufptr,len > buff[cnt].curlen ? buff[cnt].curlen : len, 0); ems_syslog(LOG_ERR, "dev_no[%s] read data: [%s]", dev_no, buff[cnt].bufptr); } free(hex_data); cnt++; } channel_unlock(chan_pr); /*通过已知配置，解析配置内容获取最大行列数，以便基于此创建足够的节点，至于创建的节点是否填写好数值，则是配置问题无需修订代码*/ int row_num = 0; int col_num = 0; list_for_each_safe(pos, lists, ¤t->read.list ) { read_item = list_entry(pos, struct rate_dev_message, list); ems_syslog(LOG_DEBUG, "Processing device config: type=%d", read_item->type); struct rate_message_node *message_item = NULL; struct list_head *pos_item, *list_item; list_for_each_safe(pos_item, list_item, &read_item->data.list) { message_item = list_entry(pos_item, struct rate_message_node, list); if(message_item->type == RATE_MSG_NODE_TZ_MON || message_item->type == RATE_MSG_NODE_TZ_DAY || message_item->type == RATE_MSG_NODE_645_TAB) { if(row_num < message_item->data.time[0]) { row_num = message_item->data.time[0]; } } if(message_item->type == RATE_MSG_NODE_TF_HOUR || message_item->type == RATE_MSG_NODE_TF_MIN || message_item->type == RATE_MSG_NODE_TF_MODE ) { if(row_num < message_item->data.time[0]) { row_num = message_item->data.time[0]; // col_num = matrix->rows[message_item->data.time[0]].Node_num; } if(col_num < message_item->data.time[1]) { col_num = message_item->data.time[1]; } } } } row_num += 1; // 由于矩阵从 0 到 n的数组下标访问, 因此需要+1 col_num += 1; matrix->rows = calloc(sizeof(struct tags_matrix_row) , row_num); if(!matrix->rows) { ems_syslog(LOG_ERR, "Failed to allocate memory for matrix row"); ret = 1; goto FREE_BUFF_ALL; } for(int i = 0; i < row_num; i++) { matrix->rows[i].Node = calloc(sizeof(struct tags_matrix_Node) , col_num); if(!matrix->rows->Node) { ems_syslog(LOG_ERR, "Failed to allocate memory for matrix node"); ret = 1; goto FREE_BUFF_ALL; } matrix->rows[i].Node_num = col_num; } matrix->row_num = row_num; int *tmp_645_tab = NULL; if(0 == strcmp(current->agreement, "DLT645")) { tmp_645_tab = calloc(sizeof(int), matrix->row_num); } /*组装矩阵数据*/ cnt = 0; if(0 == strcmp(current->agreement, "DLT645")) { list_for_each_safe(pos, lists, ¤t->read.list ) { read_item = list_entry(pos, struct rate_dev_message, list); int hex_len = strlen(buff[cnt].bufptr); void *data_tmp = buff[cnt].bufptr; char *data_jump = data_tmp; int move_len = 0, ret_value; struct rate_message_node *message_item = NULL; struct list_head *pos_item, *list_item; ems_syslog(LOG_ERR, " buff[%d].bufptr:%s", cnt, buff[cnt].bufptr); list_for_each_safe(pos_item, list_item, &read_item->data.list) { message_item = list_entry(pos_item, struct rate_message_node, list); if((data_jump - (char *)data_tmp) > hex_len ) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } switch (message_item->type) { case RATE_MSG_NODE_NON: data_jump += strlen(message_item->data.message); move_len = 0; break; case RATE_MSG_NODE_TEXT: data_jump += strlen(message_item->data.message); move_len = 0; break; case RATE_MSG_NODE_TZ_MON: if(message_item->data.time[0] > matrix->row_num) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } move_len = hex_to_value_645(data_jump, &ret_value, current->protocol) ; matrix->rows[message_item->data.time[0]].mon = ret_value; break; case RATE_MSG_NODE_TZ_DAY: if(message_item->data.time[0] > matrix->row_num) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } move_len = hex_to_value_645(data_jump, &ret_value, current->protocol) ; matrix->rows[message_item->data.time[0]].day = ret_value; break; case RATE_MSG_NODE_TF_HOUR: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } move_len = hex_to_value_645(data_jump, &ret_value, current->protocol) ; rate_node->hour = ret_value; break; case RATE_MSG_NODE_TF_MIN: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } move_len = hex_to_value_645(data_jump, &ret_value, current->protocol); rate_node->min = ret_value; break; case RATE_MSG_NODE_TF_MODE: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } move_len = hex_to_value_645(data_jump, &ret_value, current->protocol); rate_node->type = ret_value; break; case RATE_MSG_NODE_645_TAB: { // message_item->data.time[0]; move_len = hex_to_value_645(data_jump, &ret_value, current->protocol) ; tmp_645_tab[message_item->data.time[0]] = ret_value; } break; case RATE_MSG_NODE_645_CR: move_len = hex_to_value(data_jump, &ret_value, current->protocol); break; case RATE_MSG_NODE_PASSWD: break; case RATE_MSG_NODE_RTU_CRC: break; } if(message_item->type > RATE_MSG_NODE_TEXT) { data_jump += move_len; } } cnt++; } } else { list_for_each_safe(pos, lists, ¤t->read.list ) { read_item = list_entry(pos, struct rate_dev_message, list); int hex_len = strlen(buff[cnt].bufptr); void *data_tmp = buff[cnt].bufptr; char *data_jump = data_tmp; int move_len = 0, ret_value; struct rate_message_node *message_item = NULL; struct list_head *pos_item, *list_item; ems_syslog(LOG_ERR, " buff[%d].bufptr:%s", cnt, buff[cnt].bufptr); list_for_each_safe(pos_item, list_item, &read_item->data.list) { message_item = list_entry(pos_item, struct rate_message_node, list); if((data_jump - (char *)data_tmp) > hex_len ) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } switch (message_item->type) { case RATE_MSG_NODE_NON: data_jump += strlen(message_item->data.message); move_len = 0; break; case RATE_MSG_NODE_TEXT: data_jump += strlen(message_item->data.message); move_len = 0; break; case RATE_MSG_NODE_TZ_MON: if(message_item->data.time[0] > matrix->row_num) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } move_len = hex_to_value(data_jump, &ret_value, current->protocol) ; matrix->rows[message_item->data.time[0]].mon = ret_value; break; case RATE_MSG_NODE_TZ_DAY: if(message_item->data.time[0] > matrix->row_num) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } move_len = hex_to_value(data_jump, &ret_value, current->protocol) ; matrix->rows[message_item->data.time[0]].day = ret_value; break; case RATE_MSG_NODE_TF_HOUR: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } move_len = hex_to_value(data_jump, &ret_value, current->protocol) ; rate_node->hour = ret_value; break; case RATE_MSG_NODE_TF_MIN: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } move_len = hex_to_value(data_jump, &ret_value, current->protocol); rate_node->min = ret_value; break; case RATE_MSG_NODE_TF_MODE: rate_node = get_matrix_Node(message_item->data.time[0], message_item->data.time[1], matrix); if(NULL == rate_node) { ems_syslog(LOG_ERR, "Failed to get matrix node [%d][%d]", message_item->type, cnt); ret = 1; goto FREE_BUFF_ALL; } move_len = hex_to_value(data_jump, &ret_value, current->protocol); rate_node->type = ret_value; break; case RATE_MSG_NODE_645_TAB: { // message_item->data.time[0]; move_len = hex_to_value(data_jump, &ret_value, current->protocol) ; if(0 == strcmp(current->agreement, "DLT645")) { tmp_645_tab[message_item->data.time[0]] = ret_value; } } break; case RATE_MSG_NODE_645_CR: move_len = hex_to_value(data_jump, &ret_value, current->protocol); break; case RATE_MSG_NODE_PASSWD: break; case RATE_MSG_NODE_RTU_CRC: break; } if(message_item->type > RATE_MSG_NODE_TEXT) { data_jump += move_len; } } cnt++; } } if(0 == strcmp(current->agreement, "DLT645")) { // 处理表号与矩阵的关联关系 struct tags_matrix_row *row_645 = NULL; row_645 = calloc(sizeof(struct tags_matrix_row), matrix->row_num); memcpy(row_645, matrix->rows, sizeof(struct tags_matrix_row) * matrix->row_num); for (int i = 0; i < matrix->row_num; i++) { if(tmp_645_tab[i] >= 0 && tmp_645_tab[i] < matrix->row_num) { matrix->rows[tmp_645_tab[i]].mon = row_645[i].mon; matrix->rows[tmp_645_tab[i]].day = row_645[i].day; } } free(row_645); free(tmp_645_tab); } FREE_BUFF_ALL: cnt = 0; list_for_each_safe(pos, lists, ¤t->read.list ) { if(buff[cnt].bufptr) lbuff_free(&buff[cnt]); cnt++; } free(buff); return ret; } int validate_format(const char *str) { if (strlen(str) != 5) { return 1; } if (str[2] != ':') { return 2; } if (!isdigit((unsigned char)str[0]) || !isdigit((unsigned char)str[1])) { return 3; } if (!isdigit((unsigned char)str[3]) || !isdigit((unsigned char)str[4])) { return 4; } return 0; } /* return: 0:成功 -1:配置解析失败 -2:设备配置无法下发不支持写入配置 -3:未能找到该电表适配的配置 -4:报文交互失败 */ int rate_mqtt_analysis_cfg(cJSON *root) { if (!root) { ems_syslog(LOG_ERR, "the deal_accident_clear(root == NULL)!!!!"); return -1; } char *info_str_root = cJSON_PrintUnformatted(root); ems_syslog(LOG_ERR, "rate_write 报文内容: %s", info_str_root); struct tags_matrix *tmp_tags_matrix = NULL; // 提取各字段 cJSON* funcId = cJSON_GetObjectItem(root, "funcId"); cJSON* lcSN = cJSON_GetObjectItem(root, "lcSN"); cJSON* seq = cJSON_GetObjectItem(root, "seq"); cJSON* time = cJSON_GetObjectItem(root, "time"); cJSON* info = cJSON_GetObjectItem(root, "info"); if(!info || !funcId || !lcSN || !time || !seq) { ems_syslog(LOG_ERR, "报文解析失败:%p,%p,%p,%p,%p", info ,funcId ,lcSN ,seq ,time); return -1; } cJSON* dev_no = cJSON_GetObjectItem(info, "dev_no"); cJSON* passwd = cJSON_GetObjectItem(info, "passwd"); cJSON* tags = cJSON_GetObjectItem(info, "tags"); if (dev_no) printf("dev_no: %s\n", dev_no->valuestring); if (passwd) printf("passwd: %s\n", passwd->valuestring); if (!cJSON_IsArray(tags)) { ems_syslog(LOG_ERR, "报文解析失败:tags err"); return -1; } proto_forward_t *dev_var = get_proto_forward_var(); if (dev_var == NULL) { ems_syslog(LOG_ERR, ":get_proto_forward_var failed"); return -1; } int ret = 0; // 遍历所有设备和标签 for (int i = 0, m = 0; i < dev_var->channels_size; i++) { channel_t *channel = dev_var->channels[i]; ems_syslog(LOG_DEBUG, "channel:%s", channel->channel); for (int j = 0; j < channel->devs_size; j++, m++) { device_t *dev = channel->devs[j]; if(strcmp(dev->no, dev_no->valuestring)) { continue; } ret = check_rate_support(dev->use_template->vendor, dev->use_template->model); if(ret != 2) { ems_syslog(LOG_ERR, "查找配置结果:[%d] -1: 设备未找到 0: 不支持多费率配置 1: 支持读多费率配置 2: 支持读写多费率配置", ret); return -2; } goto SEND_CFG; } } ems_syslog(LOG_ERR, "查找配置结果: 设备[%s]未找到", dev_no->valuestring); return -2; SEND_CFG: tmp_tags_matrix = (struct tags_matrix *)malloc(sizeof(struct tags_matrix)); int tags_size = cJSON_GetArraySize(tags); tmp_tags_matrix->row_num = tags_size; tmp_tags_matrix->rows = calloc(tags_size, sizeof(struct tags_matrix_row)); cJSON* row = NULL; int row_index = 0; cJSON_ArrayForEach(row, tags) { if (!cJSON_IsArray(row)) continue; int col_index = 0; cJSON* item = NULL; tags_size = cJSON_GetArraySize(row); if(tags_size > 0) { tags_size -= 1; } else { ems_syslog(LOG_ERR, "行 %d 列 %d 时间格式错误: %s", row_index, col_index, item->valuestring); goto FREE_tmp_matrix; } tmp_tags_matrix->rows[row_index].Node = calloc( tags_size, sizeof(struct tags_matrix_Node)); tmp_tags_matrix->rows[row_index].Node_num = tags_size; cJSON_ArrayForEach(item, row) { if (col_index == 0) { if (cJSON_IsArray(item) && cJSON_GetArraySize(item) >= 2) { cJSON* num1 = cJSON_GetArrayItem(item, 0); cJSON* num2 = cJSON_GetArrayItem(item, 1); ems_syslog(LOG_DEBUG, "mon:%p %d day:%p %d", num1, num1 ? num1->valueint : -1, num2, num2 ? num2->valueint : -1); tmp_tags_matrix->rows[row_index].mon = num1 ? num1->valueint : 0; tmp_tags_matrix->rows[row_index].day = num2 ? num2->valueint : 0; } } else { if (cJSON_IsArray(item) && cJSON_GetArraySize(item) >= 2) { cJSON* time = cJSON_GetArrayItem(item, 0); cJSON* value = cJSON_GetArrayItem(item, 1); // char *str_tmp = "0:0"; if(validate_format(time ? time->valuestring : "0:0")) { ems_syslog(LOG_ERR, "不满足\"value:value\"格式 接口下发的json错误,不予下发,需人排查"); goto FREE_tmp_matrix; } // sscanf(str_tmp, "%d, %d", // &tmp_tags_matrix->rows[row_index].Node->hour, &tmp_tags_matrix->rows[row_index].Node->min); // tmp_tags_matrix->rows[row_index].Node->type = value ? value->valueint : 0; int hour = 0, min = 0; if (sscanf(time->valuestring, "%d:%d", &hour, &min) != 2) { ems_syslog(LOG_ERR, "行 %d 列 %d 时间格式错误: %s", row_index, col_index, time->valuestring); continue; } tmp_tags_matrix->rows[row_index].Node[col_index - 1].hour = hour; tmp_tags_matrix->rows[row_index].Node[col_index - 1].min = min; tmp_tags_matrix->rows[row_index].Node[col_index - 1].type = (enum RATE_TYPE)value->valueint; } } col_index++; } row_index++; } // 设置设备编号 memcpy(tmp_tags_matrix->dev_no, dev_no->valuestring, strlen(dev_no->valuestring) > 32 ? 32 :strlen(dev_no->valuestring)); ems_syslog(LOG_DEBUG, "解析配置成功创建配置下发线程"); return rate_send(tmp_tags_matrix); FREE_tmp_matrix: for(int i = 0; i < tmp_tags_matrix->row_num; i++) { if(tmp_tags_matrix->rows[i].Node) { free(tmp_tags_matrix->rows[i].Node); } } free(tmp_tags_matrix->rows); free(tmp_tags_matrix); return -1; } int report_rate_set_return(mqtt_emms2_var_t *var,const char *type,EMMS2_TAG_TYPE tag_type,int seq, int ret) { char pub_topic[128] = ""; struct lnxall_buff buff; lbuff_init(&buff, 64); if (var->emms2_statustopic_sn); lbuff_sprintf(&buff, "{" "\"funcId\":\"SetRate\"," "\"lcSN\": \"%s\"," "\"seq\": %d," "\"time\":%ld," "\"result\": %d}", var->topic_sn, seq, time(NULL), ret); if((ret = mqtt_publish_message(var,pub_topic,buff.bufptr,(int)strlen(buff.bufptr)))!=0) ems_syslog(LOG_WARNING,"send data error!!"); lbuff_free(&buff); return 0; } int rate_send(void * args) { ems_syslog(LOG_INFO, "多费率电表配置下发"); int ret = 0; if(NULL == args) { ems_syslog(LOG_ERR,"资源为NULL不可用"); return -4; } struct tags_matrix *cfg_matrix = (struct tags_matrix *)args; ret = rate_cfg_write( cfg_matrix->dev_no, cfg_matrix); ems_syslog(LOG_WARNING, "多费率电表:%s 配置下发结果: %d",cfg_matrix->dev_no, ret); for(int i = 0; i < cfg_matrix->row_num; i++) { if(cfg_matrix->rows[i].Node) { free(cfg_matrix->rows[i].Node); } } free(cfg_matrix->rows); free(cfg_matrix); return ret == 0 ? 0 : -4; } /** * @brief 检查设备是否支持多费率配置 * @param vendor 厂商标识字符串 * @param mode 设备型号字符串 * @return 费率支持状态： * -1: 设备未找到 * 0: 不支持多费率配置 * 1: 支持读多费率配置 * 2: 支持读写多费率配置 */ int check_rate_support(char *vendor, char *mode) { if (!vendor || !mode) { ems_syslog(LOG_ERR, "[zxf]Invalid input: vendor or mode is NULL"); return -1; } int found = 0; struct list_head *pos, *lists; struct rate_dev_cfg *current; // 遍历链表查找匹配的设备配置 list_for_each_safe(pos, lists, &rate_cfg_head.list ) { current = list_entry(pos, struct rate_dev_cfg, list); if (strcmp(vendor, current->vendor) == 0 && strcmp(mode, current->model) == 0) { found = 1; break; } } if (!found) { ems_syslog(LOG_ERR, "[zxf]Device config not found: vendor=%s, mode=%s", vendor, mode); return -1; } // 根据读写使能标志返回对应状态 if (current->read_en && current->write_en) { return 2; // 支持读写 } else if (current->read_en) { return 1; // 仅支持读 } else { return 0; // 不支持 } } /** * @brief 更新设备json信息中的费率配置 * @param dev_info 设备信息JSON对象 * @return 更新后的JSON对象 */ cJSON *update_dev_rate_info(cJSON *dev_info) { if (!dev_info) { return NULL; } // // 获取info对象 // cJSON *info = cJSON_GetObjectItemCaseSensitive(dev_info, "info"); // if (!info || !cJSON_IsObject(info)) { // ems_syslog(LOG_ERR, "Invalid or missing 'info' field"); // return dev_info; // } // 获取dev_class数组 cJSON *dev_class = cJSON_GetObjectItemCaseSensitive(dev_info, "dev_class"); if (!dev_class || !cJSON_IsArray(dev_class)) { ems_syslog(LOG_ERR, "[zxf]No valid 'dev_class' array found in dev_group_info"); return dev_info; } cJSON *class_item = NULL; cJSON_ArrayForEach(class_item, dev_class) { if (!cJSON_IsObject(class_item)) continue; // 获取devs数组 cJSON *devs = cJSON_GetObjectItemCaseSensitive(class_item, "devs"); if (!devs || !cJSON_IsArray(devs)) continue; // 遍历每个设备 cJSON *dev = NULL; cJSON_ArrayForEach(dev, devs) { if (!cJSON_IsObject(dev)) continue; // 获取vendor和model字段 cJSON *vendor = cJSON_GetObjectItemCaseSensitive(dev, "vendor"); cJSON *model = cJSON_GetObjectItemCaseSensitive(dev, "model"); if (!vendor || !cJSON_IsString(vendor) || !model || !cJSON_IsString(model)) { continue; } // 检查是否已存在rate字段（避免重复处理） if (cJSON_GetObjectItemCaseSensitive(dev, "rate")) { continue; } // 检查设备是否支持多费率配置 int rate_support = check_rate_support(vendor->valuestring, model->valuestring); // 仅当匹配上对应设备才添加rate字段（-1不添加） if (rate_support >= 0) { cJSON_AddNumberToObject(dev, "rate", rate_support); ems_syslog(LOG_ERR, "[zxf]Added rate=%d to device: vendor=%s, model=%s", rate_support, vendor->valuestring, model->valuestring); } } } return dev_info; } /*test代码段*/ int rate_test() { sleep(5); const char *cfg_str = read_file_data("/app/config/rate_test.json"); if (!cfg_str) { ems_syslog(LOG_ERR, "Failed to read rate config file: %s", RATECFG_PATH); exit(0); return -1; } struct timeval tv; gettimeofday(&tv, NULL); milliseconds = tv.tv_sec * 1000 + tv.tv_usec / 1000; // cJSON *root = cJSON_Parse(cfg_str); // rate_mqtt_analysis_cfg(root); sleep(5); struct tags_matrix matrix_tmp; rate_cfg_read("PCS_METER", &matrix_tmp); rate_zhuzhuang_mqtt_message(12, &matrix_tmp); return 0; }