1190 lines
52 KiB
Python
1190 lines
52 KiB
Python
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()
|