[bos #38044][EDF] (2023-T3) Support for automatic reparation. Python dump.

This commit is contained in:
Konstantin Leontev 2024-03-06 12:00:30 +00:00 committed by DUC ANH HOANG
parent 692be895e1
commit 127038b5d7
6 changed files with 119 additions and 60 deletions

View File

@ -21,20 +21,17 @@
import sys
from pathlib import Path
from traceback import format_exc
from qtsalome import Qt, QWidget, QMessageBox, QApplication, QGridLayout
from salome.gui import helper
from salome.kernel.studyedit import EDITOR
from salome.kernel.services import IDToObject, ObjectToID
from salome.kernel.services import IDToObject
from salome.geom import geomBuilder
from salome.geom.geomtools import GeomStudyTools
from libGEOM_Swig import GEOM_Swig
from .basedlg_ui import Ui_BaseDlg
from .geomrepairadv_execute import execute
from .geomrepairadv_logger import logger
from .geomrepairadv_common import DlgRef_1Sel_QTD, \
GEOM_RESULT_NAME_GRP, NAME_LBL, GEOM_SELECTED_LBL, GEOM_SELECTED_SHAPE
import GEOM
@ -99,7 +96,6 @@ class BaseDlg(Ui_BaseDlg, QWidget):
# that we need to pass for execution instead of original one.
# TODO: decide if we really need to pass a copy.
self._selected_object = None
self._selected_copy = None
# Put the common widgets and a child widget for a specific algorithm
# in a place right above standard buttons (defined by child_placeholder).
@ -187,25 +183,11 @@ class BaseDlg(Ui_BaseDlg, QWidget):
)
return
# Make copy to prevent unintentional changing of a source object from the algo script
builder = geomBuilder.New()
self._selected_copy = builder.MakeCopy(
self._selected_object, self.get_result_name() + '_temp')
args_dict = self.get_args()
if args_dict:
# Add the copy object first
args_dict['source_solid'] = self._selected_copy
execute(self._algo_name, args_dict)
execute(self._selected_object, self._algo_name, args_dict)
# TODO: do we need to handle here a case if the algo failed?
# Delete a copy object in any case
copy_entry = ObjectToID(self._selected_copy)
tools = GeomStudyTools()
tools.deleteShape(copy_entry)
self._selected_copy = None
def set_algoname(self, algo_name, is_default_location):
"""

View File

@ -23,11 +23,14 @@ import os
import sys
import importlib.util
from qtsalome import QApplication, QFileDialog
from salome.kernel.services import ObjectToID
from salome.geom.geomtools import GeomStudyTools
from .geomrepairadv_progress import RepairProgressDialog
from .geomrepairadv_logger import logger
from qtsalome import Qt, QApplication, QFileDialog
# Testing
import salome
@ -70,16 +73,17 @@ def module_from_filename(filename):
return module
def execute(algo_name, args_dict):
def execute(selected_object, algo_name, args_dict):
"""
Executes GEOM advanced repair algorithm.
Args:
selected_object - geom object selected by user for algorithm
algo_name - path to the algo module
args_dict - dictionary with arguments those are specific for each algo.
Returns:
False if the algo failed.
Result GEOM object or None if failed or canceled.
"""
logger.debug('execute() start')
@ -88,12 +92,49 @@ def execute(algo_name, args_dict):
algo_module = module_from_filename(algo_name)
logger.debug('algo_module: %s', algo_module)
if not algo_module:
return False
return None
# Keep the args for python dump
args_dict_str = str(args_dict)
logger.debug('args_dict_str: {}'.format(args_dict_str))
# Make copy to prevent unintentional changing of a source object from the algo script
geompy = geomBuilder.New()
selected_copy = geompy.MakeCopy(
selected_object, args_dict['result_name'] + '_temp')
# Add the copy object as a source
args_dict['source_solid'] = selected_copy
logger.debug('Create RepairProgressDialog...')
progress_dlg = RepairProgressDialog(parent=None, target=algo_module.run, args=args_dict)
result = progress_dlg.exec()
logger.info('result: %s', result)
progress_dlg.exec()
# Delete a copy object in any case
copy_entry = ObjectToID(selected_copy)
tools = GeomStudyTools()
tools.deleteShape(copy_entry)
# Python dump if execution was completed without errors
if progress_dlg.is_completed():
result_object = progress_dlg.get_result()
# Completed execution doesn't guarantee that we received a valid object
if not result_object:
logger.error('Could not get a result object after exec of %s file!', str(algo_name))
return None
geompy.FuncToPythonDump(
selected_object,
result_object,
'from salome.geom.geomrepairadv import geomrepairadv_execute\n',
'geomrepairadv_execute.execute',
'\'' + str(algo_name) + '\', ' + args_dict_str
)
return result_object
return None
def test_execution():
@ -118,6 +159,7 @@ def test_execution():
# Récupération des faces à fusionner
face_a = geompy.GetFaceNearPoint(source_solid, geompy.MakeVertex(-143, -127, 250))
face_b = geompy.GetFaceNearPoint(source_solid, geompy.MakeVertex(49,-127,250))
faces_ids = geompy.GetSubShapesIDs(source_solid, [face_a, face_b])
geompy.addToStudy(source_solid, "source_solid")
geompy.addToStudyInFather(source_solid, face_a, "face_a")
@ -126,8 +168,7 @@ def test_execution():
args_dict = {
'source_solid': source_solid,
'face_a': face_a,
'face_b': face_b,
'faces_ids': faces_ids,
'result_name': 'MergeFaces_result'
}
@ -137,7 +178,7 @@ def test_execution():
if not algo_filename:
return
execute(algo_filename, args_dict)
execute(source_solid, algo_filename, args_dict)
if __name__ == '__main__':

View File

@ -65,6 +65,9 @@ class RepairProgressDialog(QDialog, QPlainTextEdit):
# Helper flag to decide if we need to change button or close the dialog
self.canceled = False
# Helper flag to check if execution was completed without errors
self.completed = False
# Set logger to redirect logs output into the text widget
self.log_handler = QTextEditLogger(self)
logging.getLogger().addHandler(self.log_handler)
@ -124,8 +127,12 @@ class RepairProgressDialog(QDialog, QPlainTextEdit):
self.progress.setLabelText('Completed!')
self.progress.setCancelButtonText('Close')
# Set the Close button to actually close dialog
self.canceled = True
# Lets us know that we get the job done
self.completed = True
def value(self):
"""
@ -156,6 +163,22 @@ class RepairProgressDialog(QDialog, QPlainTextEdit):
super().close()
def is_completed(self):
"""
Returns True if execution was completed without errors and wasn't canceled.
"""
return self.completed
def get_result(self):
"""
Returns result of the execution or None if the execution failed.
"""
return self.thread.get_result()
def test_thread():
"""
Tests running a test function in a thread while
@ -166,8 +189,12 @@ def test_thread():
"""
progress_dlg = RepairProgressDialog(parent=None, target=test, args=None)
result = progress_dlg.exec()
logging.info('result: %s', result)
progress_dlg.exec()
if progress_dlg.is_completed():
logging.info('result: %s', progress_dlg.get_result())
else:
logging.info('Cannot get results because execution was not completed.')
def test(args, progress_emitter):
@ -182,6 +209,7 @@ def test(args, progress_emitter):
logging.debug('debug msg')
sleep(2)
# raise Exception
progress_emitter.emit()
logging.info('info msg')
@ -200,6 +228,8 @@ def test(args, progress_emitter):
progress_emitter.emit()
return "Result from test!"
if __name__ == '__main__':
app = QApplication(sys.argv)

