138 lines
5.7 KiB
Python
138 lines
5.7 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
|
||
"""
|
||
监控模型，处理试验安排与监控相关的数据
|
||
"""
|
||
|
||
from PyQt5.QtCore import QObject, pyqtProperty, pyqtSignal, QVariant, pyqtSlot
|
||
import sqlite3
|
||
import os
|
||
import json
|
||
import ast
|
||
|
||
|
||
class MonitorModel(QObject):
|
||
def __init__(self, parent=None):
|
||
super().__init__(parent)
|
||
self._monitor_data = []
|
||
self._filtered_data = [] # 仅包含dutDirectionMode==2的数据
|
||
self._load_data()
|
||
|
||
# 信号定义
|
||
dataChanged = pyqtSignal()
|
||
|
||
def _load_data(self):
|
||
"""从数据库加载数据"""
|
||
try:
|
||
# 连接数据库
|
||
db_path = os.path.join(os.path.dirname(os.path.abspath(__file__)), 'dtmgtDb.db')
|
||
if os.path.exists(db_path):
|
||
conn = sqlite3.connect(db_path)
|
||
cursor = conn.cursor()
|
||
|
||
# 查询 dutList 表中的试验数据
|
||
cursor.execute("""
|
||
SELECT SN, name, project, projectType, phase, weeks, stationAssigned,
|
||
itemOnGoing, itemsFinished, status, testReq, inspector, id,
|
||
description, deadLine, createdate, workOrder, direction_codes
|
||
FROM dutList
|
||
""")
|
||
|
||
rows = cursor.fetchall()
|
||
self._monitor_data = []
|
||
|
||
for row in rows:
|
||
# 解析direction_codes字段获取dutDirectionMode
|
||
direction_codes = row[17] if row[17] else ""
|
||
dut_direction_mode = 0
|
||
|
||
# 解析类似"[33, 35, 36]"的字符串
|
||
if direction_codes and direction_codes.startswith('[') and direction_codes.endswith(']'):
|
||
try:
|
||
# 使用ast.literal_eval安全地解析列表
|
||
codes_list = ast.literal_eval(direction_codes)
|
||
if isinstance(codes_list, list) and len(codes_list) > 0:
|
||
dut_direction_mode = codes_list[0] # 取第一个元素作为dutDirectionMode
|
||
except:
|
||
# 如果解析失败，尝试其他方式
|
||
try:
|
||
# 移除方括号并分割
|
||
codes_str = direction_codes.strip('[]')
|
||
codes_list = [int(code.strip()) for code in codes_str.split(',') if code.strip().isdigit()]
|
||
if len(codes_list) > 0:
|
||
dut_direction_mode = codes_list[0]
|
||
except:
|
||
dut_direction_mode = 0
|
||
elif direction_codes.isdigit():
|
||
# 如果是纯数字
|
||
dut_direction_mode = int(direction_codes)
|
||
|
||
monitor = {
|
||
"id": row[12] if row[12] is not None else 0,
|
||
"name": row[1] if row[1] else "",
|
||
"status": row[9] if row[9] else "",
|
||
"dutDirectionMode": dut_direction_mode,
|
||
"project": row[2] if row[2] else "",
|
||
"SN": row[0] if row[0] else "",
|
||
"description": row[13] if row[13] else ""
|
||
}
|
||
self._monitor_data.append(monitor)
|
||
|
||
conn.close()
|
||
else:
|
||
# 如果数据库不存在，使用示例数据
|
||
self._monitor_data = [
|
||
{"id": 1, "name": "测试1", "dutDirectionMode": 2, "status": "运行中", "project": "项目A"},
|
||
{"id": 2, "name": "测试2", "dutDirectionMode": 1, "status": "待开始", "project": "项目B"},
|
||
{"id": 3, "name": "测试3", "dutDirectionMode": 2, "status": "已完成", "project": "项目C"}
|
||
]
|
||
|
||
# 过滤数据，只保留dutDirectionMode==2的记录
|
||
self._filtered_data = [item for item in self._monitor_data if item.get("dutDirectionMode") == 2]
|
||
self.dataChanged.emit()
|
||
|
||
except Exception as e:
|
||
print(f"加载监控数据时出错: {e}")
|
||
import traceback
|
||
traceback.print_exc()
|
||
# 出错时使用默认数据
|
||
self._monitor_data = []
|
||
self._filtered_data = []
|
||
|
||
@pyqtProperty('QVariant', notify=dataChanged)
|
||
def monitorData(self):
|
||
"""返回所有监控数据"""
|
||
return self._monitor_data
|
||
|
||
@pyqtProperty('QVariant', notify=dataChanged)
|
||
def filteredMonitorData(self):
|
||
"""返回过滤后的监控数据（仅dutDirectionMode==2）"""
|
||
return self._filtered_data
|
||
|
||
@pyqtSlot(int, result='QVariant')
|
||
def getMonitorById(self, monitor_id):
|
||
"""根据ID获取监控信息"""
|
||
for monitor in self._monitor_data:
|
||
if monitor.get("id") == monitor_id:
|
||
return monitor
|
||
return None
|
||
|
||
@pyqtSlot(result=int)
|
||
def getMonitorCount(self):
|
||
"""获取监控总数"""
|
||
return len(self._monitor_data)
|
||
|
||
@pyqtSlot(result=int)
|
||
def getFilteredMonitorCount(self):
|
||
"""获取过滤后的监控总数"""
|
||
return len(self._filtered_data)
|
||
|
||
def get_filtered_monitors(self):
|
||
"""获取过滤后的监控列表（供视图直接调用）"""
|
||
return self._filtered_data
|
||
|
||
def get_all_monitors(self):
|
||
"""获取所有监控列表（供视图直接调用）"""
|
||
return self._monitor_data
|