diff options
Diffstat (limited to 'klippy/gcode.py')
| -rw-r--r-- | klippy/gcode.py | 263 |
1 files changed, 179 insertions, 84 deletions
diff --git a/klippy/gcode.py b/klippy/gcode.py index 975da792..427d7992 100644 --- a/klippy/gcode.py +++ b/klippy/gcode.py @@ -5,13 +5,17 @@ # This file may be distributed under the terms of the GNU GPLv3 license. import os, re, logging, collections, shlex + class CommandError(Exception): pass -Coord = collections.namedtuple('Coord', ('x', 'y', 'z', 'e')) + +Coord = collections.namedtuple("Coord", ("x", "y", "z", "e")) + class GCodeCommand: error = CommandError + def __init__(self, gcode, command, commandline, params, need_ack): self._command = command self._commandline = commandline @@ -20,12 +24,16 @@ class GCodeCommand: # Method wrappers self.respond_info = gcode.respond_info self.respond_raw = gcode.respond_raw + def get_command(self): return self._command + def get_commandline(self): return self._commandline + def get_command_parameters(self): return self._params + def get_raw_command_parameters(self): command = self._command origline = self._commandline @@ -34,12 +42,13 @@ class GCodeCommand: if origline[:param_start].upper() != command: # Skip any gcode line-number and ignore any trailing checksum param_start += origline.upper().find(command) - end = origline.rfind('*') - if end >= 0 and origline[end+1:].isdigit(): + end = origline.rfind("*") + if end >= 0 and origline[end + 1 :].isdigit(): param_end = end - if origline[param_start:param_start+1].isspace(): + if origline[param_start : param_start + 1].isspace(): param_start += 1 return origline[param_start:param_end] + def ack(self, msg=None): if not self._need_ack: return False @@ -49,52 +58,82 @@ class GCodeCommand: self.respond_raw(ok_msg) self._need_ack = False return True + # Parameter parsing helpers - class sentinel: pass - def get(self, name, default=sentinel, parser=str, minval=None, maxval=None, - above=None, below=None): + class sentinel: + pass + + def get( + self, + name, + default=sentinel, + parser=str, + minval=None, + maxval=None, + above=None, + below=None, + ): value = self._params.get(name) if value is None: if default is self.sentinel: - raise self.error("Error on '%s': missing %s" - % (self._commandline, name)) + raise self.error( + "Error on '%s': missing %s" % (self._commandline, name) + ) return default try: value = parser(value) except: - raise self.error("Error on '%s': unable to parse %s" - % (self._commandline, value)) + raise self.error( + "Error on '%s': unable to parse %s" % (self._commandline, value) + ) if minval is not None and value < minval: - raise self.error("Error on '%s': %s must have minimum of %s" - % (self._commandline, name, minval)) + raise self.error( + "Error on '%s': %s must have minimum of %s" + % (self._commandline, name, minval) + ) if maxval is not None and value > maxval: - raise self.error("Error on '%s': %s must have maximum of %s" - % (self._commandline, name, maxval)) + raise self.error( + "Error on '%s': %s must have maximum of %s" + % (self._commandline, name, maxval) + ) if above is not None and value <= above: - raise self.error("Error on '%s': %s must be above %s" - % (self._commandline, name, above)) + raise self.error( + "Error on '%s': %s must be above %s" % (self._commandline, name, above) + ) if below is not None and value >= below: - raise self.error("Error on '%s': %s must be below %s" - % (self._commandline, name, below)) + raise self.error( + "Error on '%s': %s must be below %s" % (self._commandline, name, below) + ) return value + def get_int(self, name, default=sentinel, minval=None, maxval=None): return self.get(name, default, parser=int, minval=minval, maxval=maxval) - def get_float(self, name, default=sentinel, minval=None, maxval=None, - above=None, below=None): - return self.get(name, default, parser=float, minval=minval, - maxval=maxval, above=above, below=below) + + def get_float( + self, name, default=sentinel, minval=None, maxval=None, above=None, below=None + ): + return self.get( + name, + default, + parser=float, + minval=minval, + maxval=maxval, + above=above, + below=below, + ) + # Parse and dispatch G-Code commands class GCodeDispatch: error = CommandError Coord = Coord + def __init__(self, printer): self.printer = printer self.is_fileinput = not not printer.get_start_args().get("debuginput") printer.register_event_handler("klippy:ready", self._handle_ready) printer.register_event_handler("klippy:shutdown", self._handle_shutdown) - printer.register_event_handler("klippy:disconnect", - self._handle_disconnect) + printer.register_event_handler("klippy:disconnect", self._handle_disconnect) # Command handling self.is_printer_ready = False self.mutex = printer.get_reactor().mutex() @@ -105,12 +144,21 @@ class GCodeDispatch: self.gcode_help = {} self.status_commands = {} # Register commands needed before config file is loaded - handlers = ['M110', 'M112', 'M115', - 'RESTART', 'FIRMWARE_RESTART', 'ECHO', 'STATUS', 'HELP'] + handlers = [ + "M110", + "M112", + "M115", + "RESTART", + "FIRMWARE_RESTART", + "ECHO", + "STATUS", + "HELP", + ] for cmd in handlers: - func = getattr(self, 'cmd_' + cmd) - desc = getattr(self, 'cmd_' + cmd + '_help', None) + func = getattr(self, "cmd_" + cmd) + desc = getattr(self, "cmd_" + cmd + "_help", None) self.register_command(cmd, func, True, desc) + def is_traditional_gcode(self, cmd): # A "traditional" g-code command is a letter and followed by a number try: @@ -119,6 +167,7 @@ class GCodeDispatch: return cmd[0].isupper() and cmd[1].isdigit() except: return False + def register_command(self, cmd, func, when_not_ready=False, desc=None): if func is None: old_cmd = self.ready_gcode_handlers.get(cmd) @@ -130,12 +179,18 @@ class GCodeDispatch: return old_cmd if cmd in self.ready_gcode_handlers: raise self.printer.config_error( - "gcode command %s already registered" % (cmd,)) + "gcode command %s already registered" % (cmd,) + ) if not self.is_traditional_gcode(cmd): - if (cmd.upper() != cmd or not cmd.replace('_', 'A').isalnum() - or cmd[0].isdigit() or cmd[1:2].isdigit()): + if ( + cmd.upper() != cmd + or not cmd.replace("_", "A").isalnum() + or cmd[0].isdigit() + or cmd[1:2].isdigit() + ): raise self.printer.config_error( - "Can't register '%s' as it is an invalid name" % (cmd,)) + "Can't register '%s' as it is an invalid name" % (cmd,) + ) origfunc = func func = lambda params: origfunc(self._get_extended_params(params)) self.ready_gcode_handlers[cmd] = func @@ -144,6 +199,7 @@ class GCodeDispatch: if desc is not None: self.gcode_help[cmd] = desc self._build_status_commands() + def register_mux_command(self, cmd, key, value, func, desc=None): prev = self.mux_commands.get(cmd) if prev is None: @@ -153,25 +209,32 @@ class GCodeDispatch: prev_key, prev_values = prev if prev_key != key: raise self.printer.config_error( - "mux command %s %s %s may have only one key (%s)" % ( - cmd, key, value, prev_key)) + "mux command %s %s %s may have only one key (%s)" + % (cmd, key, value, prev_key) + ) if value in prev_values: raise self.printer.config_error( - "mux command %s %s %s already registered (%s)" % ( - cmd, key, value, prev_values)) + "mux command %s %s %s already registered (%s)" + % (cmd, key, value, prev_values) + ) prev_values[value] = func + def get_command_help(self): return dict(self.gcode_help) + def get_status(self, eventtime): - return {'commands': self.status_commands} + return {"commands": self.status_commands} + def _build_status_commands(self): commands = {cmd: {} for cmd in self.gcode_handlers} for cmd in self.gcode_help: if cmd in commands: - commands[cmd]['help'] = self.gcode_help[cmd] + commands[cmd]["help"] = self.gcode_help[cmd] self.status_commands = commands + def register_output_handler(self, cb): self.output_callbacks.append(cb) + def _handle_shutdown(self): if not self.is_printer_ready: return @@ -179,32 +242,35 @@ class GCodeDispatch: self.gcode_handlers = self.base_gcode_handlers self._build_status_commands() self._respond_state("Shutdown") + def _handle_disconnect(self): self._respond_state("Disconnect") + def _handle_ready(self): self.is_printer_ready = True self.gcode_handlers = self.ready_gcode_handlers self._build_status_commands() self._respond_state("Ready") + # Parse input into commands - args_r = re.compile('([A-Z_]+|[A-Z*])') + args_r = re.compile("([A-Z_]+|[A-Z*])") + def _process_commands(self, commands, need_ack=True): for line in commands: # Ignore comments and leading/trailing spaces line = origline = line.strip() - cpos = line.find(';') + cpos = line.find(";") if cpos >= 0: line = line[:cpos] # Break line into parts and determine command parts = self.args_r.split(line.upper()) - if ''.join(parts[:2]) == 'N': + if "".join(parts[:2]) == "N": # Skip line number at start of command - cmd = ''.join(parts[3:5]).strip() + cmd = "".join(parts[3:5]).strip() else: - cmd = ''.join(parts[:3]).strip() + cmd = "".join(parts[:3]).strip() # Build gcode "params" dictionary - params = { parts[i]: parts[i+1].strip() - for i in range(1, len(parts), 2) } + params = {parts[i]: parts[i + 1].strip() for i in range(1, len(parts), 2)} gcmd = GCodeCommand(self, cmd, origline, params, need_ack) # Invoke handler for command handler = self.gcode_handlers.get(cmd, self.cmd_default) @@ -223,59 +289,68 @@ class GCodeDispatch: if not need_ack: raise gcmd.ack() + def run_script_from_command(self, script): - self._process_commands(script.split('\n'), need_ack=False) + self._process_commands(script.split("\n"), need_ack=False) + def run_script(self, script): with self.mutex: - self._process_commands(script.split('\n'), need_ack=False) + self._process_commands(script.split("\n"), need_ack=False) + def get_mutex(self): return self.mutex + def create_gcode_command(self, command, commandline, params): return GCodeCommand(self, command, commandline, params, False) + # Response handling def respond_raw(self, msg): for cb in self.output_callbacks: cb(msg) + def respond_info(self, msg, log=True): if log: logging.info(msg) - lines = [l.strip() for l in msg.strip().split('\n')] + lines = [l.strip() for l in msg.strip().split("\n")] self.respond_raw("// " + "\n// ".join(lines)) + def _respond_error(self, msg): logging.warning(msg) - lines = msg.strip().split('\n') + lines = msg.strip().split("\n") if len(lines) > 1: self.respond_info("\n".join(lines), log=False) - self.respond_raw('!! %s' % (lines[0].strip(),)) + self.respond_raw("!! %s" % (lines[0].strip(),)) if self.is_fileinput: - self.printer.request_exit('error_exit') + self.printer.request_exit("error_exit") + def _respond_state(self, state): self.respond_info("Klipper state: %s" % (state,), log=False) + # Parameter parsing helpers def _get_extended_params(self, gcmd): rawparams = gcmd.get_raw_command_parameters() # Extract args while allowing shell style quoting s = shlex.shlex(rawparams, posix=True) s.whitespace_split = True - s.commenters = '#;' + s.commenters = "#;" try: - eparams = [earg.split('=', 1) for earg in s] - eparams = { k.upper(): v for k, v in eparams } + eparams = [earg.split("=", 1) for earg in s] + eparams = {k.upper(): v for k, v in eparams} except ValueError as e: - raise self.error("Malformed command '%s'" - % (gcmd.get_commandline(),)) + raise self.error("Malformed command '%s'" % (gcmd.get_commandline(),)) # Update gcmd with new parameters gcmd._params.clear() gcmd._params.update(eparams) return gcmd + # G-Code special command handlers def cmd_default(self, gcmd): cmd = gcmd.get_command() - if cmd == 'M105': + if cmd == "M105": # Don't warn about temperature requests when not ready gcmd.ack("T:0") return - if cmd == 'M21': + if cmd == "M21": # Don't warn about sd card init when not ready return if not self.is_printer_ready: @@ -286,7 +361,7 @@ class GCodeDispatch: if cmdline: logging.debug(cmdline) return - if ' ' in cmd: + if " " in cmd: # Handle M117/M118 gcode with numeric and special characters realcmd = cmd.split()[0] if realcmd in ["M117", "M118", "M23"]: @@ -295,14 +370,16 @@ class GCodeDispatch: gcmd._command = realcmd handler(gcmd) return - elif cmd in ['M140', 'M104'] and not gcmd.get_float('S', 0.): + elif cmd in ["M140", "M104"] and not gcmd.get_float("S", 0.0): # Don't warn about requests to turn off heaters when not present return - elif cmd == 'M107' or (cmd == 'M106' and ( - not gcmd.get_float('S', 1.) or self.is_fileinput)): + elif cmd == "M107" or ( + cmd == "M106" and (not gcmd.get_float("S", 1.0) or self.is_fileinput) + ): # Don't warn about requests to turn off fan when fan not present return gcmd.respond_info('Unknown command:"%s"' % (cmd,)) + def _cmd_mux(self, command, gcmd): key, values = self.mux_commands[command] if None in values: @@ -310,43 +387,53 @@ class GCodeDispatch: else: key_param = gcmd.get(key) if key_param not in values: - raise gcmd.error("The value '%s' is not valid for %s" - % (key_param, key)) + raise gcmd.error("The value '%s' is not valid for %s" % (key_param, key)) values[key_param](gcmd) + # Low-level G-Code commands that are needed before the config file is loaded def cmd_M110(self, gcmd): # Set Current Line Number pass + def cmd_M112(self, gcmd): # Emergency Stop self.printer.invoke_shutdown("Shutdown due to M112 command") + def cmd_M115(self, gcmd): # Get Firmware Version and Capabilities - software_version = self.printer.get_start_args().get('software_version') + software_version = self.printer.get_start_args().get("software_version") kw = {"FIRMWARE_NAME": "Klipper", "FIRMWARE_VERSION": software_version} msg = " ".join(["%s:%s" % (k, v) for k, v in kw.items()]) did_ack = gcmd.ack(msg) if not did_ack: gcmd.respond_info(msg) + def request_restart(self, result): if self.is_printer_ready: - toolhead = self.printer.lookup_object('toolhead') + toolhead = self.printer.lookup_object("toolhead") print_time = toolhead.get_last_move_time() - if result == 'exit': + if result == "exit": logging.info("Exiting (print time %.3fs)" % (print_time,)) self.printer.send_event("gcode:request_restart", print_time) toolhead.dwell(0.500) toolhead.wait_moves() self.printer.request_exit(result) + cmd_RESTART_help = "Reload config file and restart host software" + def cmd_RESTART(self, gcmd): - self.request_restart('restart') + self.request_restart("restart") + cmd_FIRMWARE_RESTART_help = "Restart firmware, host, and reload config" + def cmd_FIRMWARE_RESTART(self, gcmd): - self.request_restart('firmware_restart') + self.request_restart("firmware_restart") + def cmd_ECHO(self, gcmd): gcmd.respond_info(gcmd.get_commandline(), log=False) + cmd_STATUS_help = "Report the printer status" + def cmd_STATUS(self, gcmd): if self.is_printer_ready: self._respond_state("Ready") @@ -354,7 +441,9 @@ class GCodeDispatch: msg = self.printer.get_state_message()[0] msg = msg.rstrip() + "\nKlipper state: Not ready" raise gcmd.error(msg) + cmd_HELP_help = "Report the list of available extended G-Code commands" + def cmd_HELP(self, gcmd): cmdhelp = [] if not self.is_printer_ready: @@ -365,13 +454,14 @@ class GCodeDispatch: cmdhelp.append("%-10s: %s" % (cmd, self.gcode_help[cmd])) gcmd.respond_info("\n".join(cmdhelp), log=False) + # Support reading gcode from a pseudo-tty interface class GCodeIO: def __init__(self, printer): self.printer = printer printer.register_event_handler("klippy:ready", self._handle_ready) printer.register_event_handler("klippy:shutdown", self._handle_shutdown) - self.gcode = printer.lookup_object('gcode') + self.gcode = printer.lookup_object("gcode") self.gcode_mutex = self.gcode.get_mutex() self.fd = printer.get_start_args().get("gcode_fd") self.reactor = printer.get_reactor() @@ -382,31 +472,34 @@ class GCodeIO: self.fd_handle = None if not self.is_fileinput: self.gcode.register_output_handler(self._respond_raw) - self.fd_handle = self.reactor.register_fd(self.fd, - self._process_data) + self.fd_handle = self.reactor.register_fd(self.fd, self._process_data) self.partial_input = "" self.pending_commands = [] self.bytes_read = 0 self.input_log = collections.deque([], 50) + def _handle_ready(self): self.is_printer_ready = True if self.is_fileinput and self.fd_handle is None: - self.fd_handle = self.reactor.register_fd(self.fd, - self._process_data) + self.fd_handle = self.reactor.register_fd(self.fd, self._process_data) + def _dump_debug(self): out = [] out.append("Dumping gcode input %d blocks" % (len(self.input_log),)) for eventtime, data in self.input_log: out.append("Read %f: %s" % (eventtime, repr(data))) logging.info("\n".join(out)) + def _handle_shutdown(self): if not self.is_printer_ready: return self.is_printer_ready = False self._dump_debug() if self.is_fileinput: - self.printer.request_exit('error_exit') - m112_r = re.compile(r'^(?:[nN][0-9]+)?\s*[mM]112(?:\s|$)') + self.printer.request_exit("error_exit") + + m112_r = re.compile(r"^(?:[nN][0-9]+)?\s*[mM]112(?:\s|$)") + def _process_data(self, eventtime): # Read input, separate by newline, and add to pending_commands try: @@ -416,7 +509,7 @@ class GCodeIO: return self.input_log.append((eventtime, data)) self.bytes_read += len(data) - lines = data.split('\n') + lines = data.split("\n") lines[0] = self.partial_input + lines[0] self.partial_input = lines.pop() pending_commands = self.pending_commands @@ -427,7 +520,7 @@ class GCodeIO: if not self.is_processing_data: self.reactor.unregister_fd(self.fd_handle) self.fd_handle = None - self.gcode.request_restart('exit') + self.gcode.request_restart("exit") pending_commands.append("") # Handle case where multiple commands pending if self.is_processing_data or len(pending_commands) > 1: @@ -451,18 +544,20 @@ class GCodeIO: pending_commands = self.pending_commands self.is_processing_data = False if self.fd_handle is None: - self.fd_handle = self.reactor.register_fd(self.fd, - self._process_data) + self.fd_handle = self.reactor.register_fd(self.fd, self._process_data) + def _respond_raw(self, msg): if self.pipe_is_active: try: - os.write(self.fd, (msg+"\n").encode()) + os.write(self.fd, (msg + "\n").encode()) except os.error: logging.exception("Write g-code response") self.pipe_is_active = False + def stats(self, eventtime): return False, "gcodein=%d" % (self.bytes_read,) + def add_early_printer_objects(printer): - printer.add_object('gcode', GCodeDispatch(printer)) - printer.add_object('gcode_io', GCodeIO(printer)) + printer.add_object("gcode", GCodeDispatch(printer)) + printer.add_object("gcode_io", GCodeIO(printer)) |