View File

@ -19,7 +19,6 @@
#
# Author : Konstantin Leontev (OpenCascade S.A.S)
import logging
import inspect
from traceback import format_exc
@ -56,6 +55,9 @@ class Worker(QThread):
# Set a progress emitter to update the progress from the target
self.progress_emitter = ProgressEmitter(self.progress_update, total_lines, first_line)
# Set a variable for result
self.result = None
def run(self):
"""
@ -66,7 +68,7 @@ class Worker(QThread):
# Wait mode cursor
QApplication.setOverrideCursor(Qt.WaitCursor)
self.target(self.args, self.progress_emitter)
self.result = self.target(self.args, self.progress_emitter)
# Reset the progress when finished
self.progress_update.emit(100)
@ -91,6 +93,14 @@ class Worker(QThread):
QApplication.restoreOverrideCursor()
def get_result(self):
"""
Returns result of the execution or None if the execution failed.
"""
return self.result
class ProgressEmitter():
"""
Helper class to reduce code repetition while update progress

View File

@ -24,7 +24,6 @@ import sys
from qtsalome import QGridLayout, QFrame, QMessageBox, QApplication
from libGEOM_Swig import GEOM_Swig
from salome.geom import geomBuilder
from .geomrepairadv_logger import logger
from .basedlg import BaseDlg
from .geomrepairadv_common import DlgRef_1Spin_QTD
@ -87,14 +86,8 @@ class MergeFacesDlg(BaseDlg):
)
return None
# Get faces from a temporary copy object
builder = geomBuilder.New()
faces = builder.SubShapes(self._selected_copy, faces_ids)
logger.debug('faces: %s', faces)
return {
'face_a': faces[0],
'face_b': faces[1],
'faces_ids': faces_ids,
'result_name': self.get_result_name(),
'precision': self.get_precision()
}

