3475ae00cd
- final fix for app not quiting when running a script as argument, script that has the quit_flatcam Tcl command; fixed issue #360 - fixed issue #363. The Tcl command drillcncjob does not create tool cut, does not allow creation of gcode, it forces the usage of dwell and dwelltime parameters
429 lines
15 KiB
Python
429 lines
15 KiB
Python
import sys
|
|
import re
|
|
import FlatCAMApp
|
|
import abc
|
|
import collections
|
|
from PyQt5 import QtCore, QtGui
|
|
from PyQt5.QtCore import Qt
|
|
from contextlib import contextmanager
|
|
|
|
|
|
class TclCommand(object):
|
|
|
|
# FlatCAMApp
|
|
app = None
|
|
|
|
# Logger
|
|
log = None
|
|
|
|
# List of all command aliases, to be able use old names
|
|
# for backward compatibility (add_poly, add_polygon)
|
|
aliases = []
|
|
|
|
# Dictionary of types from Tcl command, needs to be ordered
|
|
# OrderedDict should be like collections.OrderedDict([(key,value),(key2,value2)])
|
|
arg_names = collections.OrderedDict([
|
|
('name', str)
|
|
])
|
|
|
|
# dictionary of types from Tcl command, needs to be ordered.
|
|
# This is for options like -optionname value.
|
|
# OrderedDict should be like collections.OrderedDict([(key,value),(key2,value2)])
|
|
option_types = collections.OrderedDict()
|
|
|
|
# List of mandatory options for current Tcl command: required = {'name','outname'}
|
|
required = ['name']
|
|
|
|
# Structured help for current command, args needs to be ordered
|
|
# OrderedDict should be like collections.OrderedDict([(key,value),(key2,value2)])
|
|
help = {
|
|
'main': "undefined help.",
|
|
'args': collections.OrderedDict([
|
|
('argumentname', 'undefined help.'),
|
|
('optionname', 'undefined help.')
|
|
]),
|
|
'examples': []
|
|
}
|
|
|
|
# Original incoming arguments into command
|
|
original_args = None
|
|
|
|
def __init__(self, app):
|
|
self.app = app
|
|
|
|
if self.app is None:
|
|
raise TypeError('Expected app to be FlatCAMApp instance.')
|
|
|
|
if not isinstance(self.app, FlatCAMApp.App):
|
|
raise TypeError('Expected FlatCAMApp, got %s.' % type(app))
|
|
|
|
self.log = self.app.log
|
|
|
|
def raise_tcl_error(self, text):
|
|
"""
|
|
This method pass exception from python into TCL as error
|
|
so we get stacktrace and reason.
|
|
|
|
This is only redirect to self.app.raise_tcl_error
|
|
|
|
:param text: text of error
|
|
:return: none
|
|
"""
|
|
|
|
self.app.raise_tcl_error(text)
|
|
|
|
def get_current_command(self):
|
|
"""
|
|
Get current command, we are not able to get it from TCL we have to reconstruct it.
|
|
|
|
:return: current command
|
|
"""
|
|
command_string = list()
|
|
command_string.append(self.aliases[0])
|
|
|
|
if self.original_args is not None:
|
|
for arg in self.original_args:
|
|
command_string.append(arg)
|
|
|
|
return " ".join(command_string)
|
|
|
|
def get_decorated_help(self):
|
|
"""
|
|
Decorate help for TCL console output.
|
|
|
|
:return: decorated help from structure
|
|
"""
|
|
|
|
def get_decorated_command(alias_name):
|
|
|
|
command_string = []
|
|
|
|
for arg_key, arg_type in list(self.help['args'].items()):
|
|
command_string.append(get_decorated_argument(arg_key, arg_type, True))
|
|
|
|
return "> " + alias_name + " " + " ".join(command_string)
|
|
|
|
def get_decorated_argument(help_key, help_text, in_command=False):
|
|
"""
|
|
|
|
:param help_key: Name of the argument.
|
|
:param help_text:
|
|
:param in_command:
|
|
:return:
|
|
"""
|
|
option_symbol = ''
|
|
|
|
if help_key in self.arg_names:
|
|
arg_type = self.arg_names[help_key]
|
|
type_name = str(arg_type.__name__)
|
|
# in_command_name = help_key + "<" + type_name + ">"
|
|
in_command_name = help_key
|
|
|
|
elif help_key in self.option_types:
|
|
option_symbol = '-'
|
|
arg_type = self.option_types[help_key]
|
|
type_name = str(arg_type.__name__)
|
|
in_command_name = option_symbol + help_key + " <" + type_name + ">"
|
|
|
|
else:
|
|
option_symbol = ''
|
|
type_name = '?'
|
|
in_command_name = option_symbol + help_key + " <" + type_name + ">"
|
|
|
|
if in_command:
|
|
if help_key in self.required:
|
|
return in_command_name
|
|
else:
|
|
return '[' + in_command_name + "]"
|
|
else:
|
|
if help_key in self.required:
|
|
return "\t" + option_symbol + help_key + " <" + type_name + ">: " + help_text
|
|
else:
|
|
return "\t[" + option_symbol + help_key + " <" + type_name + ">: " + help_text + "]"
|
|
|
|
def get_decorated_example(example_item):
|
|
return "> " + example_item
|
|
|
|
help_string = [self.help['main']]
|
|
for alias in self.aliases:
|
|
help_string.append(get_decorated_command(alias))
|
|
|
|
for key, value in list(self.help['args'].items()):
|
|
help_string.append(get_decorated_argument(key, value))
|
|
|
|
# timeout is unique for signaled commands (this is not best oop practice, but much easier for now)
|
|
if isinstance(self, TclCommandSignaled):
|
|
help_string.append("\t[-timeout <int>: Max wait for job timeout before error.]")
|
|
|
|
for example in self.help['examples']:
|
|
help_string.append(get_decorated_example(example))
|
|
|
|
return "\n".join(help_string)
|
|
|
|
@staticmethod
|
|
def parse_arguments(args):
|
|
"""
|
|
Pre-processes arguments to detect '-keyword value' pairs into dictionary
|
|
and standalone parameters into list.
|
|
|
|
This is copy from FlatCAMApp.setup_shell().h() just for accessibility,
|
|
original should be removed after all commands will be converted
|
|
|
|
:param args: arguments from tcl to parse
|
|
:return: arguments, options
|
|
"""
|
|
|
|
options = {}
|
|
arguments = []
|
|
n = len(args)
|
|
|
|
name = None
|
|
for i in range(n):
|
|
match = re.search(r'^-([a-zA-Z].*)', args[i])
|
|
if match:
|
|
assert name is None
|
|
name = match.group(1)
|
|
continue
|
|
|
|
if name is None:
|
|
arguments.append(args[i])
|
|
else:
|
|
options[name] = args[i]
|
|
name = None
|
|
|
|
return arguments, options
|
|
|
|
def check_args(self, args):
|
|
"""
|
|
Check arguments and options for right types
|
|
|
|
:param args: arguments from tcl to check
|
|
:return: named_args, unnamed_args
|
|
"""
|
|
|
|
arguments, options = self.parse_arguments(args)
|
|
named_args = {}
|
|
unnamed_args = []
|
|
|
|
# check arguments
|
|
idx = 0
|
|
arg_names_items = list(self.arg_names.items())
|
|
for argument in arguments:
|
|
if len(self.arg_names) > idx:
|
|
key, arg_type = arg_names_items[idx]
|
|
try:
|
|
named_args[key] = arg_type(argument)
|
|
except Exception as e:
|
|
self.raise_tcl_error("Cannot cast named argument '%s' to type %s with exception '%s'."
|
|
% (key, arg_type, str(e)))
|
|
else:
|
|
unnamed_args.append(argument)
|
|
idx += 1
|
|
|
|
# check options
|
|
for key in options:
|
|
if key not in self.option_types and key != 'timeout':
|
|
self.raise_tcl_error('Unknown parameter: %s' % key)
|
|
try:
|
|
if key != 'timeout':
|
|
named_args[key] = self.option_types[key](options[key])
|
|
else:
|
|
named_args[key] = int(options[key])
|
|
except Exception as e:
|
|
self.raise_tcl_error("Cannot cast argument '-%s' to type '%s' with exception '%s'."
|
|
% (key, self.option_types[key], str(e)))
|
|
|
|
# check required arguments
|
|
for key in self.required:
|
|
if key not in named_args:
|
|
self.raise_tcl_error("Missing required argument '%s'." % key)
|
|
|
|
return named_args, unnamed_args
|
|
|
|
def raise_tcl_unknown_error(self, unknown_exception):
|
|
"""
|
|
raise Exception if is different type than TclErrorException
|
|
this is here mainly to show unknown errors inside TCL shell console
|
|
|
|
:param unknown_exception:
|
|
:return:
|
|
"""
|
|
|
|
raise unknown_exception
|
|
|
|
def raise_tcl_error(self, text):
|
|
"""
|
|
this method pass exception from python into TCL as error, so we get stacktrace and reason
|
|
:param text: text of error
|
|
:return: raise exception
|
|
"""
|
|
|
|
# because of signaling we cannot call error to TCL from here but when task
|
|
# is finished also non-signaled are handled here to better exception
|
|
# handling and displayed after command is finished
|
|
raise self.app.TclErrorException(text)
|
|
|
|
def execute_wrapper(self, *args):
|
|
"""
|
|
Command which is called by tcl console when current commands aliases are hit.
|
|
Main catch(except) is implemented here.
|
|
This method should be reimplemented only when initial checking sequence differs
|
|
|
|
:param args: arguments passed from tcl command console
|
|
:return: None, output text or exception
|
|
"""
|
|
|
|
# self.worker_task.emit({'fcn': self.exec_command_test, 'params': [text, False]})
|
|
try:
|
|
self.log.debug("TCL command '%s' executed." % str(self.__class__))
|
|
self.original_args = args
|
|
args, unnamed_args = self.check_args(args)
|
|
return self.execute(args, unnamed_args)
|
|
except Exception as unknown:
|
|
error_info = sys.exc_info()
|
|
self.log.error("TCL command '%s' failed. Error text: %s" % (str(self), str(unknown)))
|
|
self.app.display_tcl_error(unknown, error_info)
|
|
self.raise_tcl_unknown_error(unknown)
|
|
|
|
@abc.abstractmethod
|
|
def execute(self, args, unnamed_args):
|
|
"""
|
|
Direct execute of command, this method should be implemented in each descendant.
|
|
No main catch should be implemented here.
|
|
|
|
:param args: array of known named arguments and options
|
|
:param unnamed_args: array of other values which were passed into command
|
|
without -somename and we do not have them in known arg_names
|
|
:return: None, output text or exception
|
|
"""
|
|
|
|
raise NotImplementedError("Please Implement this method")
|
|
|
|
|
|
class TclCommandSignaled(TclCommand):
|
|
"""
|
|
!!! I left it here only for demonstration !!!
|
|
Go to TclCommandCncjob and into class definition put
|
|
class TclCommandCncjob(TclCommandSignaled):
|
|
also change
|
|
obj.generatecncjob(use_thread = False, **args)
|
|
to
|
|
obj.generatecncjob(use_thread = True, **args)
|
|
|
|
|
|
This class is child of TclCommand and is used for commands which create new objects
|
|
it handles all necessary stuff about blocking and passing exceptions
|
|
"""
|
|
|
|
@abc.abstractmethod
|
|
def execute(self, args, unnamed_args):
|
|
raise NotImplementedError("Please Implement this method")
|
|
|
|
output = None
|
|
|
|
def execute_call(self, args, unnamed_args):
|
|
|
|
try:
|
|
self.output = None
|
|
self.error = None
|
|
self.error_info = None
|
|
self.output = self.execute(args, unnamed_args)
|
|
except Exception as unknown:
|
|
self.error_info = sys.exc_info()
|
|
self.error = unknown
|
|
finally:
|
|
self.app.shell_command_finished.emit(self)
|
|
|
|
def execute_wrapper(self, *args):
|
|
"""
|
|
Command which is called by tcl console when current commands aliases are hit.
|
|
Main catch(except) is implemented here.
|
|
This method should be reimplemented only when initial checking sequence differs
|
|
|
|
:param args: arguments passed from tcl command console
|
|
:return: None, output text or exception
|
|
"""
|
|
|
|
@contextmanager
|
|
def wait_signal(signal, timeout=300000):
|
|
"""Block loop until signal emitted, or timeout (ms) elapses."""
|
|
loop = QtCore.QEventLoop()
|
|
|
|
# Normal termination
|
|
signal.connect(loop.quit)
|
|
|
|
# Termination by exception in thread
|
|
self.app.thread_exception.connect(loop.quit)
|
|
|
|
status = {'timed_out': False}
|
|
|
|
def report_quit():
|
|
status['timed_out'] = True
|
|
loop.quit()
|
|
|
|
yield
|
|
|
|
# Temporarily change how exceptions are managed.
|
|
oeh = sys.excepthook
|
|
ex = []
|
|
|
|
def except_hook(type_, value, traceback_):
|
|
ex.append(value)
|
|
oeh(type_, value, traceback_)
|
|
sys.excepthook = except_hook
|
|
|
|
# Terminate on timeout
|
|
if timeout is not None:
|
|
QtCore.QTimer.singleShot(timeout, report_quit)
|
|
|
|
# Block
|
|
loop.exec_()
|
|
|
|
# Restore exception management
|
|
sys.excepthook = oeh
|
|
if ex:
|
|
raise ex[0]
|
|
|
|
if status['timed_out']:
|
|
self.app.raise_tcl_unknown_error("Operation timed outed! Consider increasing option "
|
|
"'-timeout <miliseconds>' for command or "
|
|
"'set_sys global_background_timeout <miliseconds>'.")
|
|
|
|
try:
|
|
self.log.debug("TCL command '%s' executed." % str(self.__class__))
|
|
self.original_args = args
|
|
args, unnamed_args = self.check_args(args)
|
|
if 'timeout' in args:
|
|
passed_timeout = args['timeout']
|
|
del args['timeout']
|
|
else:
|
|
passed_timeout = self.app.defaults['global_background_timeout']
|
|
|
|
# set detail for processing, it will be there until next open or close
|
|
self.app.shell.open_processing(self.get_current_command())
|
|
|
|
def handle_finished(obj):
|
|
self.app.shell_command_finished.disconnect(handle_finished)
|
|
if self.error is not None:
|
|
self.raise_tcl_unknown_error(self.error)
|
|
|
|
self.app.shell_command_finished.connect(handle_finished)
|
|
|
|
with wait_signal(self.app.shell_command_finished, passed_timeout):
|
|
# every TclCommandNewObject ancestor support timeout as parameter,
|
|
# but it does not mean anything for child itself
|
|
# when operation will be really long is good to set it higher then defqault 30s
|
|
self.app.worker_task.emit({'fcn': self.execute_call, 'params': [args, unnamed_args]})
|
|
|
|
return self.output
|
|
|
|
except Exception as unknown:
|
|
# if error happens inside thread execution, then pass correct error_info to display
|
|
if self.error_info is not None:
|
|
error_info = self.error_info
|
|
else:
|
|
error_info = sys.exc_info()
|
|
self.log.error("TCL command '%s' failed." % str(self))
|
|
self.app.display_tcl_error(unknown, error_info)
|
|
self.raise_tcl_unknown_error(unknown)
|