443 lines
15 KiB
Python
443 lines
15 KiB
Python
import sys
|
|
import re
|
|
import app_Main
|
|
import abc
|
|
import collections
|
|
from PyQt5 import QtCore
|
|
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, app_Main.App):
|
|
raise TypeError('Expected FlatCAMApp, got %s.' % type(app))
|
|
|
|
self.log = self.app.log
|
|
self.error_info = None
|
|
self.error = None
|
|
|
|
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.shell.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 = [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)
|
|
|
|
option_name = None
|
|
|
|
for i in range(n):
|
|
match = re.search(r'^-([a-zA-Z].*)', args[i])
|
|
if match:
|
|
# assert option_name is None
|
|
if option_name is not None:
|
|
options[option_name] = None
|
|
|
|
option_name = match.group(1)
|
|
continue
|
|
|
|
if option_name is None:
|
|
arguments.append(args[i])
|
|
else:
|
|
options[option_name] = args[i]
|
|
option_name = None
|
|
|
|
if option_name is not None:
|
|
options[option_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':
|
|
# None options are allowed; if None then the defaults are used
|
|
# - must be implemented in the Tcl commands
|
|
if options[key] is not None:
|
|
named_args[key] = self.option_types[key](options[key])
|
|
else:
|
|
named_args[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.shell.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(type(self).__name__))
|
|
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.shell.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:
|
|
time_val = int(timeout)
|
|
QtCore.QTimer.singleShot(time_val, report_quit)
|
|
|
|
# Block
|
|
loop.exec_()
|
|
|
|
# Restore exception management
|
|
sys.excepthook = oeh
|
|
if ex:
|
|
raise ex[0]
|
|
|
|
if status['timed_out']:
|
|
self.app.shell.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(type(self).__name__))
|
|
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():
|
|
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.shell.display_tcl_error(unknown, error_info)
|
|
self.raise_tcl_unknown_error(unknown)
|