View File

@ -44,7 +44,7 @@ def run(args_dict, progress_emitter):
args_dict - arguments as pairs string : any type value
Returns:
A string with result description.
A result object.
"""
logging.info('Run Merge Faces algorithm.')
@ -52,36 +52,39 @@ def run(args_dict, progress_emitter):
if ('source_solid' not in args_dict or
'face_a' not in args_dict or
'face_b' not in args_dict or
'faces_ids' not in args_dict or
'result_name' not in args_dict):
logging.info('Cant execute an algo because the arguments are empty!')
return False
source_solid = args_dict['source_solid']
face_a = args_dict['face_a']
face_b = args_dict['face_b']
faces_ids = args_dict['faces_ids']
result_name = args_dict['result_name']
logging.info('Creating of two faces...')
progress_emitter.emit()
# Fusion des deux faces
partition = geompy.MakePartition([face_a, face_b],[])
points = [geompy.GetVertexNearPoint(partition, geompy.MakeVertex(-298, 29, 250)),
geompy.GetVertexNearPoint(partition, geompy.MakeVertex(178, 29, 250)),
geompy.GetVertexNearPoint(partition, geompy.MakeVertex(178, -282, 250)),
geompy.GetVertexNearPoint(partition, geompy.MakeVertex(-298, -282, 250))]
wire = geompy.MakePolyline(points,True)
fused_face = geompy.MakeFaceWires([wire], True)
geompy.addToStudy(fused_face, "fused_face")
# This block creates a face using passed selected faces.
# Commented to simplify output - just one object.
# # Fusion des deux faces
# faces = geompy.SubShapes(source_solid, faces_ids)
# logging.info('faces: %s', faces)
# partition = geompy.MakePartition([faces[0], faces[1]],[])
# points = [geompy.GetVertexNearPoint(partition, geompy.MakeVertex(-298, 29, 250)),
# geompy.GetVertexNearPoint(partition, geompy.MakeVertex(178, 29, 250)),
# geompy.GetVertexNearPoint(partition, geompy.MakeVertex(178, -282, 250)),
# geompy.GetVertexNearPoint(partition, geompy.MakeVertex(-298, -282, 250))]
# wire = geompy.MakePolyline(points,True)
# fused_face = geompy.MakeFaceWires([wire], True)
# geompy.addToStudy(fused_face, "fused_face")
logging.info('Creating of a new geometry from the source brep...')
progress_emitter.emit()
sleep(5)
sleep(1)
# Fusion des deux faces au sein de la boite + nettoyage de la boite
points = [geompy.GetVertexNearPoint(source_solid, geompy.MakeVertex(-298, 29, 250)),
@ -99,7 +102,7 @@ def run(args_dict, progress_emitter):
logging.info('Cleaning of the new geometry...')
progress_emitter.emit()
sleep(5)
sleep(1)
# Uncomment to simulate exception handling in a thread worker class
# raise Exception
@ -120,7 +123,7 @@ def run(args_dict, progress_emitter):
logging.info('Creating a solid...')
progress_emitter.emit()
sleep(5)
sleep(1)
# ### Création du solide
shell = geompy.MakeShell(faces)
@ -131,7 +134,7 @@ def run(args_dict, progress_emitter):
logging.info('Merge Faces algorithm was completed successfully.')
progress_emitter.emit()
return True
return solid
def test():