# Interactively identify printer axis motors via web ui prompts
#
# Copyright (C) 2026 the Klipper contributors
#
# This file may be distributed under the terms of the GNU GPLv3 license.
import glob, logging, os, re, time
AXES = "xyz"
STEPPER_NAME_R = re.compile(r'(?i)^stepper_([a-z])(\d*)$')
# Sections that follow a stepper by name (eg, "[tmc2209 stepper_x]")
DEP_SECTION_R = re.compile(r'(?i)^(tmc\d+|endstop_phase)\s+(stepper_\S+)$')
SECTION_HEADER_R = re.compile(r'^(\s*\[)([^\]]*)(\].*)$')
DIR_PIN_R = re.compile(r'^(\s*dir_pin\s*:\s*)([^#;]*)(.*)$', re.IGNORECASE)
AUTOSAVE_HEADER_R = re.compile(r'^#\*# \[([^\]]*)\]\s*$')
AUTOSAVE_OPTION_R = re.compile(r'^#\*# (\w+)\s*=\s*(.*)$')
INCLUDE_PREFIX = "include "
# Stepper options that describe the motor and its driver socket. When the
# changes are written through the SAVE_CONFIG mechanism these options are
# swapped between the stepper sections (the sections keep their names, so
# all changes end up in the "#*#" block at the end of the config file).
MOTOR_OPTIONS = ('step_pin', 'dir_pin', 'enable_pin', 'rotation_distance',
'microsteps', 'full_steps_per_rotation', 'gear_ratio',
'step_pulse_duration')
class AxisDiscovery:
def __init__(self, config):
self.printer = printer = config.get_printer()
self.gcode = printer.lookup_object('gcode')
self.wizard = None
# Register g-code commands
handlers = ['AXIS_DISCOVERY', 'AXIS_DISCOVERY_NEXT',
'AXIS_DISCOVERY_REPEAT', 'AXIS_DISCOVERY_ANSWER',
'AXIS_DISCOVERY_APPLY', 'AXIS_DISCOVERY_CANCEL']
for cmd in handlers:
func = getattr(self, 'cmd_' + cmd)
desc = getattr(self, 'cmd_' + cmd + '_help', None)
self.gcode.register_command(cmd, func, desc=desc)
printer.register_event_handler("klippy:disconnect",
self._handle_disconnect)
def _handle_disconnect(self):
self.wizard = None
# Prompt emission (compatible with the web ui "action:prompt_*" protocol)
def _prompt_action(self, action, msg=None):
self.gcode.respond_raw("// action:prompt_%s%s"
% (action, "" if msg is None else " " + msg))
def _prompt_begin(self, headline):
self._prompt_action("begin", headline)
def _prompt_text(self, msg):
self._prompt_action("text", msg)
def _prompt_button(self, msg):
self._prompt_action("button", msg)
def _prompt_footer_button(self, msg):
self._prompt_action("footer_button", msg)
def _prompt_show(self):
self._prompt_action("show")
def _prompt_end(self):
self._prompt_action("end")
# Wizard helpers
def _check_not_printing(self):
vsd = self.printer.lookup_object('virtual_sdcard', None)
if vsd is not None and vsd.is_active():
raise self.printer.command_error(
"Command not available during a print")
def _get_wizard(self, gcmd):
if self.wizard is None:
raise gcmd.error(
"No axis discovery in progress (run AXIS_DISCOVERY first)")
return self.wizard
def _lookup_motors(self):
toolhead = self.printer.lookup_object('toolhead')
kin = toolhead.get_kinematics()
if kin.__class__.__name__ != 'CartKinematics':
raise self.printer.command_error(
"AXIS_DISCOVERY supports [cartesian] kinematics only")
rails = getattr(kin, 'rails', None)
if rails is None or len(rails) != 3:
raise self.printer.command_error(
"AXIS_DISCOVERY is not supported with a dual carriage")
raw_config = self.printer.lookup_object('configfile').get_status(
0.)['config']
motors = []
for rail, axis in zip(rails, AXES):
for mcu_stepper in rail.get_steppers():
name = mcu_stepper.get_name()
m = STEPPER_NAME_R.match(name)
if m is None or m.group(1).lower() != axis:
raise self.printer.command_error(
"Unexpected stepper section '%s'" % (name,))
dir_pin = raw_config.get(name, {}).get('dir_pin', '')
motors.append({
'name': name, 'axis': axis, 'suffix': m.group(2),
'inverted': str(dir_pin).strip().startswith('!'),
'answer': None,
})
return motors
def _move_current(self, wizard):
self._check_not_printing()
motor = wizard['motors'][wizard['index']]
force_move = self.printer.lookup_object('force_move')
stepper_enable = self.printer.lookup_object('stepper_enable')
mcu_stepper = force_move.lookup_stepper(motor['name'])
did_enable = stepper_enable.set_motors_enable([motor['name']], True)
try:
force_move.manual_move(mcu_stepper, wizard['distance'],
wizard['speed'], wizard['accel'])
finally:
if did_enable:
stepper_enable.set_motors_enable([motor['name']], False)
def _show_question(self, wizard):
motor = wizard['motors'][wizard['index']]
self._prompt_begin("Axis Discovery (%d/%d)"
% (wizard['index'] + 1, len(wizard['motors'])))
self._prompt_text("Moving %s in the positive direction..."
% (motor['name'],))
self._prompt_text("Which motor position moved and in which"
" direction?")
for axis in AXES:
slots = [m for m in wizard['motors'] if m['axis'] == axis]
slots.sort(key=lambda m: (len(m['suffix']), m['suffix']))
for slot in slots:
short = axis.upper() + slot['suffix']
slot_param = ""
if slot['suffix']:
slot_param = " SLOT=%s" % (slot['suffix'],)
self._prompt_action("button_group_start")
self._prompt_button(
"%s+|AXIS_DISCOVERY_ANSWER I=%d AXIS=%s%s DIR=+"
% (short, wizard['index'], axis, slot_param))
self._prompt_button(
"%s-|AXIS_DISCOVERY_ANSWER I=%d AXIS=%s%s DIR=-"
% (short, wizard['index'], axis, slot_param))
self._prompt_action("button_group_end")
self._prompt_footer_button("Move again|AXIS_DISCOVERY_REPEAT")
self._prompt_footer_button("Cancel|AXIS_DISCOVERY_CANCEL")
self._prompt_show()
def _calc_changes(self, wizard):
motors = wizard['motors']
# Determine the target motor position of each motor. Answers
# without an explicit SLOT are assigned to the remaining free
# positions of their axis.
taken = set()
targets = [None] * len(motors)
for i, motor in enumerate(motors):
axis, suffix, sign = motor['answer']
if suffix is not None:
targets[i] = (axis, suffix)
taken.add(targets[i])
for i, motor in enumerate(motors):
if targets[i] is not None:
continue
axis = motor['answer'][0]
free = sorted(
[m['suffix'] for m in motors
if m['axis'] == axis and (axis, m['suffix']) not in taken],
key=lambda s: (len(s), s))
if not free:
raise self.printer.command_error(
"Axis %s has no free motor positions left"
% (axis.upper(),))
targets[i] = (axis, free[0])
taken.add(targets[i])
rename_map = {}
invert_new = set()
info = []
for i, motor in enumerate(motors):
t_axis, t_suffix = targets[i]
sign = motor['answer'][2]
new_name = "stepper_%s%s" % (t_axis, t_suffix)
label = motor['name']
if new_name != motor['name']:
rename_map[motor['name'].lower()] = new_name.lower()
label += " -> " + new_name
if sign < 0:
invert_new.add(new_name.lower())
label += " (dir inverted)"
info.append(label)
return rename_map, invert_new, info
def _show_summary(self, wizard):
rename_map, invert_new, info = self._calc_changes(wizard)
self._prompt_begin("Axis Discovery - confirm changes")
if not rename_map and not invert_new:
self._prompt_text("No changes required.")
else:
for line in info:
self._prompt_text(line)
reason = self._autosave_ready(rename_map, invert_new,
wizard['motors'])
if reason is None:
self._prompt_text("The changes are saved to the SAVE_CONFIG"
" block at the end of the config file.")
else:
self._prompt_text("The changes are written directly to the"
" config files (backups are kept):")
self._prompt_text(reason)
self._prompt_footer_button("Apply & Restart|AXIS_DISCOVERY_APPLY")
self._prompt_footer_button("Cancel|AXIS_DISCOVERY_CANCEL")
self._prompt_show()
# G-code commands
cmd_AXIS_DISCOVERY_help = "Interactively identify axis motors"
def cmd_AXIS_DISCOVERY(self, gcmd):
self._check_not_printing()
distance = gcmd.get_float('DISTANCE', 10., above=0., maxval=100.)
speed = gcmd.get_float('SPEED', 25., above=0., maxval=200.)
accel = gcmd.get_float('ACCEL', 0., minval=0.)
motors = self._lookup_motors()
# Manual stepper moves invalidate the toolhead position
toolhead = self.printer.lookup_object('toolhead')
toolhead.get_kinematics().clear_homing_state(AXES)
self.wizard = {'motors': motors, 'index': -1, 'spec': {},
'unspecified': {},
'distance': distance, 'speed': speed, 'accel': accel}
self._prompt_begin("Axis Discovery")
self._prompt_text("Detected motors:")
for motor in motors:
self._prompt_text("- %s (axis %s, dir_pin %s)"
% (motor['name'], motor['axis'].upper(),
"inverted" if motor['inverted']
else "normal"))
self._prompt_text("Each motor is moved %.1fmm in the positive"
" direction. Watch which carriage moves and where"
" it goes." % (distance,))
self._prompt_text("Homing is invalidated, the printer must be"
" re-homed after discovery.")
self._prompt_footer_button("Start|AXIS_DISCOVERY_NEXT I=0")
self._prompt_footer_button("Cancel|AXIS_DISCOVERY_CANCEL")
self._prompt_show()
cmd_AXIS_DISCOVERY_NEXT_help = "Move the next motor and request input"
def cmd_AXIS_DISCOVERY_NEXT(self, gcmd):
wizard = self._get_wizard(gcmd)
expected = gcmd.get_int('I', None)
if expected is not None and expected != wizard['index'] + 1:
# Stale prompt button (eg, clicked twice); ignore it
return
if wizard['index'] + 1 >= len(wizard['motors']):
self._show_summary(wizard)
return
wizard['index'] += 1
self._move_current(wizard)
self._show_question(wizard)
cmd_AXIS_DISCOVERY_REPEAT_help = "Repeat the move of the current motor"
def cmd_AXIS_DISCOVERY_REPEAT(self, gcmd):
wizard = self._get_wizard(gcmd)
if wizard['index'] < 0 or wizard['index'] >= len(wizard['motors']):
raise gcmd.error("No motor is currently being identified")
self._move_current(wizard)
self._show_question(wizard)
cmd_AXIS_DISCOVERY_ANSWER_help = \
"Report the observed motor position and direction"
def cmd_AXIS_DISCOVERY_ANSWER(self, gcmd):
wizard = self._get_wizard(gcmd)
expected = gcmd.get_int('I', None)
if expected is not None and expected != wizard['index']:
raise gcmd.error(
"Stale axis discovery answer (motor index %d expected)"
% (wizard['index'],))
if wizard['index'] < 0 or wizard['index'] >= len(wizard['motors']):
raise gcmd.error("No motor is awaiting an answer")
motor = wizard['motors'][wizard['index']]
axis = gcmd.get('AXIS').lower()
if axis not in AXES:
raise gcmd.error("Invalid AXIS '%s'" % (axis,))
suffix = gcmd.get('SLOT', None)
if suffix is not None and (axis, suffix) not in [
(m['axis'], m['suffix']) for m in wizard['motors']]:
raise gcmd.error("Invalid SLOT '%s' for axis %s"
% (suffix, axis.upper()))
direction = gcmd.get('DIR')
if direction == '+':
sign = 1
elif direction == '-':
sign = -1
else:
raise gcmd.error("Invalid DIR '%s' (must be + or -)"
% (direction,))
# Check that the target motor position is still free and that the
# axis does not receive more motors than it has positions
used = len([1 for (a, s) in wizard['spec'] if a == axis]) \
+ wizard['unspecified'].get(axis, 0)
avail = len([1 for m in wizard['motors'] if m['axis'] == axis])
if used >= avail:
raise gcmd.error(
"Axis %s has no free motor positions left"
% (axis.upper(),))
if suffix is None:
wizard['unspecified'][axis] = \
wizard['unspecified'].get(axis, 0) + 1
else:
if (axis, suffix) in wizard['spec']:
raise gcmd.error(
"Motor stepper_%s%s has already been assigned"
% (axis, suffix))
wizard['spec'][(axis, suffix)] = motor
motor['answer'] = (axis, suffix, sign)
wizard['index'] += 1
if wizard['index'] < len(wizard['motors']):
self._move_current(wizard)
self._show_question(wizard)
else:
self._show_summary(wizard)
cmd_AXIS_DISCOVERY_APPLY_help = "Apply axis discovery changes and restart"
def cmd_AXIS_DISCOVERY_APPLY(self, gcmd):
wizard = self._get_wizard(gcmd)
for motor in wizard['motors']:
if motor['answer'] is None:
raise gcmd.error("Motor %s has not been identified yet"
% (motor['name'],))
self._check_not_printing()
rename_map, invert_new, info = self._calc_changes(wizard)
if not rename_map and not invert_new:
self.wizard = None
self._prompt_end()
gcmd.respond_info("Axis discovery: no changes required")
return
reason = self._autosave_ready(rename_map, invert_new,
wizard['motors'])
self.wizard = None
self._prompt_end()
if reason is None:
# Save through the standard SAVE_CONFIG mechanism: the stepper
# sections keep their names and the motor/driver options are
# swapped between them, so all changes end up in the "#*#"
# block at the end of the config file
self._apply_autosave(rename_map, invert_new, wizard['motors'])
gcmd.respond_info("Axis discovery: changes saved to the"
" SAVE_CONFIG block, restarting...")
self.gcode.run_script_from_command("SAVE_CONFIG")
return
gcmd.respond_info("Axis discovery: %s; writing the config directly"
" (backups are kept)" % (reason,))
self._apply_config(gcmd, rename_map, invert_new)
gcmd.respond_info("Axis discovery: config updated, restarting...")
self.gcode.request_restart('restart')
cmd_AXIS_DISCOVERY_CANCEL_help = "Abort the axis discovery wizard"
def cmd_AXIS_DISCOVERY_CANCEL(self, gcmd):
self.wizard = None
self._prompt_end()
gcmd.respond_info("Axis discovery cancelled (re-home the printer)")
# SAVE_CONFIG support
def _source_of(self, rename_map, motors):
# Map each stepper section to the section whose motor/driver
# options it receives (sections keep their names)
source_of = {}
for motor in motors:
new_name = rename_map.get(motor['name'].lower(), motor['name'])
source_of[new_name.lower()] = motor['name']
return source_of
def _raw_section(self, raw, name):
if name in raw:
return raw[name]
return raw.get(name.lower(), {})
def _tmc_sections(self, raw, sec):
out = {}
for key, opts in raw.items():
parts = key.split(' ', 1)
if (len(parts) == 2 and parts[0].lower().startswith('tmc')
and parts[1].strip().lower() == sec.lower()):
out[parts[0]] = opts
return out
def _autosave_ready(self, rename_map, invert_new, motors):
# Returns None when the changes can be written through the
# SAVE_CONFIG mechanism, otherwise the reason they cannot
cfgname = self.printer.get_start_args()['config_file']
files = {}
self._read_config_file(cfgname, files, set())
main_path = os.path.abspath(cfgname)
include_sections = set()
for fname, lines in files.items():
if fname == main_path:
continue
for line in lines:
hm = SECTION_HEADER_R.match(line)
if hm is not None:
include_sections.add(hm.group(2).strip().lower())
raw = self.printer.lookup_object('configfile').get_status(
0.)['config']
for slot, src in self._source_of(rename_map, motors).items():
for name in set([slot, src]):
if name.lower() in include_sections:
return ("section [%s] is defined in an included config"
" file" % (name,))
ssec = self._raw_section(raw, slot)
osec = self._raw_section(raw, src)
for opt in MOTOR_OPTIONS:
if (opt in ssec) != (opt in osec):
return ("option '%s' is not defined in both [%s] and"
" [%s]" % (opt, slot, src))
s_tmc = self._tmc_sections(raw, slot)
o_tmc = self._tmc_sections(raw, src)
if sorted(s_tmc) != sorted(o_tmc):
return ("[%s] and [%s] do not use the same tmc driver"
" sections" % (slot, src))
for chip in s_tmc:
if sorted(s_tmc[chip]) != sorted(o_tmc[chip]):
return ("[%s %s] and [%s %s] do not define the same"
" options" % (chip, slot, chip, src))
return None
def _apply_autosave(self, rename_map, invert_new, motors):
pconfig = self.printer.lookup_object('configfile')
raw = pconfig.get_status(0.)['config']
for slot, src in self._source_of(rename_map, motors).items():
ssec = self._raw_section(raw, slot)
osec = self._raw_section(raw, src)
invert = slot in invert_new
for opt in MOTOR_OPTIONS:
if opt not in osec:
continue
val = str(osec[opt]).strip()
if opt == 'dir_pin' and invert:
val = self._invert_pin(val, slot)
if val == str(ssec.get(opt, '')).strip():
# No change necessary; keep the config file untouched
continue
pconfig.set(slot, opt, val)
s_tmc = self._tmc_sections(raw, slot)
o_tmc = self._tmc_sections(raw, src)
for chip, opts in o_tmc.items():
for opt, val in opts.items():
if str(val).strip() == str(
s_tmc.get(chip, {}).get(opt, '')).strip():
continue
pconfig.set("%s %s" % (chip, slot), opt, str(val).strip())
# Config file update support
def _map_section_name(self, name, rename_map):
stripped = name.strip()
low = stripped.lower()
if low in rename_map:
new_name = rename_map[low]
else:
m = DEP_SECTION_R.match(low)
if m is None or m.group(2) not in rename_map:
return name
new_name = "%s %s" % (m.group(1), rename_map[m.group(2)])
return name.replace(stripped, new_name, 1)
def _invert_pin(self, pin, section):
pin = pin.strip()
if not pin:
raise self.printer.command_error(
"Empty dir_pin in section '%s'" % (section,))
if pin.startswith('!'):
return pin[1:]
return '!' + pin
def _toggle_dir_pin(self, line_match, section):
new_pin = self._invert_pin(line_match.group(2), section)
return (line_match.group(1)
+ line_match.group(2).replace(
line_match.group(2).strip(), new_pin, 1)
+ line_match.group(3))
def _toggle_autosave_dir_pin(self, option_match, section):
new_pin = self._invert_pin(option_match.group(2), section)
return "#*# %s = %s" % (option_match.group(1), new_pin)
def _read_config_file(self, fname, files, visited):
fname = os.path.abspath(fname)
if fname in visited:
raise self.printer.command_error(
"Recursive include of config file '%s'" % (fname,))
visited.add(fname)
try:
f = open(fname, 'r')
data = f.read()
f.close()
except OSError:
raise self.printer.command_error(
"Unable to read config file '%s'" % (fname,))
data = data.replace('\r\n', '\n')
lines = data.split('\n')
files[fname] = lines
dirname = os.path.dirname(fname)
for line in lines:
hm = SECTION_HEADER_R.match(line)
if hm is None:
continue
name = hm.group(2).strip()
if not name.lower().startswith(INCLUDE_PREFIX):
continue
include_glob = os.path.join(dirname,
name[len(INCLUDE_PREFIX):].strip())
filenames = glob.glob(include_glob)
if not filenames and not glob.has_magic(include_glob):
raise self.printer.command_error(
"Include file '%s' does not exist" % (include_glob,))
for include_fname in sorted(filenames):
self._read_config_file(include_fname, files, visited)
visited.remove(fname)
def _apply_config(self, gcmd, rename_map, invert_new):
cfgname = self.printer.get_start_args()['config_file']
files = {}
self._read_config_file(cfgname, files, set())
# Apply renames and dir_pin inversions in memory (in the regular
# config and in the "#*#" SAVE_CONFIG autosave block, which would
# otherwise override the edits on the next start)
toggle_counts = {}
new_files = {}
for fname, lines in files.items():
cur_section = None
in_autosave = False
edits = 0
out = []
for line in lines:
hm = SECTION_HEADER_R.match(line)
am = None if hm is not None \
else AUTOSAVE_HEADER_R.match(line)
if hm is not None:
name = hm.group(2).strip()
if name.lower().startswith(INCLUDE_PREFIX):
cur_section = None
out.append(line)
continue
new_name = self._map_section_name(name, rename_map)
if new_name != name:
line = (hm.group(1)
+ hm.group(2).replace(name, new_name, 1)
+ hm.group(3))
edits += 1
cur_section = new_name.strip().lower()
in_autosave = False
out.append(line)
continue
if am is not None:
new_name = self._map_section_name(am.group(1), rename_map)
if new_name != am.group(1):
line = "#*# [%s]" % (new_name,)
edits += 1
cur_section = new_name.strip().lower()
in_autosave = True
out.append(line)
continue
if cur_section in invert_new:
if in_autosave:
om = AUTOSAVE_OPTION_R.match(line)
if om is not None and om.group(1).lower() == 'dir_pin':
line = self._toggle_autosave_dir_pin(om,
cur_section)
toggle_counts[cur_section] = \
toggle_counts.get(cur_section, 0) + 1
edits += 1
else:
dm = DIR_PIN_R.match(line)
if dm is not None:
line = self._toggle_dir_pin(dm, cur_section)
toggle_counts[cur_section] = \
toggle_counts.get(cur_section, 0) + 1
edits += 1
out.append(line)
if edits:
new_files[fname] = out
for name in invert_new:
if name not in toggle_counts:
raise gcmd.error("Unable to find dir_pin in section '%s'"
% (name,))
# Write all modified files to temporary names first so that a write
# failure leaves the current config untouched
stamp = time.strftime("-%Y%m%d_%H%M%S")
pending = []
try:
for fname, lines in new_files.items():
temp_name = fname + "_axisdiscovery.tmp"
f = open(temp_name, 'w')
f.write('\n'.join(lines))
f.close()
pending.append((fname, temp_name, fname + stamp))
except os.error as e:
for fname, temp_name, backup_name in pending:
try:
os.remove(temp_name)
except os.error:
pass
raise gcmd.error("Unable to write config file: %s" % (e,))
# Swap the new files into place (keeping a backup of each), rolling
# back already-swapped files if a rename fails. Note that the
# backups intentionally do not keep a ".cfg" suffix (unlike
# cmd_SAVE_CONFIG) so that common "[include *.cfg]" globs do not
# load the pre-edit sections on the next start.
committed = []
try:
for fname, temp_name, backup_name in pending:
os.rename(fname, backup_name)
os.rename(temp_name, fname)
committed.append((fname, backup_name))
except os.error as e:
for fname, backup_name in reversed(committed):
try:
os.rename(backup_name, fname)
except os.error:
logging.exception(
"axis_discovery: Unable to restore '%s'" % (fname,))
for fname, temp_name, backup_name in pending:
try:
os.remove(temp_name)
except os.error:
pass
raise gcmd.error("Unable to write config file '%s': %s"
% (fname, e))
for fname, backup_name in committed:
logging.info("axis_discovery: wrote '%s' (backup in '%s')",
fname, backup_name)
def load_config(config):
return AxisDiscovery(config)
Read 4 times, last 4 days ago