import asyncio import copy import re import threading import numpy as np from andon_flask_client import SimpleAndonSender from calculate import BatteryStackAnalyzer from dataTransmission import data_transmission from plc_client import * from dao.dbutils import * import time import flet as ft import struct from typing import List, Tuple, Dict, Any, Optional import requests from collections import defaultdict import sys sys.path.append(r'D:\battery') # 创建 BatteryDatabase 实例 battery_db = BatteryDatabase() sender = SimpleAndonSender() print("=== Andon 发送端测试 ===") class PlcDataProcessor: def __init__(self, ip_address: str, port: int, tcp_host: str = "0.0.0.0", tcp_port: int = 1024): self.plc = PlcClient(ip_address, port) self.real_time_data = [] # 示例数据 self._get_130_data = [] self.read_attempts = 5 # 读取尝试次数 self.battery_data_with_list_130 = [] # 130数据录入缓存 self.battery_data_with_list_566 = [] # 566据录入缓存 self.battery_data_with_list_130 = battery_db.get_latest_battery_data(limit=2000) self.tcp_host = tcp_host self.tcp_port = tcp_port self.server_socket = None self.is_running = False self.data_manager = [] self.temp_data = '' self.sel_version = '2.0' self.run_thread() # def read_and_process_data(self, page: ft.Page): # """主循环函数，读取并处理PLC数据""" # while True: # try: # # 发送实时数据到UI页面 # if page: # page.pubsub.send_all(self.real_time_data) # self._process_cycle() # except Exception as e: # print(f"发生错误: {e}") # time.sleep(3) def read_and_process_data(self, page: ft.Page): """主循环函数，读取并处理PLC数据""" while True: try: # 获取不同类型的数据 data_130 = self._get_130_data real_time_data = self.real_time_data # 发送到不同的消息频道 if page: # 发送130数据到特定频道 page.pubsub.send_all({ 'message_type': 'data_130', 'payload': data_130, 'timestamp': time.time() }) page.pubsub.send_all({ 'message_type': 'real_time', 'payload': real_time_data, 'timestamp': time.time() }) self._process_cycle() except Exception as e: print(f"发生错误: {e}") time.sleep(3) def _get_updated_real_time_data(self, shen_7, data_all): """获取更新的实时数据""" # 这里可以根据实际需求从PLC或其他数据源获取实时数据 # 示例：返回当前时间戳和一些模拟数据 return {'shen_7': shen_7, 'all': data_all} def read_main_data(self): """主循环函数，读取并处理PLC数据""" while True: self._process_cycle() time.sleep(3) def _process_cycle(self): """处理一个完整的数据读取周期""" m_values, err = self.plc.read_multiple_m(1000, 25) print(m_values, "mmmmmmmmmm") ok, errr = self.plc.write_m(1003, False) if ok: print(f"M1003 写入成功") if err: print("读取M值错误") return # 一次性发送所有状态 sender.send_mes_thread(m_values) if m_values[0]: # 处理不同的M值条件 sel_version = '' # 是否上料130片 sel_566_version = '' # 是否上料566片 if m_values[5]: sel_version = '2.0' self.sel_version = '2.0' elif m_values[6]: sel_version = '2.5' self.sel_version = '2.5' if sel_version == '2.0': bat_data = self._process_zr_data(sel_version) print('bat_data', bat_data) if bat_data is not None: P_A_130, P_C_130, source_code_130 = BatteryStackAnalyzer.convert_battery_data_simple_130( parse_all_battery_data_130(bat_data)) T_10_130 = [1.0] * len(P_A_130) # 使用默认厚度值 analyzer_130 = BatteryStackAnalyzer(P_A_130, P_C_130, T_10_130, source_code_130) results_130 = analyzer_130.analyze_all() data_list = (results_130['all_invalid_positions']["阳极压损不满足"] + results_130['all_invalid_positions']["阴极压损不满足"] + results_130['all_invalid_positions']["精度检测不满足"]) if data_list: self.plc.write_m(1026, True) self.data_write_130(1000, list(set(data_list))) self.data_write_130(1800, results_130['all_invalid_positions']["重复数据"]) else: # 写入数据 battery_data_130 = parse_all_battery_data_130(bat_data) self._get_130_data = battery_data_130 list_data = battery_data_130['box_data_1']['battery_data_1'] + \ battery_data_130['box_data_2']['battery_data_2'] list_data = list_data[::-1] battery_data_130_save = { 'cah_no': '', 'box_data_1': { 'box_data_1_no': '', 'battery_data_1': [] }, 'box_data_2': { 'box_data_2_no': '', 'battery_data_2': [] }, } battery_data_with = [] for i in range(1, 131): item = list_data[i - 1] # 创建基础字典 new_item = { 'code': item.get('serial_number', ''), 'top_qr_code': item.get('Up_QR', ''), 'bottom_qr_code': item.get('Below_QR', ''), 'h2_value': item.get('H2', 0), 'o2_value': item.get('02', 0), 'symbol': item.get('Cell_ran', ''), 'thickness': item.get('thick', ''), 'notes': item.get('source_code', ''), 'magazine_code': battery_data_130['cah_no'], 'magazine_index': i, } new_item_save = { 'serial_number': item.get('serial_number', ''), 'Up_QR': item.get('Up_QR', ''), 'Below_QR': item.get('Below_QR', ''), 'H2': item.get('H2', 0), '02': item.get('02', 0), 'Cell_ran': item.get('Cell_ran', ''), 'thick': item.get('thick', ''), 'source_code': item.get('source_code', ''), 'magazine_index': i, } battery_data_130_save['cah_no'] = battery_data_130['cah_no'] battery_data_130_save['box_data_1']['box_data_1_no'] = battery_data_130['box_data_1'][ 'box_data_1_no'] battery_data_130_save['box_data_2']['box_data_2_no'] = battery_data_130['box_data_2'][ 'box_data_2_no'] # 添加box_code if i <= 65: new_item['box_code'] = battery_data_130['box_data_1']['box_data_1_no'] new_item['box_index'] = 1 new_item_save['box_index'] = 1 battery_data_130_save['box_data_1']['battery_data_1'].append(new_item_save) else: new_item['box_code'] = battery_data_130['box_data_2']['box_data_2_no'] new_item['box_index'] = 2 new_item_save['box_index'] = 2 battery_data_130_save['box_data_2']['battery_data_2'].append(new_item_save) battery_data_with.append(new_item) if len(self.battery_data_with_list_130) < 3000: self.battery_data_with_list_130.extend(battery_data_with) else: self.battery_data_with_list_130.pop(0) self.battery_data_with_list_130.extend(battery_data_with) try: magazine_id = battery_db.create_magazine( magazine_code=battery_data_130_save['cah_no'], description="" ) box_data = [battery_data_130_save["box_data_1"]["box_data_1_no"], battery_data_130_save["box_data_2"]["box_data_2_no"]] box_ids = [] for desc in box_data: try: box_id = battery_db.create_box(magazine_id, desc) box_ids.append(box_id) except ValueError as e: print(f"{desc} 创建失败: {e}") battery_data = [ {"id": box_ids[0], "data": battery_data_130_save['box_data_1']['battery_data_1']}, {"id": box_ids[1], "data": battery_data_130_save['box_data_2']['battery_data_2']}, ] for battery in battery_data: battery_db.add_batch_batteries( battery['id'], battery['data'] ) print("写入数据") except ValueError as e: print(f"创建失败: {e}") self.plc.write_m(1005, False) self.plc.write_m(1022, True) elif sel_version == '2.5': bat_data = self._process_zr_data(sel_version) if bat_data is not None: P_A_130, P_C_130, source_code_130 = BatteryStackAnalyzer.convert_battery_data_simple_130( parse_all_battery_data_130(bat_data)) T_10_130 = [1.0] * len(P_A_130) # 使用默认厚度值 analyzer_130 = BatteryStackAnalyzer(P_A_130, P_C_130, T_10_130, source_code_130) results_130 = analyzer_130.analyze_all() data_list = (results_130['all_invalid_positions']["阳极压损不满足"] + results_130['all_invalid_positions']["阴极压损不满足"] + results_130['all_invalid_positions']["精度检测不满足"]) if data_list: self.plc.write_m(1027, True) self.data_write_130(2000, list(set(data_list))) self.data_write_130(2800, results_130['all_invalid_positions']["重复数据"]) else: # 写入数据 battery_data_130 = parse_all_battery_data_130(bat_data) self._get_130_data = battery_data_130 list_data = battery_data_130['box_data_1']['battery_data_1'] + \ battery_data_130['box_data_2']['battery_data_2'] list_data = list_data[::-1] battery_data_130_save = { 'cah_no': '', 'box_data_1': { 'box_data_1_no': '', 'battery_data_1': [] }, 'box_data_2': { 'box_data_2_no': '', 'battery_data_2': [] }, } battery_data_with = [] for i in range(1, 131): item = list_data[i - 1] # 创建基础字典 new_item = { 'code': item.get('serial_number', ''), 'top_qr_code': item.get('Up_QR', ''), 'bottom_qr_code': item.get('Below_QR', ''), 'h2_value': item.get('H2', 0), 'o2_value': item.get('02', 0), 'symbol': item.get('Cell_ran', ''), 'thickness': item.get('thick', ''), 'notes': item.get('source_code', ''), 'magazine_code': battery_data_130['cah_no'], 'magazine_index': i, } new_item_save = { 'serial_number': item.get('serial_number', ''), 'Up_QR': item.get('Up_QR', ''), 'Below_QR': item.get('Below_QR', ''), 'H2': item.get('H2', 0), '02': item.get('02', 0), 'Cell_ran': item.get('Cell_ran', ''), 'thick': item.get('thick', ''), 'source_code': item.get('source_code', ''), 'magazine_index': i, } battery_data_130_save['cah_no'] = battery_data_130['cah_no'] battery_data_130_save['box_data_1']['box_data_1_no'] = battery_data_130['box_data_1'][ 'box_data_1_no'] battery_data_130_save['box_data_2']['box_data_2_no'] = battery_data_130['box_data_2'][ 'box_data_2_no'] # 添加box_code if i <= 65: new_item['box_code'] = battery_data_130['box_data_1']['box_data_1_no'] new_item['box_index'] = 1 new_item_save['box_index'] = 1 battery_data_130_save['box_data_1']['battery_data_1'].append(new_item_save) else: new_item['box_code'] = battery_data_130['box_data_2']['box_data_2_no'] new_item['box_index'] = 2 new_item_save['box_index'] = 2 battery_data_130_save['box_data_2']['battery_data_2'].append(new_item_save) battery_data_with.append(new_item) if len(self.battery_data_with_list_130) < 3000: self.battery_data_with_list_130.extend(battery_data_with) else: self.battery_data_with_list_130.pop(0) self.battery_data_with_list_130.extend(battery_data_with) try: magazine_id = battery_db.create_magazine( magazine_code=battery_data_130_save['cah_no'], description="" ) box_data = [battery_data_130_save["box_data_1"]["box_data_1_no"], battery_data_130_save["box_data_2"]["box_data_2_no"]] box_ids = [] for desc in box_data: try: box_id = battery_db.create_box(magazine_id, desc) box_ids.append(box_id) except ValueError as e: print(f"{desc} 创建失败: {e}") battery_data = [ {"id": box_ids[0], "data": battery_data_130_save['box_data_1']['battery_data_1']}, {"id": box_ids[1], "data": battery_data_130_save['box_data_2']['battery_data_2']}, ] for battery in battery_data: battery_db.add_batch_batteries( battery['id'], battery['data'] ) print("写入数据") except ValueError as e: print(f"创建失败: {e}") self.plc.write_m(1006, False) self.plc.write_m(1023, True) if m_values[4]: if m_values[7]: sel_566_version = '2.0' elif m_values[8]: sel_566_version = '2.5' if sel_566_version == '2.0': data_566 = self._process_zr_566_data(sel_566_version) print('data_566',data_566) if data_566 and data_566.get('cah_data'): data_566['cah_data'] = [item.replace('\x00', '') for item in data_566['cah_data']] P_A_566, P_C_566, source_code_566 = BatteryStackAnalyzer.convert_battery_data_simple_566( parse_all_battery_data_566(data_566)) T_10_566 = [1.0] * len(P_A_566) # 使用默认厚度值 analyzer_566 = BatteryStackAnalyzer(P_A_566, P_C_566, T_10_566, source_code_566) results_566 = analyzer_566.analyze_all() # 修复后的代码 result_data566 = [] # 使用可哈希的字段作为键，比如source_code dj_lookup = {item['notes']: item for item in self.battery_data_with_list_130} for sublist in parse_all_battery_data_566(data_566)['cah_data']: # sublist现在是字典，我们需要用它的某个字段来匹配 # 根据你的数据结构，应该使用source_code字段 source_code = sublist.get('source_code', '') if source_code in dj_lookup: matched_dj = dj_lookup[source_code] result_data566.append(matched_dj) # 按 magazine_code 分组 grouped_data = defaultdict(list) for item in result_data566: grouped_data[item['magazine_code']].append(item) # 对每个杂志内的数据倒序排列 result = [] for magazine_code, items in grouped_data.items(): reversed_items = list(reversed(items)) result.extend(reversed_items) print('result_data566', result) if results_566["低压电池"] != '满足条件的单电池数量不足7片（只有0片）': self.real_time_data = self._get_updated_real_time_data(results_566["低压电池"], result) self.shen_7_data_write(results_566["低压电池"]) self.plc.write_m(1024, True) elif sel_566_version == '2.5': data_566 = self._process_zr_566_data(sel_566_version) if data_566 and data_566.get('cah_data'): data_566['cah_data'] = [item.replace('\x00', '') for item in data_566['cah_data']] P_A_566, P_C_566, source_code_566 = BatteryStackAnalyzer.convert_battery_data_simple_566( parse_all_battery_data_566(data_566)) T_10_566 = [1.0] * len(P_A_566) # 使用默认厚度值 analyzer_566 = BatteryStackAnalyzer(P_A_566, P_C_566, T_10_566, source_code_566) results_566 = analyzer_566.analyze_all() result_data566 = [] dj_lookup = {item['notes']: item for item in self.battery_data_with_list_130} for sublist in parse_all_battery_data_566(data_566)['cah_data']: matched_dj = dj_lookup[sublist] if sublist in dj_lookup: result_data566.append(matched_dj) # 按 magazine_code 分组 grouped_data = defaultdict(list) for item in result_data566: grouped_data[item['magazine_code']].append(item) # 对每个杂志内的数据倒序排列 result = [] for magazine_code, items in grouped_data.items(): reversed_items = list(reversed(items)) result.extend(reversed_items) if results_566["低压电池"] != '满足条件的单电池数量不足7片（只有0片）': self.real_time_data = self._get_updated_real_time_data(results_566["低压电池"], result) self.shen_7_data_write(results_566["低压电池"]) self.plc.write_m(1024, True) if m_values[9]: print('data_566_2发送') data_566_2 = self._process_zr_566_manage_data("2.0") print('data_566_2', data_566_2) if data_566_2 and data_566_2.get('cah_data'): data_566_2['cah_data'] = [item.replace('\x00', '') for item in data_566_2['cah_data']] result = {'cah_data': []} for index, data_string in enumerate(data_566_2['cah_data']): # 每46个字符分一份 chunks = [data_string[i:i + 46] for i in range(0, len(data_string), 46)] if index < 69: chunks = chunks[:8] else: chunks = chunks[:7] result['cah_data'].append(chunks) self.battery_data_with_list_566 = parse_all_battery_data_566_2(result)['cah_data'] # 首先将嵌套的dj展平为一维列表 dj_lookup = {item['notes']: item for item in self.battery_data_with_list_130} # 处理dj1中的数据 for sublist in self.battery_data_with_list_566: # 遍历dj1中的每个子列表 for item in sublist: # 遍历子列表中的每个字典 source_code = item['source_code'] if source_code in dj_lookup: # 获取dj中对应的完整数据 matched_dj = dj_lookup[source_code] # 将dj中多出的字段添加到dj1 item['magazine_code'] = matched_dj['magazine_code'] item['magazine_index'] = matched_dj['magazine_index'] item['box_code'] = matched_dj['box_code'] item['box_index'] = matched_dj['box_index'] data_566_2_list = {'cah_data': self.battery_data_with_list_566} try: # 创建批次 batch_id = battery_db.create_batch( batch_name='', description="" ) # 定义分组规则 groups = list(range(1, 72)) # 对应的group_id # 准备批量数据 battery_data_list = [] battery_data_with_groups = [] data_index = 1 # 遍历所有数据行 for i, battery_row in enumerate(data_566_2_list['cah_data']): # 跳过空行 if not battery_row: continue # 分配group_id（如果组数不够，使用默认值1） group_id = groups[i] if i < len(groups) else 1 # 遍历每行中的电池数据 for battery_encoded in battery_row[::-1]: # 解析电池数据 battery_data = self.parse_battery_data(battery_encoded) if battery_data: # 准备电池数据字典 battery_dict = { 'code': battery_data['serial_number'], 'top_qr_code': battery_data.get('Below_QR', '').strip(), 'bottom_qr_code': battery_data.get('Up_QR', '').strip(), 'h2_value': float(battery_data.get('H2', 0)), 'o2_value': float(battery_data.get('O2', 0)), 'symbol': battery_data.get('Cell_ran', ''), 'thickness': float(battery_data.get('thick', 0)), 'notes': battery_data.get('source_code', ''), 'status': 'normal', 'data_index': data_index, 'box_code': battery_data['box_code'], 'box_index': float(battery_data.get('box_index', 0)), 'magazine_code': battery_data['magazine_code'], 'magazine_index': float(battery_data.get('magazine_index', 0)), } # 添加到批量列表 battery_data_list.append(battery_dict) battery_data_with_groups.append((battery_dict, group_id)) data_index += 1 record_ids = battery_db.add_save_battery_data_batch_with_groups(batch_id, battery_data_with_groups) print(f"成功批量添加 {len(record_ids)} 条电池数据到数据库") data_list = [] P_A = [] # 阳极压损 - 使用H2字段 P_C = [] # 阴极压损 - 使用02字段 for battery_data, group_id in battery_data_with_groups: P_A.append(battery_data['h2_value']) P_C.append(battery_data['o2_value']) P_A_ww = np.mean(P_A) P_C_ww = np.mean(P_C) for battery_data, group_id in battery_data_with_groups: is_shen7 = False if battery_data['data_index'] > (len(battery_data_with_groups) - 7): is_shen7 = True PA_ratios_h2, PC_ratios_02, PA_invalid_indices, PC_invalid_indices = check_pressure_requirements( battery_data['h2_value'], battery_data['o2_value'], P_A_ww, P_C_ww) data_list.append( {'cell_counter': int(battery_data['magazine_index']), "separator1": battery_data['bottom_qr_code'], "separator2": battery_data['top_qr_code'], "delta_P_h2": battery_data['h2_value'], "delta_P_o2": battery_data['o2_value'], "cell_rank": battery_data['symbol'], "thickness": battery_data['thickness'], "god_seven_mark": is_shen7, "cell_index": battery_data['data_index'], "h2_compute": round(float(PA_ratios_h2), 2), "h2_result": PA_invalid_indices, "o2_compute": round(float(PC_ratios_02), 2), "o2_result": PC_invalid_indices, 'clip_barcode': battery_data["magazine_code"] }) asyncio.run(data_transmission(data_list)) self.battery_data_with_list_566 = [] print('顺序566保存完成') self.plc.write_m(1025, True) except KeyError as e: print(f"数据字段错误: {e}") except Exception as e: print(f"添加电池数据时发生错误: {e}") import traceback print(f"详细错误信息: {traceback.format_exc()}") # 重置标志位 self.plc.write_m(1000, False) def parse_battery_data(self, encoded_string): """ 解析电池数据编码字符串 """ try: parsed_data = { 'serial_number': encoded_string['serial_number'], 'Below_QR': encoded_string['Below_QR'], 'Up_QR': encoded_string['Up_QR'], 'H2': encoded_string['H2'], 'O2': encoded_string['02'], 'Cell_ran': encoded_string['Cell_ran'], 'thick': encoded_string['thick'], 'box_code': encoded_string['box_code'], 'box_index': encoded_string['box_index'], 'source_code': encoded_string['source_code'], 'magazine_code': encoded_string['magazine_code'], 'magazine_index': encoded_string['magazine_index'], } return parsed_data except Exception as e: print(f"解析电池数据时发生错误: {e}") return None def _process_zr_data(self, version): """读取并处理 不同版本 弹夹位置""" if version == '2.0': address = 200000 count = 4100 elif version == '2.5': address = 205000 count = 4100 else: return None # 单次最多读 500 zr_data = [] for i in range(0, count, 500): if i < count - 500: temp_data, _ = self.plc.read_r_batch(address + i, 500) else: temp_data, _ = self.plc.read_r_batch(address + i, count - i) if temp_data: zr_data.extend(temp_data) else: print(f"ZR{address + i} 批量读取失败") return None print(f"ZR{address} 原始数据长度: {len(zr_data)}") if not zr_data: print(f"读取ZR{address}数据失败") return None # 分别提取不同参数的数据 cah_data = self.connect_data(zr_data, 1, 0, count=100) # 弹夹数据 cah_data = cah_data.replace("\r", "").replace("\x00", "").replace('\x07', '') temp_str = ''.join(zr_data) data_string = temp_str.replace('\x00', '').replace('\r', '').replace('\n', '').replace('â\x01', '') json_datas = split_json_objects(data_string) if len(json_datas) != 4: return None box_data_1_no = json_datas[0].get("header", None) battery_data_1 = [] if box_data_1_no: box_data_1_no = box_data_1_no.get("filename", None) battery_data_1_1 = json_datas[0].get('data') battery_data_1_2 = json_datas[1].get('data') if battery_data_1_1 and battery_data_1_2: battery_data_1 = battery_data_1_1 + battery_data_1_2 battery_data_1 = [battery_data_1[i:i + 46] for i in range(0, len(battery_data_1), 46)] box_data_2_no = json_datas[2].get("header", None) if box_data_2_no: box_data_2_no = box_data_2_no.get("filename", None) battery_data_2_1 = json_datas[2].get('data') battery_data_2_2 = json_datas[3].get('data') battery_data_2 = [] if battery_data_2_1 and battery_data_2_2: battery_data_2 = battery_data_2_1 + battery_data_2_2 battery_data_2 = [battery_data_2[i:i + 46] for i in range(0, len(battery_data_2), 46)] data = {"cah_no": cah_data, "box_data_1": { "box_data_1_no": box_data_1_no, "battery_data_1": battery_data_1}, "box_data_2": { "box_data_2_no": box_data_2_no, "battery_data_2": battery_data_2}, } return data def _process_zr_566_data(self, version): """读取并处理 566 数据""" if version == '2.0': address = 210000 count = 13019 elif version == '2.5': address = 230000 count = 13019 else: return None # 单次最多读 500 zr_data_566 = [] for i in range(0, count, 500): if i < count - 500: temp_data, _ = self.plc.read_r_batch(address + i, 500) else: temp_data, _ = self.plc.read_r_batch(address + i, count - i) if temp_data: zr_data_566.extend(temp_data) else: print(f"ZR{address + i} 批量读取失败") if not zr_data_566: print(f"读取ZR{address}数据失败") return None # 提取 电池信息 cah_data = self.connect_data_2_list(zr_data_566, 23, 0, count=13019) # 弹夹数据 data = { "cah_data": cah_data } # 添加到结果列表 return data def _process_zr_566_manage_data(self, version): """ 566 数据 处理后 有分组""" if version == '2.0': address = 250000 count = 14201 elif version == '2.5': address = 230000 count = 13018 else: return None # 单次最多读 500 zr_data_566 = [] for i in range(0, count, 500): if i < count - 500: temp_data, _ = self.plc.read_r_batch(address + i, 500) else: temp_data, _ = self.plc.read_r_batch(address + i, count - i) if temp_data: zr_data_566.extend(temp_data) else: print(f"ZR{address + i} 批量读取失败") if not zr_data_566: print(f"读取ZR{address}数据失败") return None # 提取 电池信息 cah_data = self.connect_data_2_list(zr_data_566, 200, 0, count=200 * 69 + 1) cah_data1 = self.connect_data_2_list(zr_data_566, 200, 200 * 69, count=401) # 弹夹数据 cah_data.extend(cah_data1) data = { "cah_data": cah_data } # 添加到结果列表 return data @staticmethod def extract_data_pairs(zr_data: List[str], interval: int = 25, start_offset: int = 15, count: int = 2) -> List[Dict[str, Any]]: """ 按照指定间隔和起始偏移提取多个数据 参数: zr_data: 原始数据列表 interval: 数据间隔 start_offset: 起始偏移位置 count: 要提取的数据数量 返回格式: [{'values': [v1, v2, v3], 'positions': [p1, p2, p3], 'float_values': [f1, f2, f3]}, ...] """ extracted_groups = [] max_groups = (len(zr_data) - start_offset - count + 1) // interval for group_index in range(max_groups + 1): base_index = group_index * interval positions = [base_index + start_offset + i for i in range(count)] # 检查所有位置是否有效 if all(pos < len(zr_data) for pos in positions): values = [zr_data[pos] for pos in positions] # 对每两个相邻的值进行浮点数转换（如果有偶数个值） float_values = [] for i in range(0, len(values) - 1, 2): float_val = PlcDataProcessor.combine_to_float(values[i], values[i + 1]) float_values.append(float_val) extracted_groups.append({ 'values': values, 'positions': positions, 'float_values': float_values, 'group_index': group_index }) return extracted_groups @staticmethod def connect_data(zr_data: List[str], interval: int = 25, start_offset: int = 15, count: int = 2): """ 按照指定间隔和起始偏移提取多个数据 参数: zr_data: 原始数据列表 interval: 数据间隔 start_offset: 起始偏移位置 count: 要提取的数据数量 返回格式: [{'values': [v1, v2, v3], 'positions': [p1, p2, p3], 'float_values': [f1, f2, f3]}, ...] """ end_offset = start_offset + count if (start_offset + count) < len(zr_data) else len(zr_data) values = '' for group_index in range(start_offset, end_offset): values += str(zr_data[group_index]) return values @staticmethod def connect_data_2_list(zr_data: List[str], interval, start_offset, count: int = 2): """ 按照指定间隔和起始偏移提取多个数据 参数: zr_data: 原始数据列表 interval: 数据间隔 start_offset: 起始偏移位置 count: 要提取的数据数量 """ end_offset = start_offset + count if (start_offset + count) < len(zr_data) else len(zr_data) data_list = [] index = 0 values = '' for group_index in range(start_offset, end_offset): if index % interval == 0 and group_index != start_offset: data_list.append(values) values = '' values += str(zr_data[group_index]) index += 1 return data_list @staticmethod def combine_to_int32(low_word: int, high_word: int) -> int: """将两个16位整数合成32位有符号整数""" return (high_word << 16) | (low_word & 0xFFFF) @staticmethod def combine_to_float(low_word: int, high_word: int) -> float: """将两个16位整数合成32位浮点数""" int32_value = PlcDataProcessor.combine_to_int32(low_word, high_word) try: float_bytes = struct.pack('>I', int32_value & 0xFFFFFFFF) return struct.unpack('>f', float_bytes)[0] except (struct.error, TypeError): return float('nan') @staticmethod def data_calculate(): pass @staticmethod # 数据保存 def data_saving(batch_name: str, description: str): # 创建数据库实例 db = BatteryDatabase() # 创建新批次 batch_id = db.create_batch(batch_name, description) print(f"创建批次成功，批次ID: {batch_id}") return batch_id def shen_7_data_write(self, shen_7_data): # shen_7_data: 7片低压QR数据 shen_7_data_str = '' for qr_data in shen_7_data: shen_7_data_str += qr_data res, _ = self.plc.write_d_shen_7_batch(start_address=1200, values=shen_7_data_str) if res: print("shen 7 低压损2.0写人成功") else: print("shen 7 低压损2.0写人失败") def shen_7_data_write_25(self, shen_7_data): # shen_7_data: 7片低压QR数据 shen_7_data_str = '' for qr_data in shen_7_data: shen_7_data_str += qr_data res, _ = self.plc.write_d_shen_7_batch(start_address=2200, values=shen_7_data_str) if res: print("shen 7 低压损2.5写人成功") else: print("shen 7 低压损2.5写人失败") def data_write_130(self, start_address, data): # shen_7_data: 7片低压QR数据 data_str = '' for qr_data in data: data_str += qr_data res, _ = self.plc.write_d_130_batch(start_address=start_address, values=data_str) if res: print("shen 7 上料扫码NG2.0人成功") else: print("shen 7 上料扫码NG2.0写人失败") # def shen_7_data_write_13025(self, shen_7_data): # # shen_7_data: 7片低压QR数据 # shen_7_data_str = '' # for qr_data in shen_7_data: # shen_7_data_str += qr_data # res, _ = self.plc.write_d_shen_7_batch(start_address=2000, values=shen_7_data_str) # if res: # print("shen 7 上料扫码NG2.5人成功") # else: # print("shen 7 上料扫码NG2.5写人失败") # 下方为TCP 服务器端程序 def handle_client(self, client_socket, client_address): """处理客户端连接""" client_id = f"{client_address[0]}:{client_address[1]}" logger.info(f"客户端连接: {client_id}") client_ip = client_address[0] try: while self.is_running: # 根据ip将不同数据进行筛选 if (self.sel_version == '2.0' and (client_ip == '192.168.2.14')) \ or (self.sel_version == '2.5' and (client_ip == '192.168.2.15')): data = client_socket.recv(2048) logger.info(f"收到数据: {data}") if not data: break recv_content = data.decode("utf-8") self.temp_data += recv_content if recv_content.endswith("\r"): # 结束符 # 数据处理 往plc中写入 todo # 2.0丛D4001开始 到 D5999 # 2.5从D6000开始 到 D7999 self.write_130_batch(self.temp_data, self.sel_version) send_content = "recv ok" # 对字符串进行编码成为二进制数据 response = send_content.encode("utf-8") if response: client_socket.send(response) else: # 默认响应 client_socket.send(f"收到: {recv_content}".encode("utf-8")) except Exception as e: logger.error(f"客户端处理错误: {e}") finally: client_socket.close() logger.info(f"客户端断开: {client_id}") def run(self): """运行服务器""" self.server_socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM) self.server_socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) self.server_socket.bind((self.tcp_host, self.tcp_port)) self.server_socket.listen(128) self.server_socket.settimeout(1.0) self.is_running = True logger.info(f"服务器启动: {self.tcp_host}:{self.tcp_port}") try: while self.is_running: try: client_socket, client_address = self.server_socket.accept() thread = threading.Thread( target=self.handle_client, args=(client_socket, client_address), daemon=True ) thread.start() except socket.timeout: continue except KeyboardInterrupt: logger.info("服务器关闭") finally: self.stop() def write_130_batch(self, write_data, sel_version): # 写入后需要改变 标记位寄存器的值, 写入前需要查询标记位数寄存器 的值 # print(write_data) if self.sel_version == '2.0': m_values, err = self.plc.read_multiple_m(1013, 1) elif self.sel_version == '2.5': m_values, err = self.plc.read_multiple_m(1014, 1) else: m_values = None print('&&&&&') print(m_values) if m_values and m_values[0] == 1: temp_list = [] if len(write_data) > 1000: for i in range(0, len(write_data), 1000): temp_list.append(write_data[i:i+1000]) else: temp_list.append(write_data) for index, data_item in enumerate(temp_list): if sel_version == '2.0': res, _ = self.plc.write_d_130_ascii_batch(start_address=4001 + index * 500, values=data_item) if res: print("130 上料扫码NG2.0人成功") else: print("130 上料扫码NG2.0写人失败") d4038, d_err = self.plc.read_d(4038) if d_err: print(f"回读 D4038 失败: {d_err}") else: c1 = chr(d4038 & 0xFF) if 32 <= (d4038 & 0xFF) < 127 else '.' c2 = chr((d4038 >> 8) & 0xFF) if 32 <= ((d4038 >> 8) & 0xFF) < 127 else '.' print(f"回读 D4038: raw={d4038} hex={d4038:04X} ascii='{c1}{c2}'") elif sel_version == '2.5': res, _ = self.plc.write_d_130_ascii_batch(start_address=6000 + index * 500, values=data_item) if res: print("130 上料扫码NG2.5人成功") else: print("130 上料扫码NG2.5写人失败") elif m_values and m_values[0] == 0: temp_list = [] if len(write_data) > 1000: for i in range(0, len(write_data), 1000): temp_list.append(write_data[i:i+1000]) else: temp_list.append(write_data) for index, data_item in enumerate(temp_list): if sel_version == '2.0': res, _ = self.plc.write_d_130_ascii_batch(start_address=4001 + index * 500, values=data_item) if res: print("130 上料扫码NG2.0人成功") else: print("130 上料扫码NG2.0写人失败") d4038, d_err = self.plc.read_d(4038) if d_err: print(f"回读 D4038 失败: {d_err}") else: c1 = chr(d4038 & 0xFF) if 32 <= (d4038 & 0xFF) < 127 else '.' c2 = chr((d4038 >> 8) & 0xFF) if 32 <= ((d4038 >> 8) & 0xFF) < 127 else '.' print(f"回读 D4038: raw={d4038} hex={d4038:04X} ascii='{c1}{c2}'") elif sel_version == '2.5': res, _ = self.plc.write_d_130_ascii_batch(start_address=6000 + index * 500, values=data_item) if res: print("130 上料扫码NG2.5人成功") else: print("130 上料扫码NG2.5写人失败") # 数据置空 self.temp_data = '' def stop(self): """停止服务器""" self.is_running = False if self.server_socket: self.server_socket.close() def run_thread(self): thread = threading.Thread(target=self.run) thread.start() # 通用方法 def check_pressure_requirements(P_A, P_C, P_A_ww, P_C_ww): # 检查阳极压损比例要求 PA_ratios_h2 = P_A / P_A_ww PA_invalid_indices = (PA_ratios_h2 > 0.80) and (PA_ratios_h2 < 1.125) if PA_invalid_indices: PA_invalid_indices = 'OK' else: PA_invalid_indices = 'NG' # 检查阴极压损比例要求 PC_ratios_02 = P_C / P_C_ww PC_invalid_indices = (PC_ratios_02 > 0.80) and (PC_ratios_02 < 1.20) if PA_invalid_indices: PC_invalid_indices = 'OK' else: PC_invalid_indices = 'NG' return PA_ratios_h2, PC_ratios_02, PA_invalid_indices, PC_invalid_indices def parse_all_battery_data_130(data): """ 解析所有battery_data，将字符串格式转换为结构化字典 """ if data is None: return None # 创建数据的深拷贝 import copy parsed_data = copy.deepcopy(data) # 解析 battery_data_1 (57条数据) parsed_battery_data_1 = [] for item in data['box_data_1']['battery_data_1']: parsed_item = { 'serial_number': item[0:2], 'Up_QR': item[2:17], 'Below_QR': item[17:32], 'H2': int(item[32:36]) / 100, '02': int(item[36:40]) / 100, 'Cell_ran': item[40], 'thick': int(item[41:]) / 10000, 'source_code': item, } parsed_battery_data_1.append(parsed_item) # 解析 battery_data_2 (8条数据) parsed_battery_data_2 = [] for item in data['box_data_2']['battery_data_2']: parsed_item = { 'serial_number': item[0:2], 'Up_QR': item[2:17], 'Below_QR': item[17:32], 'H2': int(item[32:36]) / 100, '02': int(item[36:40]) / 100, 'Cell_ran': item[40], 'thick': int(item[41:]) / 10000, 'source_code': item, } parsed_battery_data_2.append(parsed_item) # 替换battery_data部分 parsed_data['box_data_1']['battery_data_1'] = parsed_battery_data_1 parsed_data['box_data_2']['battery_data_2'] = parsed_battery_data_2 return parsed_data def parse_all_battery_data_566(data): """ 解析所有battery_data，将字符串格式转换为结构化字典 """ if data is not None: # 创建数据的深拷贝 import copy parsed_data = copy.deepcopy(data) parsed_battery_data_3 = [] for item in data['cah_data']: parsed_item = { 'serial_number': item[0:2], 'Up_QR': item[2:17], 'Below_QR': item[17:32], 'H2': int(item[32:36]) / 100, '02': int(item[36:40]) / 100, 'Cell_ran': item[40], 'thick': item[41:], 'source_code': item, } parsed_battery_data_3.append(parsed_item) parsed_data['cah_data'] = parsed_battery_data_3 return parsed_data return None def parse_all_battery_data_566_2(data): """ 解析所有battery_data，将字符串格式转换为结构化字典 """ result = {'cah_data': []} for battery_list in data['cah_data']: if not battery_list: # 如果是空列表 result['cah_data'].append([]) continue parsed_battery_data = [] for item in battery_list: parsed_item = { 'serial_number': item[0:2], 'Up_QR': item[2:17], 'Below_QR': item[17:32], 'H2': int(item[32:36]) / 100, '02': int(item[36:40]) / 100, 'Cell_ran': item[40], 'thick': int(item[41:]) / 10000, 'source_code': item, } parsed_battery_data.append(parsed_item) result['cah_data'].append(parsed_battery_data) return result def split_json_objects(data_string): # 匹配以 { 开头，到 } 结束的完整对象 if not data_string: return [] brace_count = 0 current_object = '' json_objects = [] in_object = False for char in data_string: if char == '{': if not in_object: in_object = True brace_count += 1 if in_object: current_object += char if char == '}': brace_count -= 1 if brace_count == 0 and in_object: # 找到一个完整的JSON对象 json_objects.append(current_object) current_object = '' in_object = False try: json_objects = [json.loads(str_data) for str_data in json_objects] except Exception as e: print(e, '&&&') return [] return json_objects def start(page: ft.Page): """启动函数""" processor = PlcDataProcessor('192.168.2.1', 4995) processor.read_and_process_data(page) def main(): """启动函数""" processor = PlcDataProcessor('192.168.2.1', 4995) processor.read_main_data() # # if __name__ == "__main__": main()