b98954dccd
more detaild error print including python trace when more complex unknown error
reinplement drillcncjob
fix camlib problem with all drills("all" was already there) but it crashes on tools without points, when no tools "all" is as default
add timeout to all helps if command is signaled
397 lines
14 KiB
Python
397 lines
14 KiB
Python
import sys
|
|
import re
|
|
import FlatCAMApp
|
|
import abc
|
|
import collections
|
|
from PyQt4 import QtCore
|
|
from contextlib import contextmanager
|
|
|
|
|
|
class TclCommand(object):
|
|
|
|
# FlatCAMApp
|
|
app = None
|
|
|
|
# logger
|
|
log = None
|
|
|
|
# array 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()
|
|
|
|
# array 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 = []
|
|
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 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):
|
|
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 = "<" + type_name + ">"
|
|
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 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 = 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, 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, 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, unknownException):
|
|
"""
|
|
raise Exception if is different type than TclErrorException
|
|
this is here mainly to show unknown errors inside TCL shell console
|
|
:param unknownException:
|
|
:return:
|
|
"""
|
|
|
|
#if not isinstance(unknownException, self.TclErrorException):
|
|
# self.raise_tcl_error("Unknown error: %s" % str(unknownException))
|
|
#else:
|
|
raise unknownException
|
|
|
|
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
|
|
"""
|
|
|
|
# becouse of signaling we cannot call error to TCL from here but when task is finished
|
|
# also nonsiglaned arwe handled here to better exception handling and diplay 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." % str(self))
|
|
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(TclCommand.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 neccessary stuff about blocking and passing exeptions
|
|
"""
|
|
|
|
# default timeout for operation is 10 sec, but it can be much more
|
|
default_timeout = 10000
|
|
|
|
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=10000):
|
|
"""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:
|
|
self.raise_tcl_error(str(ex[0]))
|
|
|
|
if status['timed_out']:
|
|
self.app.raise_tcl_unknown_error('Operation timed out!')
|
|
|
|
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.default_timeout
|
|
|
|
# set detail for processing, it will be there until next open or close
|
|
self.app.shell.open_proccessing(self.get_current_command())
|
|
|
|
def handle_finished(obj):
|
|
self.app.shell_command_finished.disconnect(handle_finished)
|
|
if self.error is not None:
|
|
self.log.error("TCL command '%s' failed." % str(self))
|
|
self.app.display_tcl_error(self.error, self.error_info)
|
|
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:
|
|
self.log.error("TCL command '%s' failed." % str(self))
|
|
self.raise_tcl_unknown_error(unknown) |