Pyqt 和通用python,这可以被认为是一种正确的编码方法吗?
Pyqt and general python, can this be considered a correct approach for coding?
我有一个包含复选框的对话框 window,当每个复选框都被选中时,需要实例化一个特定的 class 并且 运行 一个单独线程上的任务(每个复选框一个)。我有 14 个复选框来检查 .isChecked()
属性 并且可以理解地检查每个复选框的 returned 布尔值效率不高并且需要更多编码。
因此我决定获取与复选框元素对应的所有子项,只获取那些被选中的,将它们的名称附加到列表中并循环遍历它们,将它们的名称与 d 字典匹配,其中键是复选框,值为对应的class进行实例化。
示例:
# class dictionary
self.summary_runnables = {'dupStreetCheckBox': [DupStreetDesc(),0],
'notStreetEsuCheckBox': [StreetsNoEsuDesc(),1],
'notType3CheckBox': [Type3Desc(False),2],
'incFootPathCheckBox': [Type3Desc(True),2],
'dupEsuRefCheckBox': [DupEsuRef(True),3],
'notEsuStreetCheckBox': [NoLinkEsuStreets(),4],
'invCrossRefCheckBox': [InvalidCrossReferences()],
'startEndCheckBox': [CheckStartEnd(tol=10),8],
'tinyEsuCheckBox': [CheckTinyEsus("esu",1)],
'notMaintReinsCheckBox': [CheckMaintReins()],
'asdStartEndCheckBox': [CheckAsdCoords()],
'notMaintPolysCheckBox': [MaintNoPoly(),16],
'notPolysMaintCheckBox': [PolyNoMaint()],
'tinyPolysCheckBox': [CheckTinyEsus("rd_poly",1)]}
# looping through list
self.long_task = QThreadPool(None).globalInstance()
self.long_task.setMaxThreadCount(1)
start_report = StartReport(val_file_path)
end_report = EndReport()
# start_report.setAutoDelete(False)
# end_report.setAutoDelete(False)
end_report.signals.result.connect(self.log_progress)
end_report.signals.finished.connect(self.show_finished)
# end_report.setAutoDelete(False)
start_report.signals.result.connect(self.log_progress)
self.long_task.start(start_report)
# print str(self.check_boxes_names)
for check_box_name in self.check_boxes_names:
run_class = self.summary_runnables[check_box_name]
if run_class[0].__class__.__name__ is 'CheckStartEnd':
run_class[0].tolerance = tolerance
runnable = run_class[0]()
runnable.signals.result.connect(self.log_progress)
self.long_task.start(runnable)
self.long_task.start(end_report)
运行可用的示例(即使其中一些使用不同的全局函数)
我不能 post 将内容写入文件的全局函数太多了,而且并非所有 14 个任务都执行相同类型的函数。这些函数的参数是其他字典的 int 键,这些字典包含报告静态内容和 SQL 对 return 报告主要动态内容的查询。
class StartReport(QRunnable):
def __init__(self, file_path):
super(StartReport,self).__init__()
# open the db connection in thread
db.open()
self.signals = GeneralSignals()
# self.simple_signal = SimpleSignal()
# print self.signals.result
self.file_path = file_path
self.task = "Starting Report"
self.progress = 1
self.org_name = org_name
self.user = user
self.report_title = "Validation Report"
print "instantiation of start report "
def run(self):
self.signals.result.emit(self.task, self.progress)
if self.file_path is None:
print "I started and found file none "
return
else:
global report_file
# create the file and prints the header
report_file = open(self.file_path, 'wb')
report_file.write(str(self.report_title) + ' for {0} \n'.format(self.org_name))
report_file.write('Created on : {0} at {1} By : {2} \n'.format(datetime.today().strftime("%d/%m/%Y"),
datetime.now().strftime("%H:%M"),
str(self.user)))
report_file.write(
"------------------------------------------------------------------------------------------ \n \n \n \n")
report_file.flush()
os.fsync(report_file.fileno())
class EndReport(QRunnable):
def __init__(self):
super(EndReport,self).__init__()
self.signals = GeneralSignals()
self.task = "Finishing report"
self.progress = 100
def run(self):
self.signals.result.emit(self.task, self.progress)
if report_file is not None:
# write footer and close file
report_file.write("\n \n \n")
report_file.write("---------- End of Report -----------")
report_file.flush()
os.fsync(report_file.fileno())
report_file.close()
self.signals.finished.emit()
# TODO: checking whether opening a db connection in thread might affect the db on the GUI
# if db.isOpen():
# db.close()
else:
return
class DupStreetDesc(QRunnable):
"""
duplicate street description report section creation
:return: void if the report is to text
list[string] if the report is to screen
"""
def __init__(self):
super(DupStreetDesc,self).__init__()
self.signals = GeneralSignals()
self.task = "Checking duplicate street descriptions..."
self.progress = 16.6
def run(self):
self.signals.result.emit(self.task,self.progress)
if report_file is None:
print "report file is none "
# items_list = write_content(0, 0, 0, 0)
# for item in items_list:
# self.signals.list.emit(item)
else:
write_content(0, 0, 0, 0)
现在,我以前使用过这种方法,并且在不使用多处理的情况下它始终运行良好。在这种情况下,它在某种程度上工作得很好,我可以第一次 运行 任务但是如果我第二次尝试 运行 我得到以下 Python 错误:
self.long_task.start(run_class[0])
RuntimeError: wrapped C/C++ object of type DupStreetDesc has been deleted
我尝试在循环中 运行 之前使用 run_class[0].setAutoDelete(False)
但是 pyQt 崩溃并出现小型转储错误(我正在 运行 宁 QGIS 中的代码)并且我的程序几乎没有机会了解发生了什么。
另一方面,如果我 运行 我的 class 是分开的,用 if else 语句检查每个复选框,那么它工作正常,我可以 运行任务和 C++ classes 没有被删除,但这不是一个好的编码方法,至少从我很少的经验来看是这样。
有没有其他人可以建议一种不同的方法,以便在不使用太多代码行的情况下顺利地完成 运行?或者知道是否有更有效的模式来处理这个问题,我认为这一定很常见?
看来你应该为每个runnable创建一个新的实例,并让Qt自动删除它。所以您的字典条目可能如下所示:
'dupStreetCheckBox': [lambda: DupStreetDesc(), 0],
然后你可以做:
for check_box_name in self.check_boxes_names:
run_class = self.summary_runnables[check_box_name]
runnable = run_class[0]()
runnable.signals.result.connect(self.log_progress)
self.long_task.start(runnable)
我不知道为什么 setAutoDelete
不起作用(假设您在 在 启动线程池之前调用它)。我想可能有一个错误,但如果没有一个完整的工作示例来测试是不可能确定的。
我有一个包含复选框的对话框 window,当每个复选框都被选中时,需要实例化一个特定的 class 并且 运行 一个单独线程上的任务(每个复选框一个)。我有 14 个复选框来检查 .isChecked()
属性 并且可以理解地检查每个复选框的 returned 布尔值效率不高并且需要更多编码。
因此我决定获取与复选框元素对应的所有子项,只获取那些被选中的,将它们的名称附加到列表中并循环遍历它们,将它们的名称与 d 字典匹配,其中键是复选框,值为对应的class进行实例化。
示例:
# class dictionary
self.summary_runnables = {'dupStreetCheckBox': [DupStreetDesc(),0],
'notStreetEsuCheckBox': [StreetsNoEsuDesc(),1],
'notType3CheckBox': [Type3Desc(False),2],
'incFootPathCheckBox': [Type3Desc(True),2],
'dupEsuRefCheckBox': [DupEsuRef(True),3],
'notEsuStreetCheckBox': [NoLinkEsuStreets(),4],
'invCrossRefCheckBox': [InvalidCrossReferences()],
'startEndCheckBox': [CheckStartEnd(tol=10),8],
'tinyEsuCheckBox': [CheckTinyEsus("esu",1)],
'notMaintReinsCheckBox': [CheckMaintReins()],
'asdStartEndCheckBox': [CheckAsdCoords()],
'notMaintPolysCheckBox': [MaintNoPoly(),16],
'notPolysMaintCheckBox': [PolyNoMaint()],
'tinyPolysCheckBox': [CheckTinyEsus("rd_poly",1)]}
# looping through list
self.long_task = QThreadPool(None).globalInstance()
self.long_task.setMaxThreadCount(1)
start_report = StartReport(val_file_path)
end_report = EndReport()
# start_report.setAutoDelete(False)
# end_report.setAutoDelete(False)
end_report.signals.result.connect(self.log_progress)
end_report.signals.finished.connect(self.show_finished)
# end_report.setAutoDelete(False)
start_report.signals.result.connect(self.log_progress)
self.long_task.start(start_report)
# print str(self.check_boxes_names)
for check_box_name in self.check_boxes_names:
run_class = self.summary_runnables[check_box_name]
if run_class[0].__class__.__name__ is 'CheckStartEnd':
run_class[0].tolerance = tolerance
runnable = run_class[0]()
runnable.signals.result.connect(self.log_progress)
self.long_task.start(runnable)
self.long_task.start(end_report)
运行可用的示例(即使其中一些使用不同的全局函数)
我不能 post 将内容写入文件的全局函数太多了,而且并非所有 14 个任务都执行相同类型的函数。这些函数的参数是其他字典的 int 键,这些字典包含报告静态内容和 SQL 对 return 报告主要动态内容的查询。
class StartReport(QRunnable):
def __init__(self, file_path):
super(StartReport,self).__init__()
# open the db connection in thread
db.open()
self.signals = GeneralSignals()
# self.simple_signal = SimpleSignal()
# print self.signals.result
self.file_path = file_path
self.task = "Starting Report"
self.progress = 1
self.org_name = org_name
self.user = user
self.report_title = "Validation Report"
print "instantiation of start report "
def run(self):
self.signals.result.emit(self.task, self.progress)
if self.file_path is None:
print "I started and found file none "
return
else:
global report_file
# create the file and prints the header
report_file = open(self.file_path, 'wb')
report_file.write(str(self.report_title) + ' for {0} \n'.format(self.org_name))
report_file.write('Created on : {0} at {1} By : {2} \n'.format(datetime.today().strftime("%d/%m/%Y"),
datetime.now().strftime("%H:%M"),
str(self.user)))
report_file.write(
"------------------------------------------------------------------------------------------ \n \n \n \n")
report_file.flush()
os.fsync(report_file.fileno())
class EndReport(QRunnable):
def __init__(self):
super(EndReport,self).__init__()
self.signals = GeneralSignals()
self.task = "Finishing report"
self.progress = 100
def run(self):
self.signals.result.emit(self.task, self.progress)
if report_file is not None:
# write footer and close file
report_file.write("\n \n \n")
report_file.write("---------- End of Report -----------")
report_file.flush()
os.fsync(report_file.fileno())
report_file.close()
self.signals.finished.emit()
# TODO: checking whether opening a db connection in thread might affect the db on the GUI
# if db.isOpen():
# db.close()
else:
return
class DupStreetDesc(QRunnable):
"""
duplicate street description report section creation
:return: void if the report is to text
list[string] if the report is to screen
"""
def __init__(self):
super(DupStreetDesc,self).__init__()
self.signals = GeneralSignals()
self.task = "Checking duplicate street descriptions..."
self.progress = 16.6
def run(self):
self.signals.result.emit(self.task,self.progress)
if report_file is None:
print "report file is none "
# items_list = write_content(0, 0, 0, 0)
# for item in items_list:
# self.signals.list.emit(item)
else:
write_content(0, 0, 0, 0)
现在,我以前使用过这种方法,并且在不使用多处理的情况下它始终运行良好。在这种情况下,它在某种程度上工作得很好,我可以第一次 运行 任务但是如果我第二次尝试 运行 我得到以下 Python 错误:
self.long_task.start(run_class[0])
RuntimeError: wrapped C/C++ object of type DupStreetDesc has been deleted
我尝试在循环中 运行 之前使用 run_class[0].setAutoDelete(False)
但是 pyQt 崩溃并出现小型转储错误(我正在 运行 宁 QGIS 中的代码)并且我的程序几乎没有机会了解发生了什么。
另一方面,如果我 运行 我的 class 是分开的,用 if else 语句检查每个复选框,那么它工作正常,我可以 运行任务和 C++ classes 没有被删除,但这不是一个好的编码方法,至少从我很少的经验来看是这样。
有没有其他人可以建议一种不同的方法,以便在不使用太多代码行的情况下顺利地完成 运行?或者知道是否有更有效的模式来处理这个问题,我认为这一定很常见?
看来你应该为每个runnable创建一个新的实例,并让Qt自动删除它。所以您的字典条目可能如下所示:
'dupStreetCheckBox': [lambda: DupStreetDesc(), 0],
然后你可以做:
for check_box_name in self.check_boxes_names:
run_class = self.summary_runnables[check_box_name]
runnable = run_class[0]()
runnable.signals.result.connect(self.log_progress)
self.long_task.start(runnable)
我不知道为什么 setAutoDelete
不起作用(假设您在 在 启动线程池之前调用它)。我想可能有一个错误,但如果没有一个完整的工作示例来测试是不可能确定的。