Basic Information:
Printer Model: Proforge 4
MCU / Printerboard: BTT Pi + BTT Octopus MAX EZ
gcode.py
i’m working on a project modifying a proforge 4 tool changing printer into a five axis edition with additional two rotational axis, i basically plan to replace 2 of 4 z axis into a & b rotational axis. What im currently understand is i need to bring some changes on following scripts:
a). gcode.py → allow system understand gcode of additional axis G X Y Z A B F.
b). kinematics [corexy] → add supports of additional axis a & b
c). printer.cfg → replace z and z1 axis into a and b
if there is anything requires any changes, im quite struggling on how the klipper files interact with each others. And i now have done the changes on gcode.py and corexy.py, but i currently cannot confirm if im working on right thing (no errors poped up for now). If u have any suggestions or reccomendations on whatever my works or any dicumentations worth for self learning please lend me your hands, great thanks to all your kindness. 
Great thanks to @ marianocarpentier the project has provided me a lot insight for my works
changes.zip (7.0 KB)
# Code for handling the kinematics of corexy robots
#
# Copyright (C) 2017-2021 Kevin O'Connor <kevin@koconnor.net>
#
# This file may be distributed under the terms of the GNU GPLv3 license.
import logging, math
import stepper
class CoreXYKinematics:
def __init__(self, toolhead, config):
# Setup axis rails
self.rails = [stepper.LookupMultiRail(config.getsection('stepper_' + n))
for n in 'xyz']
# Setup rotors for A and B
self.rotors = [stepper.LookupMultiRail(config.getsection('stepper_' + n))
for n in 'ab']
for s in self.rails[1].get_steppers():
self.rails[0].get_endstops()[0][0].add_stepper(s)
for s in self.rails[0].get_steppers():
self.rails[1].get_endstops()[0][0].add_stepper(s)
self.rails[0].setup_itersolve('corexy_stepper_alloc', b'+')
self.rails[1].setup_itersolve('corexy_stepper_alloc', b'-')
self.rails[2].setup_itersolve('cartesian_stepper_alloc', b'z')
self.rotors[0].setup_itersolve('rotab_stepper_alloc', b'a')
self.rotors[1].setup_itersolve('rotab_stepper_alloc', b'b')
for s in self.get_steppers():
s.set_trapq(toolhead.get_trapq())
toolhead.register_step_generator(s.generate_steps)
# Setup boundary checks
max_velocity, max_accel = toolhead.get_max_velocity()
self.max_z_velocity = config.getfloat(
'max_z_velocity', max_velocity, above=0., maxval=max_velocity)
self.max_z_accel = config.getfloat(
'max_z_accel', max_accel, above=0., maxval=max_accel)
self.limits = [(1.0, -1.0)] * 3
ranges = [r.get_range() for r in self.rails]
self.axes_min = toolhead.Coord(*[r[0] for r in ranges], e=0.)
self.axes_max = toolhead.Coord(*[r[1] for r in ranges], e=0.)
def get_steppers(self):
return [s for rail in self.rails for s in rail.get_steppers()]
def calc_position(self, stepper_positions):
pos = [stepper_positions[rail.get_name()] for rail in self.rails]
return [0.5 * (pos[0] + pos[1]), 0.5 * (pos[0] - pos[1]), pos[2]]
def set_position(self, newpos, homing_axes):
for i, rail in enumerate(self.rails):
rail.set_position(newpos)
if "xyz"[i] in homing_axes:
self.limits[i] = rail.get_range()
def clear_homing_state(self, clear_axes):
for axis, axis_name in enumerate("xyz"):
if axis_name in clear_axes:
self.limits[axis] = (1.0, -1.0)
def home(self, homing_state):
# Each axis is homed independently and in order
for axis in homing_state.get_axes():
rail = self.rails[axis]
# Determine movement
position_min, position_max = rail.get_range()
hi = rail.get_homing_info()
homepos = [None, None, None, None]
homepos[axis] = hi.position_endstop
forcepos = list(homepos)
if hi.positive_dir:
forcepos[axis] -= 1.5 * (hi.position_endstop - position_min)
else:
forcepos[axis] += 1.5 * (position_max - hi.position_endstop)
# Perform homing
homing_state.home_rails([rail], forcepos, homepos)
def _check_endstops(self, move):
end_pos = move.end_pos
for i in (0, 1, 2):
if (move.axes_d[i]
and (end_pos[i] < self.limits[i][0]
or end_pos[i] > self.limits[i][1])):
if self.limits[i][0] > self.limits[i][1]:
raise move.move_error("Must home axis first")
raise move.move_error()
def check_move(self, move):
limits = self.limits
xpos, ypos = move.end_pos[:2]
if (xpos < limits[0][0] or xpos > limits[0][1]
or ypos < limits[1][0] or ypos > limits[1][1]):
self._check_endstops(move)
if not move.axes_d[2]:
# Normal XY move - use defaults
return
# Move with Z - update velocity and accel for slower Z axis
self._check_endstops(move)
z_ratio = move.move_d / abs(move.axes_d[2])
move.limit_speed(
self.max_z_velocity * z_ratio, self.max_z_accel * z_ratio)
def get_status(self, eventtime):
axes = [a for a, (l, h) in zip("xyz", self.limits) if l <= h]
return {
'homed_axes': "".join(axes),
'axis_minimum': self.axes_min,
'axis_maximum': self.axes_max,
}
def load_kinematics(toolhead, config):
return CoreXYKinematics(toolhead, config)
# Parse gcode commands
#
# Copyright (C) 2016-2024 Kevin O'Connor <kevin@koconnor.net>
#
# 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', 'a', 'b', 'e'))
class GCodeCommand:
error = CommandError
def __init__(self, gcode, command, commandline, params, need_ack):
self._command = command
self._commandline = commandline
self._params = params
self._need_ack = need_ack
# 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
param_start = len(command)
param_end = len(origline)
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():
param_end = end
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
ok_msg = "ok"
if msg:
ok_msg = "ok %s" % (msg,)
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):
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))
return default
try:
value = parser(value)
except:
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))
if maxval is not None and value > 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))
if below is not None and value >= 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)
# 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)
# Command handling
self.is_printer_ready = False
self.mutex = printer.get_reactor().mutex()
self.output_callbacks = []
self.base_gcode_handlers = self.gcode_handlers = {}
self.ready_gcode_handlers = {}
self.mux_commands = {}
self.gcode_help = {}
self.status_commands = {}
# Register commands needed before config file is loaded
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)
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:
cmd = cmd.upper().split()[0]
val = float(cmd[1:])
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)
if cmd in self.ready_gcode_handlers:
del self.ready_gcode_handlers[cmd]
if cmd in self.base_gcode_handlers:
del self.base_gcode_handlers[cmd]
self._build_status_commands()
return old_cmd
if cmd in self.ready_gcode_handlers:
raise self.printer.config_error(
"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()):
raise self.printer.config_error(
"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
if when_not_ready:
self.base_gcode_handlers[cmd] = func
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:
handler = lambda gcmd: self._cmd_mux(cmd, gcmd)
self.register_command(cmd, handler, desc=desc)
self.mux_commands[cmd] = prev = (key, {})
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))
if value in prev_values:
raise self.printer.config_error(
"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}
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]
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
self.is_printer_ready = False
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*])')
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(';')
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':
# Skip line number at start of command
cmd = ''.join(parts[3:5]).strip()
else:
cmd = ''.join(parts[:3]).strip()
# Build gcode "params" dictionary
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)
try:
handler(gcmd)
except self.error as e:
self._respond_error(str(e))
self.printer.send_event("gcode:command_error")
if not need_ack:
raise
except:
msg = 'Internal error on command:"%s"' % (cmd,)
logging.exception(msg)
self.printer.invoke_shutdown(msg)
self._respond_error(msg)
if not need_ack:
raise
gcmd.ack()
def run_script_from_command(self, script):
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)
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')]
self.respond_raw("// " + "\n// ".join(lines))
def _respond_error(self, msg):
logging.warning(msg)
lines = msg.strip().split('\n')
if len(lines) > 1:
self.respond_info("\n".join(lines), log=False)
self.respond_raw('!! %s' % (lines[0].strip(),))
if self.is_fileinput:
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 = '#;'
try:
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(),))
# 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':
# Don't warn about temperature requests when not ready
gcmd.ack("T:0")
return
if cmd == 'M21':
# Don't warn about sd card init when not ready
return
if not self.is_printer_ready:
raise gcmd.error(self.printer.get_state_message()[0])
return
if not cmd:
cmdline = gcmd.get_commandline()
if cmdline:
logging.debug(cmdline)
return
if ' ' in cmd:
# Handle M117/M118 gcode with numeric and special characters
realcmd = cmd.split()[0]
if realcmd in ["M117", "M118", "M23"]:
handler = self.gcode_handlers.get(realcmd, None)
if handler is not None:
gcmd._command = realcmd
handler(gcmd)
return
elif cmd in ['M140', 'M104'] and not gcmd.get_float('S', 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)):
# 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:
key_param = gcmd.get(key, None)
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))
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')
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')
print_time = toolhead.get_last_move_time()
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')
cmd_FIRMWARE_RESTART_help = "Restart firmware, host, and reload config"
def cmd_FIRMWARE_RESTART(self, gcmd):
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")
return
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:
cmdhelp.append("Printer is not ready - not all commands available.")
cmdhelp.append("Available extended commands:")
for cmd in sorted(self.gcode_handlers):
if cmd in self.gcode_help:
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_mutex = self.gcode.get_mutex()
self.fd = printer.get_start_args().get("gcode_fd")
self.reactor = printer.get_reactor()
self.is_printer_ready = False
self.is_processing_data = False
self.is_fileinput = not not printer.get_start_args().get("debuginput")
self.pipe_is_active = True
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.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)
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|$)')
def _process_data(self, eventtime):
# Read input, separate by newline, and add to pending_commands
try:
data = str(os.read(self.fd, 4096).decode())
except (os.error, UnicodeDecodeError):
logging.exception("Read g-code")
return
self.input_log.append((eventtime, data))
self.bytes_read += len(data)
lines = data.split('\n')
lines[0] = self.partial_input + lines[0]
self.partial_input = lines.pop()
pending_commands = self.pending_commands
pending_commands.extend(lines)
self.pipe_is_active = True
# Special handling for debug file input EOF
if not data and self.is_fileinput:
if not self.is_processing_data:
self.reactor.unregister_fd(self.fd_handle)
self.fd_handle = None
self.gcode.request_restart('exit')
pending_commands.append("")
# Handle case where multiple commands pending
if self.is_processing_data or len(pending_commands) > 1:
if len(pending_commands) < 20:
# Check for M112 out-of-order
for line in lines:
if self.m112_r.match(line) is not None:
self.gcode.cmd_M112(None)
if self.is_processing_data:
if len(pending_commands) >= 20:
# Stop reading input
self.reactor.unregister_fd(self.fd_handle)
self.fd_handle = None
return
# Process commands
self.is_processing_data = True
while pending_commands:
self.pending_commands = []
with self.gcode_mutex:
self.gcode._process_commands(pending_commands)
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)
def _respond_raw(self, msg):
if self.pipe_is_active:
try:
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))