# coding=utf-8 # python 2 # Copyright (c) 2026 TormachTips.com. All rights reserved. # Licensed under the TormachTips Personal Use License. # Permission is granted only for private personal use and private personal modification. # No sharing, publication, distribution, resale, sublicensing, screenshots, code excerpts, # benchmarks, or videos are permitted without prior written permission. # Requests: tormach.1100m@gmail.com # Information page: https://tormachtips.com/plugins.htm ############################################# ## ## ## Recovery Mode Installer 0.96 ## ## www.tormachtips.com ## ## ## ############################################# # 0.96 - Replaced single line manual move with multi-line text area - 6/02/2026 # 0.95 - Public Beta - 6/01/2026 import os import re import time import fcntl import shutil import stat import gtk import glib import constants from ui_hooks import plugin CURRENT_VER = "0.96" SCRIPT_NAME = "Recovery Mode Installer" DESCRIPTION = "Installs PathPilot Recovery Mode Suite." ENABLED = 1 DEV_MACHINE = 1 DEV_MACHINE_FLAG = "/home/operator/gcode/python/dev_machine.txt" PATCH_LOCK_PATH = "/tmp/tt_pathpilot_file_patch.lock" PATCH_LOCK_TIMEOUT_SECONDS = 30.0 TORMACH_MILL_BASE_INI = "/home/operator/tmc/configs/tormach_mill/tormach_mill_base.ini" REMAP_PY = "/home/operator/tmc/configs/tormach_mill/python/remap.py" TOOL_CHANGE_NGC = "/home/operator/tmc/configs/tormach_mill/nc_subs/tormach_tool_change.ngc" NC_SUBS_DIR = "/home/operator/tmc/configs/tormach_mill/nc_subs" GCODE_PYTHON_DIR = "/home/operator/gcode/python" CHECKPOINT_NGC_PATH = os.path.join(NC_SUBS_DIR, "beagle_recovery_checkpoint.ngc") DIALOG_PATH = os.path.join(GCODE_PYTHON_DIR, "beagle_recovery_dialog.py") GLADE_CUSTOM_TAB_PLUGIN = os.path.join(GCODE_PYTHON_DIR, "glade_custom_tab_plugin.py") REQUEST_FILE = os.path.join(GCODE_PYTHON_DIR, "beagle_recovery_request.txt") STATE_FILE = os.path.join(GCODE_PYTHON_DIR, "beagle_recovery_state.txt") COORDS_FILE = os.path.join(GCODE_PYTHON_DIR, "beagle_recovery_coords.txt") DEBUG_LOG = os.path.join(GCODE_PYTHON_DIR, "beagle_recovery_debug.log") LOCK_FILE = os.path.join(GCODE_PYTHON_DIR, "beagle_recovery_dialog.lock") INI_PATCH_START = "# BEAGLE RECOVERY MODE START" INI_PATCH_END = "# BEAGLE RECOVERY MODE END" REMAP_PATCH_START = "# BEAGLE RECOVERY MODE START" REMAP_PATCH_END = "# BEAGLE RECOVERY MODE END" TOOL_CHANGE_PATCH_START = "(BEAGLE RECOVERY TOOL CHANGE PATCH START)" TOOL_CHANGE_PATCH_END = "(BEAGLE RECOVERY TOOL CHANGE PATCH END)" BEAGLE_INI_BLOCK = """# BEAGLE RECOVERY MODE START REMAP=M250 modalgroup=10 py=beagle_m6_recovery_M250 REMAP=M252 modalgroup=10 ngc=beagle_recovery_checkpoint REMAP=M253 modalgroup=10 py=beagle_recovery_poll_M253 # BEAGLE RECOVERY MODE END""" BEAGLE_REMAP_BLOCK = r'''# BEAGLE RECOVERY MODE START BEAGLE_RECOVERY_REQUEST_PATH = "/home/operator/gcode/python/beagle_recovery_request.txt" BEAGLE_RECOVERY_STATE_PATH = "/home/operator/gcode/python/beagle_recovery_state.txt" BEAGLE_RECOVERY_LOG_PATH = "/home/operator/gcode/python/beagle_recovery_debug.log" BEAGLE_RECOVERY_DIALOG_PATH = "/home/operator/gcode/python/beagle_recovery_dialog.py" BEAGLE_RECOVERY_COORDS_PATH = "/home/operator/gcode/python/beagle_recovery_coords.txt" BEAGLE_RECOVERY_STOP_SPINDLE = 1 BEAGLE_RECOVERY_SMALL_STEP = 0.1000 BEAGLE_RECOVERY_LARGE_STEP = 1.0000 def beagle_recovery_log(message): try: timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f") f = open(BEAGLE_RECOVERY_LOG_PATH, "a") f.write("%s | %s\n" % (timestamp, message)) f.close() except Exception: pass def beagle_recovery_read_file(path): try: f = open(path, "r") value = f.read().strip().upper() f.close() return value except Exception: return "" def beagle_recovery_write_file(path, value): try: f = open(path, "w") f.write(value) f.close() except Exception: pass def beagle_recovery_request_file_has_data(): try: import os return os.path.isfile(BEAGLE_RECOVERY_REQUEST_PATH) and os.path.getsize(BEAGLE_RECOVERY_REQUEST_PATH) > 0 except Exception: return False def beagle_recovery_launch_dialog(): try: import subprocess subprocess.Popen(["/usr/bin/python", BEAGLE_RECOVERY_DIALOG_PATH]) except Exception as e: beagle_recovery_log("dialog launch failed | %s" % str(e)) def beagle_recovery_wcs_name(index_value): try: index_value = int(index_value) except Exception: index_value = 1 if index_value == 1: return "G54" if index_value == 2: return "G55" if index_value == 3: return "G56" if index_value == 4: return "G57" if index_value == 5: return "G58" if index_value == 6: return "G59" if index_value == 7: return "G59.1" if index_value == 8: return "G59.2" if index_value == 9: return "G59.3" return "G54" def beagle_recovery_format_positions_from_status(status): status.poll() x = float(status.actual_position[0]) y = float(status.actual_position[1]) z = float(status.actual_position[2]) a = float(status.actual_position[3]) wcs_name = beagle_recovery_wcs_name(status.g5x_index) try: g5x = status.g5x_offset except Exception: g5x = [0.0, 0.0, 0.0, 0.0] try: g92 = status.g92_offset except Exception: g92 = [0.0, 0.0, 0.0, 0.0] try: tool = status.tool_offset except Exception: tool = [0.0, 0.0, 0.0, 0.0] wx = x - float(g5x[0]) - float(g92[0]) wy = y - float(g5x[1]) - float(g92[1]) wz = z - float(g5x[2]) - float(g92[2]) - float(tool[2]) wa = a - float(g5x[3]) - float(g92[3]) g53_line = "G53 X%.4f Y%.4f Z%.4f A%.4f" % (x, y, z, a) wcs_line = "%s X% .4f Y% .4f Z% .4f A% .4f" % (wcs_name, wx, wy, wz, wa) return g53_line, wcs_name, wcs_line def beagle_recovery_capture_g53_position(self): try: g53_line, wcs_name, wcs_line = beagle_recovery_format_positions_from_status(self.status) x = float(self.status.actual_position[0]) y = float(self.status.actual_position[1]) z = float(self.status.actual_position[2]) a = float(self.status.actual_position[3]) self.params["_beagle_recovery_return_x"] = x self.params["_beagle_recovery_return_y"] = y self.params["_beagle_recovery_return_z"] = z self.params["_beagle_recovery_return_a"] = a beagle_recovery_write_file( BEAGLE_RECOVERY_COORDS_PATH, "Saved G53: %s\n Paused at: %s" % (g53_line, wcs_line)) beagle_recovery_log("captured return position | %s | %s" % (g53_line, wcs_line)) except Exception as e: beagle_recovery_write_file( BEAGLE_RECOVERY_COORDS_PATH, "Saved G53: position unavailable\nPaused at: position unavailable") beagle_recovery_log("capture return position failed | %s" % str(e)) def beagle_recovery_capture_spindle_state(self): try: self.params["_beagle_recovery_stop_spindle"] = float(BEAGLE_RECOVERY_STOP_SPINDLE) self.params["_beagle_recovery_spindle_was_on"] = 0.0 self.params["_beagle_recovery_spindle_rpm"] = 0.0 self.params["_beagle_recovery_spindle_dir"] = 3.0 if BEAGLE_RECOVERY_STOP_SPINDLE == 0: beagle_recovery_log("spindle capture skipped | stop disabled") return self.status.poll() rpm = 0.0 try: rpm = float(self.params["_speed"]) except Exception: try: rpm = float(self.status.settings[2]) except Exception: rpm = 0.0 spindle_on = False try: spindle_on = bool(self.status.spindle_enabled) except Exception: spindle_on = rpm > 0.0 if spindle_on and rpm > 0.0: self.params["_beagle_recovery_spindle_was_on"] = 1.0 self.params["_beagle_recovery_spindle_rpm"] = rpm try: if self.spindle_turning == emccanon.CANON_COUNTERCLOCKWISE: self.params["_beagle_recovery_spindle_dir"] = 4.0 except Exception: pass beagle_recovery_log("captured spindle | on rpm %.1f dir %.0f" % ( self.params["_beagle_recovery_spindle_rpm"], self.params["_beagle_recovery_spindle_dir"])) else: beagle_recovery_log("captured spindle | off rpm %.1f" % rpm) except Exception as e: self.params["_beagle_recovery_stop_spindle"] = 0.0 self.params["_beagle_recovery_spindle_was_on"] = 0.0 beagle_recovery_log("spindle capture failed | %s" % str(e)) def beagle_recovery_reset_params(self): self.params["_beagle_recovery_request_code"] = 0.0 self.params["_beagle_recovery_source"] = 0.0 self.params["_beagle_recovery_stop_spindle"] = 0.0 self.params["_beagle_recovery_spindle_was_on"] = 0.0 self.params["_beagle_recovery_spindle_rpm"] = 0.0 self.params["_beagle_recovery_spindle_dir"] = 3.0 self.params["_beagle_recovery_move_axis"] = -1.0 self.params["_beagle_recovery_move_distance"] = 0.0 self.params["_beagle_recovery_move_feed"] = 5.0 self.params["_beagle_recovery_manual_g"] = 1.0 self.params["_beagle_recovery_manual_f"] = 5.0 self.params["_beagle_recovery_manual_x"] = 0.0 self.params["_beagle_recovery_manual_y"] = 0.0 self.params["_beagle_recovery_manual_z"] = 0.0 self.params["_beagle_recovery_manual_a"] = 0.0 self.params["_beagle_recovery_manual_mode"] = 91.0 self.params["_beagle_recovery_spindle_request_rpm"] = 0.0 self.params["_beagle_recovery_manual_has_x"] = 0.0 self.params["_beagle_recovery_manual_has_y"] = 0.0 self.params["_beagle_recovery_manual_has_z"] = 0.0 self.params["_beagle_recovery_manual_has_a"] = 0.0 self.params["_beagle_recovery_manual_axis_combo"] = 0.0 def beagle_recovery_set_waiting(self, source_code): beagle_recovery_write_file(BEAGLE_RECOVERY_STATE_PATH, "WAITING") self.params["_beagle_recovery_request_code"] = 10.0 self.params["_beagle_recovery_source"] = float(source_code) def beagle_recovery_parse_spindle_command(self, request): try: text = request.strip().upper() if text.startswith("GCODE:"): text = text[6:].strip() text = text.replace(",", " ") words = text.split() if len(words) < 1: return False # Dialog button format: SPINDLE_CW:2500 / SPINDLE_CCW:2500 / SPINDLE_STOP if text.startswith("SPINDLE_CW:"): rpm = float(text.split(":", 1)[1].strip()) if rpm <= 0.0: beagle_recovery_log("spindle command rejected | rpm must be positive | %s" % text) return False self.params["_beagle_recovery_spindle_request_rpm"] = rpm self.params["_beagle_recovery_request_code"] = 30.0 beagle_recovery_log("spindle command accepted | M3 S%.1f" % rpm) return True if text.startswith("SPINDLE_CCW:"): rpm = float(text.split(":", 1)[1].strip()) if rpm <= 0.0: beagle_recovery_log("spindle command rejected | rpm must be positive | %s" % text) return False self.params["_beagle_recovery_spindle_request_rpm"] = rpm self.params["_beagle_recovery_request_code"] = 31.0 beagle_recovery_log("spindle command accepted | M4 S%.1f" % rpm) return True if text == "SPINDLE_STOP": self.params["_beagle_recovery_spindle_request_rpm"] = 0.0 self.params["_beagle_recovery_request_code"] = 32.0 beagle_recovery_log("spindle command accepted | M5") return True # Typed manual format: M3 S2500 / M4 S2500 / M5 m_word = words[0] if m_word in ("M5", "M05"): self.params["_beagle_recovery_spindle_request_rpm"] = 0.0 self.params["_beagle_recovery_request_code"] = 32.0 beagle_recovery_log("spindle command accepted | M5") return True if m_word not in ("M3", "M03", "M4", "M04"): return False rpm = 0.0 for word in words[1:]: if len(word) < 2: beagle_recovery_log("spindle command rejected | bad word | %s" % word) return False if word[0] != "S": beagle_recovery_log("spindle command rejected | only S word allowed with M3/M4 | %s" % text) return False rpm = float(word[1:]) if rpm <= 0.0: beagle_recovery_log("spindle command rejected | M3/M4 requires positive S word | %s" % text) return False self.params["_beagle_recovery_spindle_request_rpm"] = rpm if m_word in ("M3", "M03"): self.params["_beagle_recovery_request_code"] = 30.0 beagle_recovery_log("spindle command accepted | M3 S%.1f" % rpm) else: self.params["_beagle_recovery_request_code"] = 31.0 beagle_recovery_log("spindle command accepted | M4 S%.1f" % rpm) return True except Exception as e: beagle_recovery_log("spindle command rejected | exception | %s | %s" % (request, str(e))) return False def beagle_recovery_parse_manual_move(self, request): try: text = request.strip().upper() manual_mode = 91.0 if text.startswith("GCODE90:"): manual_mode = 90.0 text = text[8:].strip() elif text.startswith("GCODE91:"): manual_mode = 91.0 text = text[8:].strip() elif text.startswith("GCODE:"): manual_mode = 91.0 text = text[6:].strip() text = text.replace(",", " ") words = text.split() if len(words) < 1: return False g_mode = None feed = 5.0 x = 0.0 y = 0.0 z = 0.0 a = 0.0 has_x = 0.0 has_y = 0.0 has_z = 0.0 has_a = 0.0 axis_seen = False for word in words: if len(word) < 2: beagle_recovery_log("manual move rejected | bad word | %s" % word) return False letter = word[0] value_text = word[1:] if letter == "G": if value_text in ("0", "00"): g_mode = 0.0 elif value_text in ("1", "01"): g_mode = 1.0 else: beagle_recovery_log("manual move rejected | only G0/G1 allowed | %s" % text) return False elif letter == "F": feed = float(value_text) if feed <= 0.0: beagle_recovery_log("manual move rejected | feed must be positive | %s" % text) return False elif letter == "X": x = float(value_text) has_x = 1.0 axis_seen = True elif letter == "Y": y = float(value_text) has_y = 1.0 axis_seen = True elif letter == "Z": z = float(value_text) has_z = 1.0 axis_seen = True elif letter == "A": a = float(value_text) has_a = 1.0 axis_seen = True else: beagle_recovery_log("manual move rejected | unsupported word %s | %s" % (letter, text)) return False if g_mode is None: beagle_recovery_log("manual move rejected | missing G0/G1 | %s" % text) return False if axis_seen == False: beagle_recovery_log("manual move rejected | no axis word | %s" % text) return False self.params["_beagle_recovery_manual_g"] = g_mode self.params["_beagle_recovery_manual_f"] = feed self.params["_beagle_recovery_manual_x"] = x self.params["_beagle_recovery_manual_y"] = y self.params["_beagle_recovery_manual_z"] = z self.params["_beagle_recovery_manual_a"] = a self.params["_beagle_recovery_manual_has_x"] = has_x self.params["_beagle_recovery_manual_has_y"] = has_y self.params["_beagle_recovery_manual_has_z"] = has_z self.params["_beagle_recovery_manual_has_a"] = has_a axis_combo = 0.0 if has_x == 1.0: axis_combo += 1.0 if has_y == 1.0: axis_combo += 2.0 if has_z == 1.0: axis_combo += 4.0 if has_a == 1.0: axis_combo += 8.0 self.params["_beagle_recovery_manual_axis_combo"] = axis_combo self.params["_beagle_recovery_manual_mode"] = manual_mode self.params["_beagle_recovery_request_code"] = 21.0 beagle_recovery_log( "manual move accepted | mode G%.0f | %s | G%.0f F%.4f X%.4f Y%.4f Z%.4f A%.4f | has X%.0f Y%.0f Z%.0f A%.0f" % (manual_mode, text, g_mode, feed, x, y, z, a, has_x, has_y, has_z, has_a)) return True except Exception as e: beagle_recovery_log("manual move rejected | exception | %s | %s" % (request, str(e))) return False def beagle_recovery_poll_common(self, source_code): if self.task == 0: return INTERP_OK try: # request codes: # 0 = no request # 1 = G30 # 2 = LEFT # 3 = RIGHT # 4 = ZUP # 5 = ZDOWN # 6 = RETURN TO SAVED G53 # 8 = START RECOVERY # 9 = DONE # 10 = WAITING / RECOVERY ACTIVE # # source codes: # 1 = M6 # 2 = CHECKPOINT beagle_recovery_reset_params(self) state = beagle_recovery_read_file(BEAGLE_RECOVERY_STATE_PATH) if beagle_recovery_request_file_has_data() == False: if state == "WAITING": self.params["_beagle_recovery_request_code"] = 10.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK request = beagle_recovery_read_file(BEAGLE_RECOVERY_REQUEST_PATH) beagle_recovery_write_file(BEAGLE_RECOVERY_REQUEST_PATH, "") if request == "": if state == "WAITING": self.params["_beagle_recovery_request_code"] = 10.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK beagle_recovery_log("request read and cleared | source %.0f | %s" % (float(source_code), request)) self.error_handler.log("BEAGLE recovery request: %s" % request) if request == "M6_RECOVER": beagle_recovery_capture_g53_position(self) beagle_recovery_set_waiting(self, 1) beagle_recovery_launch_dialog() return INTERP_OK if request == "CHECKPOINT_RECOVER" or request == "PAUSE": beagle_recovery_capture_g53_position(self) beagle_recovery_capture_spindle_state(self) beagle_recovery_set_waiting(self, 2) beagle_recovery_launch_dialog() return INTERP_OK if request == "G30": self.params["_beagle_recovery_request_code"] = 1.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK if request == "LEFT": self.params["_beagle_recovery_request_code"] = 2.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK if request == "RIGHT": self.params["_beagle_recovery_request_code"] = 3.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK if request == "ZUP": self.params["_beagle_recovery_request_code"] = 4.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK if request == "ZDOWN": self.params["_beagle_recovery_request_code"] = 5.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK if request == "RETURN" or request == "RETURN_XYZ": self.params["_beagle_recovery_request_code"] = 6.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK if request == "RETURN_XY": self.params["_beagle_recovery_request_code"] = 7.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK move_map = { "XN1": (0.0, -BEAGLE_RECOVERY_LARGE_STEP, 20.0), "XN01": (0.0, -BEAGLE_RECOVERY_SMALL_STEP, 20.0), "XP01": (0.0, BEAGLE_RECOVERY_SMALL_STEP, 20.0), "XP1": (0.0, BEAGLE_RECOVERY_LARGE_STEP, 20.0), "YN1": (1.0, -BEAGLE_RECOVERY_LARGE_STEP, 20.0), "YN01": (1.0, -BEAGLE_RECOVERY_SMALL_STEP, 20.0), "YP01": (1.0, BEAGLE_RECOVERY_SMALL_STEP, 20.0), "YP1": (1.0, BEAGLE_RECOVERY_LARGE_STEP, 20.0), "ZN1": (2.0, -BEAGLE_RECOVERY_LARGE_STEP, 5.0), "ZN01": (2.0, -BEAGLE_RECOVERY_SMALL_STEP, 5.0), "ZP01": (2.0, BEAGLE_RECOVERY_SMALL_STEP, 5.0), "ZP1": (2.0, BEAGLE_RECOVERY_LARGE_STEP, 5.0),} move_request = request move_feed_override = None if request.startswith("MOVE:"): try: parts = request.split(":") if len(parts) >= 2: move_request = parts[1].strip().upper() if len(parts) >= 3: move_feed_override = float(parts[2]) if move_feed_override <= 0.0: move_feed_override = None except Exception: move_request = "" move_feed_override = None if move_request == "RETURN" or move_request == "RETURN_XYZ": feed = 20.0 if move_feed_override is not None: feed = move_feed_override self.params["_beagle_recovery_request_code"] = 6.0 self.params["_beagle_recovery_source"] = float(source_code) self.params["_beagle_recovery_move_feed"] = feed return INTERP_OK if move_request == "RETURN_XY": feed = 20.0 if move_feed_override is not None: feed = move_feed_override self.params["_beagle_recovery_request_code"] = 7.0 self.params["_beagle_recovery_source"] = float(source_code) self.params["_beagle_recovery_move_feed"] = feed return INTERP_OK if move_request in move_map: axis, distance, feed = move_map[move_request] if move_feed_override is not None: feed = move_feed_override self.params["_beagle_recovery_request_code"] = 20.0 self.params["_beagle_recovery_source"] = float(source_code) self.params["_beagle_recovery_move_axis"] = axis self.params["_beagle_recovery_move_distance"] = distance self.params["_beagle_recovery_move_feed"] = feed return INTERP_OK if request.startswith("SPINDLE_") or request.startswith("GCODE:M"): if beagle_recovery_parse_spindle_command(self, request): self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK self.error_handler.log("BEAGLE recovery spindle command rejected. Use M3 S####, M4 S####, or M5.") if state == "WAITING": self.params["_beagle_recovery_request_code"] = 10.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK if request.startswith("GCODE:") or request.startswith("GCODE90:") or request.startswith("GCODE91:"): if beagle_recovery_parse_manual_move(self, request): self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK self.error_handler.log("BEAGLE recovery manual move rejected. Use one G0/G1 line with only X/Y/Z/A/F words.") if state == "WAITING": self.params["_beagle_recovery_request_code"] = 10.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK if request == "RUN_BLOCK": self.params["_beagle_recovery_request_code"] = 50.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK if request == "DONE": beagle_recovery_write_file(BEAGLE_RECOVERY_STATE_PATH, "IDLE") self.params["_beagle_recovery_request_code"] = 9.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK beagle_recovery_log("unknown request ignored | %s" % request) if state == "WAITING": self.params["_beagle_recovery_request_code"] = 10.0 self.params["_beagle_recovery_source"] = float(source_code) return INTERP_OK except Exception as e: beagle_recovery_log("exception | %s" % str(e)) self.set_errormsg("BEAGLE recovery poll failed: %s" % str(e)) return INTERP_ERROR def beagle_m6_recovery_M250(self, **words): return beagle_recovery_poll_common(self, 1) def beagle_recovery_poll_M253(self, **words): return beagle_recovery_poll_common(self, 2) # BEAGLE RECOVERY MODE END''' BEAGLE_CHECKPOINT_NGC = r'''o sub (Default source is CHECKPOINT. If called as o<...> call [1], source is M6.) #<_beagle_recovery_ngc_source> = 2 o9300 if [#1 EQ 1] #<_beagle_recovery_ngc_source> = 1 o9300 endif #<_beagle_recovery_request_code> = 0 M66 P63 L0 o9301 if [#<_beagle_recovery_ngc_source> EQ 1] M250 o9301 else M253 o9301 endif M66 P63 L0 o9302 if [#<_beagle_recovery_request_code> EQ 0] M66 P63 L0 o return o9302 endif o9303 if [#<_beagle_recovery_stop_spindle> EQ 1] M5 M66 P63 L0 o9303 endif o9304 while [#<_beagle_recovery_request_code> NE 9] M66 P63 L0 o9305 if [#<_beagle_recovery_ngc_source> EQ 1] M250 o9305 else M253 o9305 endif M66 P63 L0 (Legacy Go To G30 request) o9310 if [#<_beagle_recovery_request_code> EQ 1] G30 G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9310 endif (Legacy X negative 2.000 request) o9311 if [#<_beagle_recovery_request_code> EQ 2] G91 G1 F20 X-2.0000 G90 G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9311 endif (Legacy X positive 2.000 request) o9312 if [#<_beagle_recovery_request_code> EQ 3] G91 G1 F20 X2.0000 G90 G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9312 endif (Legacy Z positive 0.100 request) o9313 if [#<_beagle_recovery_request_code> EQ 4] G91 G1 F5 Z0.1000 G90 G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9313 endif (Legacy Z negative 0.100 request) o9314 if [#<_beagle_recovery_request_code> EQ 5] G91 G1 F5 Z-0.1000 G90 G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9314 endif (Return to captured G53 XYZ position) o9315 if [#<_beagle_recovery_request_code> EQ 6] G90 G53 G1 F#<_beagle_recovery_move_feed> X#<_beagle_recovery_return_x> Y#<_beagle_recovery_return_y> Z#<_beagle_recovery_return_z> A#<_beagle_recovery_return_a> G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9315 endif (Return to captured G53 XY position only) o9326 if [#<_beagle_recovery_request_code> EQ 7] G90 G53 G1 F#<_beagle_recovery_move_feed> X#<_beagle_recovery_return_x> Y#<_beagle_recovery_return_y> G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9326 endif (Pendant-style incremental move request) o9316 if [#<_beagle_recovery_request_code> EQ 20] o9317 if [#<_beagle_recovery_move_axis> EQ 0] G91 G1 F#<_beagle_recovery_move_feed> X#<_beagle_recovery_move_distance> G90 o9317 endif o9318 if [#<_beagle_recovery_move_axis> EQ 1] G91 G1 F#<_beagle_recovery_move_feed> Y#<_beagle_recovery_move_distance> G90 o9318 endif o9319 if [#<_beagle_recovery_move_axis> EQ 2] G91 G1 F#<_beagle_recovery_move_feed> Z#<_beagle_recovery_move_distance> G90 o9319 endif G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9316 endif (Manual typed G0/G1 move request. Mode can be G90 absolute or G91 incremental.) o9322 if [#<_beagle_recovery_request_code> EQ 21] o9323 if [#<_beagle_recovery_manual_mode> EQ 91] G91 o9323 else G90 o9323 endif o9324 if [#<_beagle_recovery_manual_g> EQ 1] F#<_beagle_recovery_manual_f> o9324 endif (X only) o9335 if [#<_beagle_recovery_manual_axis_combo> EQ 1] o9336 if [#<_beagle_recovery_manual_g> EQ 0] G0 X#<_beagle_recovery_manual_x> o9336 else G1 X#<_beagle_recovery_manual_x> o9336 endif o9335 endif (Y only) o9337 if [#<_beagle_recovery_manual_axis_combo> EQ 2] o9338 if [#<_beagle_recovery_manual_g> EQ 0] G0 Y#<_beagle_recovery_manual_y> o9338 else G1 Y#<_beagle_recovery_manual_y> o9338 endif o9337 endif (X Y) o9339 if [#<_beagle_recovery_manual_axis_combo> EQ 3] o9340 if [#<_beagle_recovery_manual_g> EQ 0] G0 X#<_beagle_recovery_manual_x> Y#<_beagle_recovery_manual_y> o9340 else G1 X#<_beagle_recovery_manual_x> Y#<_beagle_recovery_manual_y> o9340 endif o9339 endif (Z only) o9341 if [#<_beagle_recovery_manual_axis_combo> EQ 4] o9342 if [#<_beagle_recovery_manual_g> EQ 0] G0 Z#<_beagle_recovery_manual_z> o9342 else G1 Z#<_beagle_recovery_manual_z> o9342 endif o9341 endif (X Z) o9343 if [#<_beagle_recovery_manual_axis_combo> EQ 5] o9344 if [#<_beagle_recovery_manual_g> EQ 0] G0 X#<_beagle_recovery_manual_x> Z#<_beagle_recovery_manual_z> o9344 else G1 X#<_beagle_recovery_manual_x> Z#<_beagle_recovery_manual_z> o9344 endif o9343 endif (Y Z) o9345 if [#<_beagle_recovery_manual_axis_combo> EQ 6] o9346 if [#<_beagle_recovery_manual_g> EQ 0] G0 Y#<_beagle_recovery_manual_y> Z#<_beagle_recovery_manual_z> o9346 else G1 Y#<_beagle_recovery_manual_y> Z#<_beagle_recovery_manual_z> o9346 endif o9345 endif (X Y Z) o9347 if [#<_beagle_recovery_manual_axis_combo> EQ 7] o9348 if [#<_beagle_recovery_manual_g> EQ 0] G0 X#<_beagle_recovery_manual_x> Y#<_beagle_recovery_manual_y> Z#<_beagle_recovery_manual_z> o9348 else G1 X#<_beagle_recovery_manual_x> Y#<_beagle_recovery_manual_y> Z#<_beagle_recovery_manual_z> o9348 endif o9347 endif (A only) o9349 if [#<_beagle_recovery_manual_axis_combo> EQ 8] o9350 if [#<_beagle_recovery_manual_g> EQ 0] G0 A#<_beagle_recovery_manual_a> o9350 else G1 A#<_beagle_recovery_manual_a> o9350 endif o9349 endif G90 G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9322 endif (Spindle clockwise request) o9330 if [#<_beagle_recovery_request_code> EQ 30] M3 S#<_beagle_recovery_spindle_request_rpm> G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9330 endif (Spindle counter-clockwise request) o9331 if [#<_beagle_recovery_request_code> EQ 31] M4 S#<_beagle_recovery_spindle_request_rpm> G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9331 endif (Spindle stop request) o9332 if [#<_beagle_recovery_request_code> EQ 32] M5 G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9332 endif (Run temporary recovery block) o9360 if [#<_beagle_recovery_request_code> EQ 50] o call G90 G4 P0.25 M66 P63 L0 #<_beagle_recovery_request_code> = 10 o9360 endif G4 P0.10 M66 P63 L0 o9304 endwhile (If this recovery was entered from M6, force spindle off before returning to tool-change prompt.) o9333 if [#<_beagle_recovery_ngc_source> EQ 1] M5 M66 P63 L0 o9333 endif o9320 if [[#<_beagle_recovery_spindle_was_on> EQ 1] AND [#<_beagle_recovery_ngc_source> NE 1]] o9321 if [#<_beagle_recovery_spindle_dir> EQ 4] M4 S#<_beagle_recovery_spindle_rpm> o9321 else M3 S#<_beagle_recovery_spindle_rpm> o9321 endif M66 P63 L0 o9320 endif o endsub M2''' BEAGLE_DIALOG_PY = r'''#!/usr/bin/python # coding=utf-8 # python 2 # Copyright (c) 2026 TormachTips.com. All rights reserved. # Licensed under the TormachTips Personal Use License. # Permission is granted only for private personal use and private personal modification. # No sharing, publication, distribution, resale, sublicensing, screenshots, code excerpts, # benchmarks, or videos are permitted without prior written permission. # Requests: tormach.1100m@gmail.com # Information page: https://tormachtips.com/plugins.htm ############################################# ## ## ## Beagle Recovery Dialog 1.00 ## ## www.tormachtips.com ## ## ## ############################################# import os import sys import gtk import fcntl import datetime import re try: import linuxcnc except Exception: linuxcnc = None BASE_PATH = "/home/operator/gcode/python" REQUEST_FILE = os.path.join(BASE_PATH, "beagle_recovery_request.txt") STATE_FILE = os.path.join(BASE_PATH, "beagle_recovery_state.txt") COORDS_FILE = os.path.join(BASE_PATH, "beagle_recovery_coords.txt") LOCK_FILE = os.path.join(BASE_PATH, "beagle_recovery_dialog.lock") LOG_FILE = os.path.join(BASE_PATH, "beagle_recovery_debug.log") BLOCK_NGC_FILE = "/home/operator/tmc/configs/tormach_mill/nc_subs/beagle_recovery_block.ngc" SOFT_LIMIT_MARGIN = 0.0001 RECOVERY_SMALL_STEP = 0.1000 RECOVERY_LARGE_STEP = 1.0000 PENDANT_REPEAT_MS = 175 INI_LIMIT_FILES = [ "/home/operator/tmc/configs/tormach_mill/tormach_mill_base.ini", "/home/operator/tmc/configs/tormach_mill/tormach_mill.ini",] def log(message): try: timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f") f = open(LOG_FILE, "a") f.write("%s | dialog | %s\n" % (timestamp, message)) f.close() except Exception: pass def read_file(path, default_value=""): try: f = open(path, "r") value = f.read().strip() f.close() if value: return value except Exception: pass return default_value def read_state(): return read_file(STATE_FILE, "").upper() def read_coords(): return read_file(COORDS_FILE, "G53 position unavailable") def write_request(value): try: f = open(REQUEST_FILE, "w") f.write(value) f.close() log("request written | %s" % value) except Exception as e: log("request write failed | %s" % str(e)) def acquire_lock(): fp = open(LOCK_FILE, "w") try: fcntl.flock(fp.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) return fp except IOError: fp.close() return None class BeagleRecoveryDialog(object): def __init__(self): self.stat = None self.soft_limits = {} self.saved_g53_position = None self._repeat_request = None self._repeat_active = False if linuxcnc is not None: try: self.stat = linuxcnc.stat() except Exception: self.stat = None self.dialog = gtk.Dialog("Mid-Cycle Recovery Mode", None, gtk.DIALOG_DESTROY_WITH_PARENT) self.dialog.set_position(gtk.WIN_POS_CENTER) self.dialog.set_keep_above(True) self.dialog.set_modal(False) self.dialog.set_default_size(760, 430) coords_text = read_coords() self.saved_g53_position = self.parse_saved_g53_position(coords_text) self.soft_limits = self.load_soft_limits() header_box = gtk.VBox(False, 4) header_box.set_border_width(8) self.dialog.vbox.pack_start(header_box, False, False, 0) title_label = gtk.Label("You are in Recovery Mode. Be careful! All moves below are INCREMENTAL!") title_label.set_alignment(0.0, 0.5) title_label.set_line_wrap(False) title_label.set_size_request(780, 24) header_box.pack_start(title_label, False, False, 0) saved_lines = coords_text.splitlines() saved_g53_text = "Saved G53: unavailable" saved_wcs_text = "hams3 WCS: unavailable" if len(saved_lines) >= 1: saved_g53_text = saved_lines[0] if len(saved_lines) >= 2: saved_wcs_text = saved_lines[1] # saved_g53_label = gtk.Label(saved_g53_text) # saved_g53_label.set_alignment(0.0, 0.5) # saved_g53_label.set_line_wrap(False) # saved_g53_label.set_size_request(920, 24) # header_box.pack_start(saved_g53_label, False, False, 0) saved_wcs_label = gtk.Label(saved_wcs_text) saved_wcs_label.set_use_markup(True) saved_wcs_label.set_markup("%s" % saved_wcs_text) saved_wcs_label.set_alignment(0.0, 0.5) saved_wcs_label.set_line_wrap(False) saved_wcs_label.set_size_request(780, 24) header_box.pack_start(saved_wcs_label, False, False, 0) # self.actual_g53_label = gtk.Label("Current G53: unavailable") # self.actual_g53_label.set_alignment(0.0, 0.5) # self.actual_g53_label.set_line_wrap(False) # self.actual_g53_label.set_size_request(920, 24) # header_box.pack_start(self.actual_g53_label, False, False, 0) self.actual_wcs_label = gtk.Label("Current Active WCS: unavailable") self.actual_wcs_label.set_use_markup(True) self.actual_wcs_label.set_alignment(0.0, 0.5) self.actual_wcs_label.set_line_wrap(False) self.actual_wcs_label.set_size_request(780, 24) header_box.pack_start(self.actual_wcs_label, False, False, 0) main_hbox = gtk.HBox(False, 8) self.dialog.vbox.pack_start(main_hbox, False, False, 8) lower_box = gtk.HBox(False, 10) lower_box.set_border_width(8) self.dialog.vbox.pack_start(lower_box, False, False, 0) lower_left_box = gtk.VBox(False, 6) lower_box.pack_start(lower_left_box, False, False, 0) spindle_box = gtk.HBox(False, 6) lower_left_box.pack_start(spindle_box, False, False, 0) spindle_label = gtk.Label("Spindle RPM:") spindle_label.set_alignment(0.0, 0.5) spindle_box.pack_start(spindle_label, False, False, 0) self.spindle_entry = gtk.Entry() self.spindle_entry.set_width_chars(10) self.spindle_entry.set_text("2500") spindle_box.pack_start(self.spindle_entry, False, False, 0) spindle_button_box = gtk.HBox(False, 6) lower_left_box.pack_start(spindle_button_box, False, False, 0) spindle_cw_button = gtk.Button("CW") spindle_cw_button.set_size_request(70, 32) spindle_cw_button.connect("clicked", self.on_spindle_cw) spindle_button_box.pack_start(spindle_cw_button, False, False, 0) spindle_ccw_button = gtk.Button("CCW") spindle_ccw_button.set_size_request(70, 32) spindle_ccw_button.connect("clicked", self.on_spindle_ccw) spindle_button_box.pack_start(spindle_ccw_button, False, False, 0) spindle_stop_button = gtk.Button("Stop") spindle_stop_button.set_size_request(80, 32) spindle_stop_button.connect("clicked", self.on_spindle_stop) spindle_button_box.pack_start(spindle_stop_button, False, False, 0) block_mode_label = gtk.Label("Recovery block mode:") block_mode_label.set_alignment(0.0, 0.5) lower_left_box.pack_start(block_mode_label, False, False, 8) self.manual_g91_radio = gtk.RadioButton(None, "Incremental (G91)") self.manual_g90_radio = gtk.RadioButton(self.manual_g91_radio, "Absolute (G90)") self.manual_g91_radio.set_active(True) lower_left_box.pack_start(self.manual_g91_radio, False, False, 0) lower_left_box.pack_start(self.manual_g90_radio, False, False, 0) block_right_box = gtk.VBox(False, 4) lower_box.pack_start(block_right_box, False, False, 0) block_label = gtk.Label("Recovery block:") block_label.set_alignment(0.0, 0.5) block_right_box.pack_start(block_label, False, False, 0) self.block_buffer = gtk.TextBuffer() self.block_buffer.set_text("G1 F10 Z-5\nG1 F90 Z0") self.block_view = gtk.TextView(self.block_buffer) self.block_view.set_wrap_mode(gtk.WRAP_NONE) self.block_view.set_size_request(430, 130) block_scroll = gtk.ScrolledWindow() block_scroll.set_policy(gtk.POLICY_AUTOMATIC, gtk.POLICY_AUTOMATIC) block_scroll.add(self.block_view) block_right_box.pack_start(block_scroll, False, False, 0) pendant_grid = gtk.Table(5, 5, False) pendant_grid.set_row_spacings(5) pendant_grid.set_col_spacings(5) main_hbox.pack_start(pendant_grid, False, False, 8) pendant_feed_box = gtk.VBox(False, 2) pendant_feed_label = gtk.Label("Feedrate") pendant_feed_label.set_alignment(0.5, 0.5) pendant_feed_box.pack_start(pendant_feed_label, False, False, 0) self.pendant_feed_entry = gtk.Entry() self.pendant_feed_entry.set_width_chars(7) self.pendant_feed_entry.set_text("20") pendant_feed_box.pack_start(self.pendant_feed_entry, False, False, 0) pendant_grid.attach(pendant_feed_box, 0, 1, 0, 1) z_box = gtk.VBox(False, 5) main_hbox.pack_start(z_box, False, False, 4) self.attach_button(pendant_grid, "Move Y+ %.4f" % RECOVERY_LARGE_STEP, "YP1", 2, 3, 0, 1) self.attach_button(pendant_grid, "Move Y+ %.4f" % RECOVERY_SMALL_STEP, "YP01", 2, 3, 1, 2) self.attach_x_button(pendant_grid, "Move X- %.4f" % RECOVERY_LARGE_STEP, "XN1", 0, 1, 2, 3) self.attach_x_button(pendant_grid, "Move X- %.4f" % RECOVERY_SMALL_STEP, "XN01", 1, 2, 2, 3) return_box = gtk.VBox(False, 2) return_xy_button = gtk.Button("Return to XY") return_xy_button.set_size_request(140, 24) return_xy_button.connect("clicked", self.on_request, "RETURN_XY") return_box.pack_start(return_xy_button, True, True, 0) return_xyz_button = gtk.Button("Return to XYZ") return_xyz_button.set_size_request(140, 24) return_xyz_button.connect("clicked", self.on_request, "RETURN_XYZ") return_box.pack_start(return_xyz_button, True, True, 0) pendant_grid.attach(return_box, 2, 3, 2, 3) self.attach_x_button(pendant_grid, "Move X+ %.4f" % RECOVERY_SMALL_STEP, "XP01", 3, 4, 2, 3) self.attach_x_button(pendant_grid, "Move X+ %.4f" % RECOVERY_LARGE_STEP, "XP1", 4, 5, 2, 3) self.attach_button(pendant_grid, "Move Y- %.4f" % RECOVERY_SMALL_STEP, "YN01", 2, 3, 3, 4) self.attach_button(pendant_grid, "Move Y- %.4f" % RECOVERY_LARGE_STEP, "YN1", 2, 3, 4, 5) self.pack_button(z_box, "Go to G30", "G30") self.pack_button(z_box, "Move Z+ %.4f" % RECOVERY_LARGE_STEP, "ZP1") self.pack_button(z_box, "Move Z+ %.4f" % RECOVERY_SMALL_STEP, "ZP01") self.pack_button(z_box, "Move Z- %.4f" % RECOVERY_SMALL_STEP, "ZN01") self.pack_button(z_box, "Move Z- %.4f" % RECOVERY_LARGE_STEP, "ZN1") bottom_box = gtk.HBox(False, 8) bottom_box.set_border_width(8) self.dialog.vbox.pack_start(bottom_box, False, False, 0) bottom_box.pack_start(gtk.Label(""), True, True, 0) run_block_button = gtk.Button("Run Block") run_block_button.set_size_request(150, 40) run_block_button.connect("clicked", self.on_run_block) bottom_box.pack_start(run_block_button, False, False, 0) clear_block_button = gtk.Button("Clear Block") clear_block_button.set_size_request(150, 40) clear_block_button.connect("clicked", self.on_clear_block) bottom_box.pack_start(clear_block_button, False, False, 0) done_button = gtk.Button("DONE") done_button.set_size_request(300, 40) done_button.connect("clicked", self.on_done) bottom_box.pack_start(done_button, False, False, 0) bottom_box.pack_start(gtk.Label(""), True, True, 0) self._done_sent = False self.dialog.connect("delete-event", self.on_delete_event) self.dialog.connect("key-press-event", self.on_key_press_event) self.dialog.connect("destroy", self.on_destroy) def attach_button(self, table, label, request, left, right, top, bottom): button = gtk.Button(label) button.set_size_request(140, 50) self.connect_repeat_move_button(button, request) table.attach(button, left, right, top, bottom) return button def attach_x_button(self, table, label, request, left, right, top, bottom): button = gtk.Button(label) button.set_size_request(105, 50) self.connect_repeat_move_button(button, request) table.attach(button, left, right, top, bottom) return button def connect_repeat_move_button(self, button, request): button.connect("button-press-event", self.on_repeat_move_press, request) button.connect("button-release-event", self.on_repeat_move_release) button.connect("leave-notify-event", self.on_repeat_move_leave) def get_text_buffer_text(self, text_buffer): start_iter = text_buffer.get_start_iter() end_iter = text_buffer.get_end_iter() return text_buffer.get_text(start_iter, end_iter, True) def strip_recovery_block_line(self, line): text = line.strip() if not text: return "" if text.startswith(";"): return "" if text.startswith("("): return "" if ";" in text: text = text.split(";", 1)[0].strip() return text def line_is_spindle_command(self, text): upper = text.upper().strip() return ( upper.startswith("M3") or upper.startswith("M03") or upper.startswith("M4") or upper.startswith("M04") or upper.startswith("M5") or upper.startswith("M05") ) def validate_recovery_block_line(self, text, target, modal_mode): upper = text.upper().replace(",", " ") words = upper.split() if len(words) < 1: return target, modal_mode, "" if upper in ("G90", "G91"): if upper == "G90": return target, 90, "" return target, 91, "" if self.line_is_spindle_command(upper): m_word = words[0] if m_word in ("M5", "M05"): if len(words) != 1: return target, modal_mode, "M5 must be used by itself." return target, modal_mode, "" if m_word not in ("M3", "M03", "M4", "M04"): return target, modal_mode, "Unsupported M command: %s" % m_word saw_s = False for word in words[1:]: if len(word) < 2 or word[0] != "S": return target, modal_mode, "M3/M4 lines may only use an S word." rpm = float(word[1:]) if rpm <= 0.0: return target, modal_mode, "Spindle RPM must be positive." saw_s = True if not saw_s: return target, modal_mode, "M3/M4 requires an S word." return target, modal_mode, "" g_motion = None saw_axis = False for word in words: if len(word) < 2: return target, modal_mode, "Bad word: %s" % word letter = word[0] value_text = word[1:] if letter == "G": if value_text in ("0", "00"): g_motion = "G0" elif value_text in ("1", "01"): g_motion = "G1" elif value_text == "4": g_motion = "G4" else: return target, modal_mode, "Only G0, G1, G4, G90, and G91 are allowed." elif letter == "F": feed = float(value_text) if feed <= 0.0: return target, modal_mode, "Feedrate must be positive." elif letter == "P": if g_motion != "G4": return target, modal_mode, "P is only allowed with G4 dwell." dwell = float(value_text) if dwell < 0.0: return target, modal_mode, "G4 dwell cannot be negative." elif letter in ("X", "Y", "Z", "A"): value = float(value_text) saw_axis = True if modal_mode == 90: target[letter] = self.active_wcs_to_g53(letter, value) else: target[letter] = target[letter] + value else: return target, modal_mode, "Unsupported word: %s" % letter if g_motion is None: return target, modal_mode, "Each motion line must include G0, G1, or G4." if g_motion in ("G0", "G1") and saw_axis == False: return target, modal_mode, "G0/G1 lines must include at least one axis." return target, modal_mode, "" def validate_recovery_block(self, text, start_absolute_mode): current = self.get_current_g53_position() if current is None: return None, "Current G53 position is unavailable." target = dict(current) if start_absolute_mode: modal_mode = 90 else: modal_mode = 91 clean_lines = [] for raw_line in text.splitlines(): line = self.strip_recovery_block_line(raw_line) if not line: continue try: target, modal_mode, error = self.validate_recovery_block_line(line, target, modal_mode) except Exception as e: return None, "Bad line: %s\n\n%s" % (line, str(e)) if error: return None, "Bad line: %s\n\n%s" % (line, error) if not self.guard_target_g53(target, line): return None, "Move exceeds soft limits." clean_lines.append(line) if len(clean_lines) == 0: return None, "Recovery block is empty." return clean_lines, "" def write_recovery_block_ngc(self, clean_lines, start_absolute_mode): try: f = open(BLOCK_NGC_FILE, "w") try: f.write("o sub\n\n") if start_absolute_mode: f.write(" G90\n") else: f.write(" G91\n") for line in clean_lines: f.write(" %s\n" % line) f.write("\n G90\n") f.write("\n") f.write("o endsub\n\n") f.write("M2\n") finally: f.close() log("recovery block written | %d lines" % len(clean_lines)) return True except Exception as e: self.show_limit_warning("Recovery block write failed.\n\n%s" % str(e)) log("recovery block write failed | %s" % str(e)) return False def on_run_block(self, widget): text = self.get_text_buffer_text(self.block_buffer) start_absolute_mode = False try: start_absolute_mode = self.manual_g90_radio.get_active() except Exception: start_absolute_mode = False clean_lines, error = self.validate_recovery_block(text, start_absolute_mode) if error: self.show_limit_warning("Recovery block blocked.\n\n%s" % error) return if not self.write_recovery_block_ngc(clean_lines, start_absolute_mode): return write_request("RUN_BLOCK") def on_clear_block(self, widget): self.block_buffer.set_text("") def get_spindle_rpm_text(self): try: value = self.spindle_entry.get_text().strip() if value: return value except Exception: pass return "0" def on_spindle_cw(self, widget): write_request("SPINDLE_CW:" + self.get_spindle_rpm_text()) def on_spindle_ccw(self, widget): write_request("SPINDLE_CCW:" + self.get_spindle_rpm_text()) def on_spindle_stop(self, widget): write_request("SPINDLE_STOP") def pack_button(self, box, label, request): button = gtk.Button(label) button.set_size_request(150, 50) if request in ("ZP1", "ZP01", "ZN01", "ZN1"): self.connect_repeat_move_button(button, request) else: button.connect("clicked", self.on_request, request) box.pack_start(button, False, False, 2) return button def show(self): self.dialog.show_all() self.update_actual_position() gtk.timeout_add(250, self.poll_state) gtk.timeout_add(250, self.update_actual_position) def poll_state(self): if read_state() != "WAITING": self.dialog.destroy() return False return True def show_limit_warning(self, message): try: dialog = gtk.MessageDialog( self.dialog, gtk.DIALOG_MODAL | gtk.DIALOG_DESTROY_WITH_PARENT, gtk.MESSAGE_WARNING, gtk.BUTTONS_OK, message) dialog.set_title("Move Blocked") dialog.set_keep_above(True) dialog.set_position(gtk.WIN_POS_CENTER_ON_PARENT) dialog.run() dialog.destroy() except Exception as e: log("limit warning failed | %s" % str(e)) def parse_ini_limits_from_file(self, path): limits = {} try: if not os.path.isfile(path): return limits current_section = "" f = open(path, "r") try: for raw in f: line = raw.strip() if not line: continue if line.startswith("#") or line.startswith(";"): continue if line.startswith("[") and line.endswith("]"): current_section = line[1:-1].strip().upper() continue if "=" not in line: continue key, value = line.split("=", 1) key = key.strip().upper() value = value.strip().split("#", 1)[0].split(";", 1)[0].strip() axis = None if current_section in ("AXIS_X", "AXIS_0", "JOINT_0"): axis = "X" elif current_section in ("AXIS_Y", "AXIS_1", "JOINT_1"): axis = "Y" elif current_section in ("AXIS_Z", "AXIS_2", "JOINT_2"): axis = "Z" elif current_section in ("AXIS_A", "AXIS_3", "JOINT_3"): axis = "A" if axis is None: continue if axis not in limits: limits[axis] = {"min": None, "max": None} if key == "MIN_LIMIT": limits[axis]["min"] = float(value) elif key == "MAX_LIMIT": limits[axis]["max"] = float(value) finally: f.close() except Exception as e: log("soft limit parse failed | %s | %s" % (path, str(e))) return limits def load_soft_limits(self): merged = {} for path in INI_LIMIT_FILES: file_limits = self.parse_ini_limits_from_file(path) for axis in file_limits: if axis not in merged: merged[axis] = {"min": None, "max": None} if file_limits[axis].get("min") is not None: merged[axis]["min"] = file_limits[axis].get("min") if file_limits[axis].get("max") is not None: merged[axis]["max"] = file_limits[axis].get("max") log("soft limits loaded | %s" % str(merged)) return merged def parse_saved_g53_position(self, coords_text): try: match = re.search( r"G53\s+X\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s+Y\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s+Z\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))\s+A\s*([+-]?(?:\d+(?:\.\d*)?|\.\d+))", coords_text, re.IGNORECASE) if match: return { "X": float(match.group(1)), "Y": float(match.group(2)), "Z": float(match.group(3)), "A": float(match.group(4)),} except Exception as e: log("saved G53 parse failed | %s" % str(e)) return None def get_current_g53_position(self): if self.stat is None: return None try: self.stat.poll() return { "X": float(self.stat.actual_position[0]), "Y": float(self.stat.actual_position[1]), "Z": float(self.stat.actual_position[2]), "A": float(self.stat.actual_position[3]),} except Exception as e: log("current G53 read failed | %s" % str(e)) return None def get_current_offsets(self): try: self.stat.poll() try: g5x = self.stat.g5x_offset except Exception: g5x = [0.0, 0.0, 0.0, 0.0] try: g92 = self.stat.g92_offset except Exception: g92 = [0.0, 0.0, 0.0, 0.0] try: tool = self.stat.tool_offset except Exception: tool = [0.0, 0.0, 0.0, 0.0] return g5x, g92, tool except Exception as e: log("offset read failed | %s" % str(e)) return [0.0, 0.0, 0.0, 0.0], [0.0, 0.0, 0.0, 0.0], [0.0, 0.0, 0.0, 0.0] def active_wcs_to_g53(self, axis, value): g5x, g92, tool = self.get_current_offsets() if axis == "X": return value + float(g5x[0]) + float(g92[0]) if axis == "Y": return value + float(g5x[1]) + float(g92[1]) if axis == "Z": return value + float(g5x[2]) + float(g92[2]) + float(tool[2]) if axis == "A": return value + float(g5x[3]) + float(g92[3]) return value def target_exceeds_soft_limits(self, target_g53): for axis in ("X", "Y", "Z", "A"): if axis not in target_g53: continue if axis not in self.soft_limits: continue value = target_g53[axis] axis_limits = self.soft_limits[axis] min_limit = axis_limits.get("min") max_limit = axis_limits.get("max") if min_limit is not None and value < (min_limit - SOFT_LIMIT_MARGIN): return axis, value, min_limit, max_limit if max_limit is not None and value > (max_limit + SOFT_LIMIT_MARGIN): return axis, value, min_limit, max_limit return None def guard_target_g53(self, target_g53, source_text): if not target_g53: return True violation = self.target_exceeds_soft_limits(target_g53) if violation is None: return True axis, value, min_limit, max_limit = violation self.show_limit_warning( "Move blocked.\n\n" "%s would exceed the machine soft limits.\n\n" "Axis: %s\n" "Target G53: %.4f\n" "Min limit: %s\n" "Max limit: %s" % ( source_text, axis, value, "%.4f" % min_limit if min_limit is not None else "unknown", "%.4f" % max_limit if max_limit is not None else "unknown")) log("move blocked by soft limit | %s | %s %.4f" % (source_text, axis, value)) return False def guard_pendant_or_return_request(self, request): current = self.get_current_g53_position() if current is None: self.show_limit_warning("Move blocked.\n\nCurrent G53 position is unavailable, so soft-limit safety could not verify the move.") return False target = dict(current) move_delta = { "XN1": ("X", -RECOVERY_LARGE_STEP), "XN01": ("X", -RECOVERY_SMALL_STEP), "XP01": ("X", RECOVERY_SMALL_STEP), "XP1": ("X", RECOVERY_LARGE_STEP), "YN1": ("Y", -RECOVERY_LARGE_STEP), "YN01": ("Y", -RECOVERY_SMALL_STEP), "YP01": ("Y", RECOVERY_SMALL_STEP), "YP1": ("Y", RECOVERY_LARGE_STEP), "ZN1": ("Z", -RECOVERY_LARGE_STEP), "ZN01": ("Z", -RECOVERY_SMALL_STEP), "ZP01": ("Z", RECOVERY_SMALL_STEP), "ZP1": ("Z", RECOVERY_LARGE_STEP),} if request in move_delta: axis, delta = move_delta[request] target[axis] = target[axis] + delta return self.guard_target_g53(target, request) if request == "RETURN_XY": if self.saved_g53_position is None: self.show_limit_warning("Move blocked.\n\nSaved G53 return position is unavailable.") return False target["X"] = self.saved_g53_position["X"] target["Y"] = self.saved_g53_position["Y"] return self.guard_target_g53(target, request) if request == "RETURN_XYZ" or request == "RETURN": if self.saved_g53_position is None: self.show_limit_warning("Move blocked.\n\nSaved G53 return position is unavailable.") return False target["X"] = self.saved_g53_position["X"] target["Y"] = self.saved_g53_position["Y"] target["Z"] = self.saved_g53_position["Z"] target["A"] = self.saved_g53_position["A"] return self.guard_target_g53(target, request) return True def parse_manual_move_target_g53(self, text, absolute_mode): current = self.get_current_g53_position() if current is None: return None, "Current G53 position is unavailable." target = dict(current) try: clean = text.strip().upper() clean = clean.replace(",", " ") words = clean.split() if len(words) < 1: return None, "Manual move is empty." g_seen = False axis_seen = False for word in words: if len(word) < 2: return None, "Bad word in manual move: %s" % word letter = word[0] value_text = word[1:] if letter == "G": if value_text in ("0", "00", "1", "01"): g_seen = True else: return None, "Only G0/G1 manual moves can be soft-limit checked." elif letter == "F": pass elif letter in ("X", "Y", "Z", "A"): value = float(value_text) axis_seen = True if absolute_mode: target[letter] = self.active_wcs_to_g53(letter, value) else: target[letter] = target[letter] + value else: return None, "Unsupported manual move word: %s" % letter if not g_seen: return None, "Manual move must include G0 or G1." if not axis_seen: return None, "Manual move must include at least one axis." return target, "" except Exception as e: return None, "Manual move soft-limit check failed: %s" % str(e) def guard_manual_move(self, text, absolute_mode): target, error = self.parse_manual_move_target_g53(text, absolute_mode) if error: self.show_limit_warning("Move blocked.\n\n%s" % error) return False return self.guard_target_g53(target, text) def get_wcs_name(self, index_value): try: index_value = int(index_value) except Exception: index_value = 1 if index_value == 1: return "G54" if index_value == 2: return "G55" if index_value == 3: return "G56" if index_value == 4: return "G57" if index_value == 5: return "G58" if index_value == 6: return "G59" if index_value == 7: return "G59.1" if index_value == 8: return "G59.2" if index_value == 9: return "G59.3" return "G54" def update_actual_position(self): if self.stat is None: # self.actual_g53_label.set_text("Current G53: unavailable") self.actual_wcs_label.set_text("Current Active WCS: unavailable") return True try: self.stat.poll() x = float(self.stat.actual_position[0]) y = float(self.stat.actual_position[1]) z = float(self.stat.actual_position[2]) a = float(self.stat.actual_position[3]) wcs_name = self.get_wcs_name(self.stat.g5x_index) try: g5x = self.stat.g5x_offset except Exception: g5x = [0.0, 0.0, 0.0, 0.0] try: g92 = self.stat.g92_offset except Exception: g92 = [0.0, 0.0, 0.0, 0.0] try: tool = self.stat.tool_offset except Exception: tool = [0.0, 0.0, 0.0, 0.0] wx = x - float(g5x[0]) - float(g92[0]) wy = y - float(g5x[1]) - float(g92[1]) wz = z - float(g5x[2]) - float(g92[2]) - float(tool[2]) wa = a - float(g5x[3]) - float(g92[3]) # self.actual_g53_label.set_text( # "Current G53: G53 X%.4f Y%.4f Z%.4f A%.4f" % # (x, y, z, a) # ) self.actual_wcs_label.set_markup(" Currently: %s X% .4f Y% .4f Z% .4f A% .4f" % (wcs_name, wx, wy, wz, wa)) except Exception: self.actual_wcs_label.set_text("Current Active WCS: unavailable") return True def get_pendant_feed_text(self): try: value = self.pendant_feed_entry.get_text().strip() if value: return value except Exception: pass return "20" def is_pendant_move_request(self, request): return request in ( "XN1", "XN01", "XP01", "XP1", "YN1", "YN01", "YP01", "YP1", "ZN1", "ZN01", "ZP01", "ZP1", "RETURN_XY", "RETURN_XYZ", "RETURN") def send_move_request_once(self, request): if self.is_pendant_move_request(request): if not self.guard_pendant_or_return_request(request): return False write_request("MOVE:%s:%s" % (request, self.get_pendant_feed_text())) return True write_request(request) return True def on_repeat_move_press(self, widget, event, request): self._repeat_request = request self._repeat_active = True if not self.send_move_request_once(request): self.stop_repeat_move() return True gtk.timeout_add(PENDANT_REPEAT_MS, self.repeat_move_tick) return True def repeat_move_tick(self): if self._repeat_active == False: return False if not self._repeat_request: return False if not self.send_move_request_once(self._repeat_request): self.stop_repeat_move() return False return True def stop_repeat_move(self): self._repeat_active = False self._repeat_request = None def on_repeat_move_release(self, widget, event): self.stop_repeat_move() return True def on_repeat_move_leave(self, widget, event): self.stop_repeat_move() return False def on_request(self, widget, request): self.send_move_request_once(request) def send_done_once(self): if self._done_sent: return self._done_sent = True write_request("DONE") def on_done(self, widget): self.stop_repeat_move() self.send_done_once() self.dialog.destroy() def on_delete_event(self, widget, event): self.stop_repeat_move() self.send_done_once() return False def on_key_press_event(self, widget, event): key_name = gtk.gdk.keyval_name(event.keyval) if key_name in ("Escape", "F4"): self.stop_repeat_move() self.send_done_once() self.dialog.destroy() return True return False def on_destroy(self, widget): gtk.main_quit() def main(): lock_fp = acquire_lock() if lock_fp is None: return 0 if read_state() != "WAITING": return 0 log("dialog opened") dialog = BeagleRecoveryDialog() dialog.show() gtk.main() log("dialog closed") return 0 if __name__ == "__main__": sys.exit(main())''' class TTFilePatchLock(object): def __init__(self, owner, error_handler, lock_path, timeout_seconds): self.owner = owner self.error_handler = error_handler self.lock_path = lock_path self.timeout_seconds = timeout_seconds self.fp = None def write(self, message): try: self.error_handler.write("[%s] %s" % (self.owner, message), constants.ALARM_LEVEL_QUIET) except: pass def __enter__(self): start_time = time.time() self.fp = open(self.lock_path, "a+") self.write("Waiting for patch lock: %s" % self.lock_path) while True: try: fcntl.flock(self.fp.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB) self.write("Patch lock acquired: %s" % self.lock_path) return self except IOError: elapsed = time.time() - start_time if elapsed >= self.timeout_seconds: raise RuntimeError("Timed out waiting for patch lock: %s" % self.lock_path) time.sleep(0.1) def __exit__(self, exc_type, exc_value, traceback_obj): try: if self.fp: fcntl.flock(self.fp.fileno(), fcntl.LOCK_UN) self.write("Patch lock released: %s" % self.lock_path) finally: try: if self.fp: self.fp.close() except: pass return False class UserPlugin(plugin): def __init__(self): plugin.__init__(self, "%s %s" % (SCRIPT_NAME, CURRENT_VER)) dev_machine_found = os.path.exists(DEV_MACHINE_FLAG) if dev_machine_found: plugin_enabled = DEV_MACHINE else: plugin_enabled = ENABLED if plugin_enabled: glib.timeout_add(3000, self.try_patch) return if dev_machine_found: self.error_handler.write("[%s] Dev machine found. Plugin loaded, but disabled by DEV_MACHINE." % SCRIPT_NAME, constants.ALARM_LEVEL_QUIET) else: self.error_handler.write("[%s] Plugin loaded, but disabled." % SCRIPT_NAME, constants.ALARM_LEVEL_QUIET) def prerequisites_ok(self): if os.path.isfile(GLADE_CUSTOM_TAB_PLUGIN): return True message = ( "Recovery Mode Installer cannot install because the required custom tab plugin was not found.\n\n" "Missing file:\n" "%s\n\n" "Download/install glade_custom_tab_plugin.py first, then reboot PathPilot or reload plugins." ) % GLADE_CUSTOM_TAB_PLUGIN try: self.error_handler.write("[%s] %s" % (SCRIPT_NAME, message), constants.ALARM_LEVEL_HIGH) except Exception: pass try: dialog = gtk.MessageDialog( None, gtk.DIALOG_MODAL, gtk.MESSAGE_WARNING, gtk.BUTTONS_OK, message) dialog.set_title("Recovery Mode Installer") dialog.set_keep_above(True) dialog.run() dialog.destroy() except Exception: pass return False def debug(self, message): try: self.error_handler.write("[%s] %s" % (SCRIPT_NAME, message), constants.ALARM_LEVEL_QUIET) except: pass def safe_script_name(self): safe_name = SCRIPT_NAME.lower() safe_name = safe_name.replace(" ", "_") safe_name = re.sub(r"[^a-z0-9_]+", "_", safe_name) return safe_name def read_file(self, path): self.debug("Reading file: %s" % path) f = open(path, "r") try: return f.read() finally: f.close() def write_file(self, path, content): self.debug("Writing file: %s" % path) f = open(path, "w") try: f.write(content) finally: f.close() def make_chronological_backup(self, target_path): timestamp = time.strftime("%Y%m%d_%H%M%S") safe_name = self.safe_script_name() backup_path = "%s.%s.%s.bak" % (target_path, timestamp, safe_name) counter = 1 final_path = backup_path while os.path.exists(final_path): final_path = "%s.%03d" % (backup_path, counter) counter += 1 shutil.copy2(target_path, final_path) self.debug("Backup created: %s" % final_path) return final_path def staged_new_path(self, target_path): return "%s.%s.%d.new" % (target_path, self.safe_script_name(), os.getpid()) def staged_old_path(self, target_path): return "%s.%s.%d.old" % (target_path, self.safe_script_name(), os.getpid()) def write_staged_file(self, target_path, content): staged_path = self.staged_new_path(target_path) self.write_file(staged_path, content) return staged_path def cleanup_staged_files(self, target_paths): for path in target_paths: for staged_path in (self.staged_new_path(path), self.staged_old_path(path)): try: if os.path.exists(staged_path): os.remove(staged_path) except: pass def restore_from_old_files(self, target_paths): for path in target_paths: old_path = self.staged_old_path(path) try: if os.path.exists(old_path): if os.path.exists(path): os.remove(path) os.rename(old_path, path) self.debug("Rollback restored: %s" % path) except Exception as e: self.error_handler.write("[%s] ERROR: Rollback failed for %s: %s" % (SCRIPT_NAME, path, str(e)), constants.ALARM_LEVEL_HIGH) def swap_staged_files_into_place(self, target_paths, staged_paths): try: for path in target_paths: new_path = staged_paths[path] old_path = self.staged_old_path(path) if os.path.exists(old_path): os.remove(old_path) os.rename(path, old_path) os.rename(new_path, path) for path in target_paths: old_path = self.staged_old_path(path) if os.path.exists(old_path): os.remove(old_path) except Exception: self.error_handler.write("[%s] ERROR: Swap failed. Attempting rollback." % SCRIPT_NAME, constants.ALARM_LEVEL_HIGH) self.restore_from_old_files(target_paths) raise def remove_marked_block(self, content, start_marker, end_marker): lines = content.splitlines(True) output = [] skipping = False for line in lines: if start_marker in line: skipping = True continue if skipping: if end_marker in line: skipping = False continue output.append(line) return "".join(output) def replace_marked_block_if_present(self, content, start_marker, end_marker, replacement): start_index = content.find(start_marker) if start_index < 0: return content, False end_index = content.find(end_marker, start_index) if end_index < 0: raise RuntimeError("Found start marker but not end marker: %s" % start_marker) line_start = content.rfind("\n", 0, start_index) + 1 line_end = content.find("\n", end_index) if line_end < 0: line_end = len(content) else: line_end += 1 if not replacement.endswith("\n"): replacement += "\n" new_content = content[:line_start] + replacement + content[line_end:] return new_content, True def insert_after_line(self, content, needle, block): lines = content.splitlines(True) for i, line in enumerate(lines): if needle in line: if not block.endswith("\n"): block += "\n" lines.insert(i + 1, block) return "".join(lines) raise RuntimeError("Could not find insertion line: %s" % needle) def insert_before_line(self, content, needle, block): lines = content.splitlines(True) for i, line in enumerate(lines): if needle in line: if not block.endswith("\n"): block += "\n" lines.insert(i, block) return "".join(lines) raise RuntimeError("Could not find insertion line: %s" % needle) def patch_ini_content(self, content): original = content content = self.remove_marked_block(content, INI_PATCH_START, INI_PATCH_END) content = self.insert_after_line( content, "REMAP = M81 modalgroup=7 py=restore_user_modals_M81", BEAGLE_INI_BLOCK) return content, content != original def patch_remap_content(self, content): original = content if REMAP_PATCH_START not in BEAGLE_REMAP_BLOCK or REMAP_PATCH_END not in BEAGLE_REMAP_BLOCK: raise RuntimeError("Embedded remap block is missing BEAGLE markers.") content = self.remove_marked_block(content, REMAP_PATCH_START, REMAP_PATCH_END) content = self.insert_before_line( content, "def restore_user_modals_M81", BEAGLE_REMAP_BLOCK) return content, content != original def find_matching_tool_to_do_block(self, lines, start_index): depth = 0 for i in range(start_index, len(lines)): line = lines[i] if "o if" in line: depth += 1 if "o endif" in line: depth -= 1 if depth == 0: return i return -1 def patch_tool_change_content(self, content): original = content replacement = ( " %s\n" " o if [#<_new_tool> NE #<_old_tool>]\n" " M5\n" " o call [12] [#<_new_tool>]\n" " M66 P63 L0\n" " o call [1]\n" " M66 P63 L0\n" " o9020 if [#<_beagle_recovery_request_code> EQ 9]\n" " o call [12] [#<_new_tool>]\n" " o9020 endif\n" " o endif\n" " %s\n" ) % (TOOL_CHANGE_PATCH_START, TOOL_CHANGE_PATCH_END) content, replaced_existing = self.replace_marked_block_if_present( content, TOOL_CHANGE_PATCH_START, TOOL_CHANGE_PATCH_END, replacement) if replaced_existing: return content, content != original lines = content.splitlines(True) start_index = -1 for i, line in enumerate(lines): if "o if [#<_new_tool> NE #<_old_tool>]" in line: start_index = i break if start_index < 0: raise RuntimeError("Could not find o block in tormach_tool_change.ngc") end_index = self.find_matching_tool_to_do_block(lines, start_index) if end_index < 0: raise RuntimeError("Could not find matching o endif in tormach_tool_change.ngc") content = "".join(lines[:start_index] + [replacement] + lines[end_index + 1:]) return content, content != original def verify_content_has_marker(self, path, content, marker): if marker not in content: raise RuntimeError("Staged content for %s does not contain marker: %s" % (path, marker)) def install_or_update_file(self, target_path, payload, executable=False): if os.path.exists(target_path): existing = self.read_file(target_path) if existing == payload: self.debug("Already current: %s" % target_path) return False, None backup_path = None if os.path.exists(target_path): backup_path = self.make_chronological_backup(target_path) self.write_file(target_path, payload) if executable: mode = os.stat(target_path).st_mode os.chmod(target_path, mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH) self.debug("Installed file: %s" % target_path) return True, backup_path def initialize_runtime_files(self): if not os.path.isdir(GCODE_PYTHON_DIR): os.makedirs(GCODE_PYTHON_DIR) self.write_file(REQUEST_FILE, "") self.write_file(STATE_FILE, "IDLE\n") self.write_file(COORDS_FILE, "G53 position unavailable\n") if not os.path.exists(DEBUG_LOG): self.write_file(DEBUG_LOG, "") try: if os.path.exists(LOCK_FILE): os.remove(LOCK_FILE) except: pass def patch_three_existing_files(self): target_paths = [TORMACH_MILL_BASE_INI, REMAP_PY, TOOL_CHANGE_NGC] self.cleanup_staged_files(target_paths) ini_content = self.read_file(TORMACH_MILL_BASE_INI) remap_content = self.read_file(REMAP_PY) tool_change_content = self.read_file(TOOL_CHANGE_NGC) patched_ini, ini_changed = self.patch_ini_content(ini_content) patched_remap, remap_changed = self.patch_remap_content(remap_content) patched_tool_change, tool_change_changed = self.patch_tool_change_content(tool_change_content) staged_paths = {} changed_paths = [] backup_paths = [] if ini_changed: backup_paths.append(self.make_chronological_backup(TORMACH_MILL_BASE_INI)) staged_paths[TORMACH_MILL_BASE_INI] = self.write_staged_file(TORMACH_MILL_BASE_INI, patched_ini) self.verify_content_has_marker(TORMACH_MILL_BASE_INI, patched_ini, INI_PATCH_START) changed_paths.append(TORMACH_MILL_BASE_INI) if remap_changed: backup_paths.append(self.make_chronological_backup(REMAP_PY)) staged_paths[REMAP_PY] = self.write_staged_file(REMAP_PY, patched_remap) self.verify_content_has_marker(REMAP_PY, patched_remap, REMAP_PATCH_START) changed_paths.append(REMAP_PY) if tool_change_changed: backup_paths.append(self.make_chronological_backup(TOOL_CHANGE_NGC)) staged_paths[TOOL_CHANGE_NGC] = self.write_staged_file(TOOL_CHANGE_NGC, patched_tool_change) self.verify_content_has_marker(TOOL_CHANGE_NGC, patched_tool_change, TOOL_CHANGE_PATCH_START) changed_paths.append(TOOL_CHANGE_NGC) if changed_paths: self.swap_staged_files_into_place(changed_paths, staged_paths) return changed_paths, backup_paths def show_success_dialog(self, changed_any): if changed_any: message = ( "Beagle Recovery was installed or updated.\n\n" "Restart PathPilot before testing M6 Recovery or M252 Checkpoints.") else: message = "Beagle Recovery is already installed. No changes were needed." dialog = gtk.MessageDialog( None, gtk.DIALOG_DESTROY_WITH_PARENT, gtk.MESSAGE_INFO, gtk.BUTTONS_OK, message) dialog.set_title(SCRIPT_NAME) dialog.set_keep_above(True) dialog.set_modal(False) def on_response(widget, response_id): widget.destroy() dialog.connect("response", on_response) dialog.show_all() def try_patch(self): if not self.prerequisites_ok(): return False try: if not os.path.isfile(TORMACH_MILL_BASE_INI): raise RuntimeError("Missing file: %s" % TORMACH_MILL_BASE_INI) if not os.path.isfile(REMAP_PY): raise RuntimeError("Missing file: %s" % REMAP_PY) if not os.path.isfile(TOOL_CHANGE_NGC): raise RuntimeError("Missing file: %s" % TOOL_CHANGE_NGC) if not os.path.isdir(NC_SUBS_DIR): raise RuntimeError("Missing directory: %s" % NC_SUBS_DIR) if not os.path.isdir(GCODE_PYTHON_DIR): os.makedirs(GCODE_PYTHON_DIR) changed_any = False backup_paths = [] with TTFilePatchLock(SCRIPT_NAME, self.error_handler, PATCH_LOCK_PATH, PATCH_LOCK_TIMEOUT_SECONDS): changed_paths, patch_backups = self.patch_three_existing_files() backup_paths.extend(patch_backups) checkpoint_changed, checkpoint_backup = self.install_or_update_file(CHECKPOINT_NGC_PATH, BEAGLE_CHECKPOINT_NGC, executable=False) dialog_changed, dialog_backup = self.install_or_update_file(DIALOG_PATH, BEAGLE_DIALOG_PY, executable=True) if checkpoint_backup: backup_paths.append(checkpoint_backup) if dialog_backup: backup_paths.append(dialog_backup) if changed_paths or checkpoint_changed or dialog_changed: self.initialize_runtime_files() changed_any = bool(changed_paths or checkpoint_changed or dialog_changed) if backup_paths: for backup_path in backup_paths: self.error_handler.write("[%s] Backup created: %s" % (SCRIPT_NAME, backup_path), constants.ALARM_LEVEL_MEDIUM) if changed_any: self.error_handler.write("[%s] Recovery Mode Plugin installed or updated." % SCRIPT_NAME, constants.ALARM_LEVEL_MEDIUM) self.error_handler.write("[%s] Restart PathPilot before testing." % SCRIPT_NAME, constants.ALARM_LEVEL_MEDIUM) else: self.error_handler.write("[%s] Already installed. No changes made." % SCRIPT_NAME, constants.ALARM_LEVEL_QUIET) if changed_any: glib.idle_add(self.show_success_dialog, changed_any) except Exception as e: self.error_handler.write("[%s] Error: %s" % (SCRIPT_NAME, str(e)), constants.ALARM_LEVEL_LOW) return False DESCRIPTION_LONG = """Installs PathPilot Recovery Mode Suite."""