wanghanbo
/
battery
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

first commit

This commit is contained in:
wanghanbo 2026-09-14 08:46:02 +08:00
parent aeb5016d74
commit 4590abfddd
5 changed files with 73 additions and 58 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
andon_flask_client.py
View File

@ -30,7 +30,7 @@ class AndonEndpoint(Enum):
class SimpleAndonSender:
"""简化版 Andon 发送器"""
def __init__(self, server_ip: str = "192.168.70.35", port: int = 5100, timeout: int = 5):
def __init__(self, server_ip: str = "192.168.2.1", port: int = 5100, timeout: int = 5):
self.base_url = f"http://{server_ip}:{port}"
self.timeout = timeout
@ -169,7 +169,6 @@ class SimpleAndonSender:
return None
if __name__ == '__main__':
"""主函数 - 监听键盘输入并调用自定义方法"""
andon_sender = SimpleAndonSender(server_ip='127.0.0.1', port=5100)

20
dao/dbutils.py
View File

@ -12,7 +12,7 @@ logger = logging.getLogger(__name__)
class BatteryDatabase:
def __init__(self, db_path=r"D:\Comproject\battery\battery_data.db"):
def __init__(self, db_path="battery_data.db"):
self.db_path = db_path
# self.init_database()
@ -500,12 +500,12 @@ class BatteryDatabase:
logger.error(f"通过二维码精确查询组信息失败: {e}")
return []
# # 使用示例和测试
def main():
# 创建数据库实例
db = BatteryDatabase("../battery_data.db")
full_start_time = "2026-08-12 15:41:38.093294"
full_end_time = "2026-08-12 15:41:38.093294"
batches = db.get_batches_by_time_range_with_daily_sort(full_start_time,full_end_time)
print(batches)
if __name__ == "__main__":
main()
# def main():
# # 创建数据库实例
# db = BatteryDatabase("../battery_data.db")
# full_start_time = "2025-11-06 11:38:51"
# full_end_time = "2025-11-06 11:38:51"
# batches = db.get_batches_by_time_range_with_daily_sort(full_start_time,full_end_time)
# print(batches)
# if __name__ == "__main__":
# main()

3
main.py
View File

@ -35,6 +35,8 @@ def main(page: ft.Page):
cross_axis_margin=2,
)
)
threading.Thread(target=start, args=(page,), daemon=True).start()
# 设置窗口大小和禁止调整
page.window.width = 1000
page.window.height = 700
@ -935,7 +937,6 @@ def main(page: ft.Page):
])
page.add(background_layout)
threading.Thread(target=start, args=(page,), daemon=True).start()
s.page_initialized = True # 页面构建完成，设置初始化标志
# 确保这是唯一的 ft.app 调用

29
plc_client.py
View File

@ -1,6 +1,5 @@
import socket
import struct
import threading
import time
from typing import Optional, List, Tuple, Dict
@ -15,7 +14,6 @@ class PlcClient:
self.timeout = PLC_TIMEOUT
self.socket = None
self.connected = False
self._lock = threading.Lock()
self._connect_with_retry()
def _connect_with_retry(self) -> bool:
@ -54,21 +52,20 @@ class PlcClient:
def _send_and_receive(self, cmd_hex: str, expected_response_length: int = 1024) -> Tuple[
Optional[bytes], Optional[str]]:
"""发送命令并接收响应，处理连接问题"""
with self._lock:
if not self._ensure_connection():
return None, "PLC连接不可用"
if not self._ensure_connection():
return None, "PLC连接不可用"
try:
self.socket.sendall(self._hex_str_to_bytes(cmd_hex))
buffer = self.socket.recv(expected_response_length)
return buffer, None
except socket.timeout:
return None, "PLC响应超时"
except socket.error as ex:
self.connected = False
return None, f"通信错误: {str(ex)}"
except Exception as ex:
return None, f"未知错误: {str(ex)}"
try:
self.socket.sendall(self._hex_str_to_bytes(cmd_hex))
buffer = self.socket.recv(expected_response_length)
return buffer, None
except socket.timeout:
return None, "PLC响应超时"
except socket.error as ex:
self.connected = False
return None, f"通信错误: {str(ex)}"
except Exception as ex:
return None, f"未知错误: {str(ex)}"
def read_m(self, address: int) -> Tuple[Optional[bool], Optional[str]]:
try:

74
sevice/airtight_service.py
View File

@ -17,7 +17,6 @@ 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()
@ -42,7 +41,6 @@ class PlcDataProcessor:
self.temp_data = ''
self.sel_version = '2.0'
self.run_thread()
# def read_and_process_data(self, page: ft.Page):
# """主循环函数，读取并处理PLC数据"""
# while True:
@ -94,21 +92,14 @@ class PlcDataProcessor:
self._process_cycle()
time.sleep(3)
def _heartbeat_1003(self):
"""每秒把 M1003 写为 False"""
while True:
try:
self.plc.write_m(1003, False)
except Exception as e:
print(f"M1003 写入失败: {e}")
time.sleep(1)
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
@ -361,7 +352,7 @@ class PlcDataProcessor:
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)
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(
@ -430,7 +421,8 @@ class PlcDataProcessor:
self.shen_7_data_write(results_566["低压电池"])
self.plc.write_m(1024, True)
if m_values[9]:
data_566_2 = self._process_zr_566_manage_data("2.5")
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']]
@ -908,9 +900,8 @@ class PlcDataProcessor:
# 2.0丛D4001开始 到 D5999
# 2.5从D6000开始 到 D7999
self.write_130_batch(self.temp_data, self.sel_version)
print("--------")
print(self.temp_data)
print("--------")
send_content = "recv ok"
# 对字符串进行编码成为二进制数据
response = send_content.encode("utf-8")
@ -963,25 +954,56 @@ class PlcDataProcessor:
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] == 0:
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])
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)
# print(res)
# print(_)
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:
@ -1000,7 +1022,6 @@ class PlcDataProcessor:
def run_thread(self):
thread = threading.Thread(target=self.run)
thread.start()
threading.Thread(target=self._heartbeat_1003).start()
# 通用方法
@ -1158,13 +1179,10 @@ 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__":