# coding: utf-8
# python 2 only
# 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
#############################################
## ##
## Duplicate PathPilot Preventer v0.96 ##
## www.tormachtips.com ##
## ##
#############################################
# 0.96 - Added shared file lock, chronological backups, atomic writes, executable-bit restore, shell syntax validation, and verbose patch debug logging. - 5/03/2026
# 0.95 - Public beta. - 4/2/26
import os
import re
import time
import fcntl
import shutil
import stat
import subprocess
import tempfile
import glib
import constants
from ui_hooks import plugin
CURRENT_VER = "0.96"
SCRIPT_NAME = "Duplicate PathPilot Preventer"
DESCRIPTION = "Tries to prevent PathPilot from running on top of itself. Useful if you open PathPilot from a Terminal window."
ENABLED = 1
DEV_MACHINE = 0
DEV_MACHINE_FLAG = "/home/operator/gcode/python/dev_machine.txt"
TARGET_FILE = "/home/operator/operator_login"
MARKER = "# BEAGLE: DUPLICATE PATHPILOT PREVENTER PATCH INSTALLED"
PATCH_START = "# BEAGLE: DUPLICATE PATHPILOT PREVENTER PATCH START"
PATCH_END = "# BEAGLE: DUPLICATE PATHPILOT PREVENTER PATCH END"
PATCH_LOCK_PATH = "/tmp/tt_pathpilot_file_patch.lock"
PATCH_LOCK_TIMEOUT_SECONDS = 10
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.write("Opening patch lock file: %s" % self.lock_path)
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:
self.write("Patch lock timeout after %.1f seconds: %s" % (elapsed, self.lock_path))
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:
self.write("Releasing patch lock: %s" % self.lock_path)
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()
self.write("Closed patch lock file: %s" % self.lock_path)
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)
self.error_handler.write("[%s] To enable, open script, find ENABLED = 0 and change to ENABLED = 1." % SCRIPT_NAME, constants.ALARM_LEVEL_QUIET)
def debug(self, message):
self.error_handler.write("[%s] %s" % (SCRIPT_NAME, message), constants.ALARM_LEVEL_QUIET)
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)
with open(path, "r") as f:
content = f.read()
self.debug("Read %d bytes from file: %s" % (len(content), path))
return content
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
self.debug("Creating chronological backup.")
self.debug("Backup source: %s" % target_path)
self.debug("Backup target: %s" % final_path)
shutil.copy2(target_path, final_path)
self.debug("Backup complete: %s" % final_path)
return final_path
def chmod_executable(self, target_path):
current_mode = os.stat(target_path).st_mode
executable_mode = current_mode | stat.S_IXUSR | stat.S_IXGRP | stat.S_IXOTH
self.debug("Restoring executable bit.")
self.debug("chmod target: %s" % target_path)
self.debug("chmod mode before: %o" % current_mode)
self.debug("chmod mode after: %o" % executable_mode)
os.chmod(target_path, executable_mode)
if not os.access(target_path, os.X_OK):
raise RuntimeError("File is not executable after chmod: %s" % target_path)
self.debug("Executable permission confirmed: %s" % target_path)
def atomic_write_file(self, target_path, content):
original_stat = os.stat(target_path)
temp_path = "%s.%s.%d.tmp" % (
target_path,
self.safe_script_name(),
os.getpid())
self.debug("Writing temporary patched file: %s" % temp_path)
with open(temp_path, "w") as f:
f.write(content)
self.debug("Temporary patched file written: %s" % temp_path)
try:
os.chmod(temp_path, original_stat.st_mode)
self.debug("Copied original file mode to temporary file: %o" % original_stat.st_mode)
except Exception as e:
self.debug("Could not copy original mode to temporary file: %s" % str(e))
try:
os.chown(temp_path, original_stat.st_uid, original_stat.st_gid)
self.debug("Copied original owner/group to temporary file.")
except Exception as e:
self.debug("Could not copy original owner/group to temporary file: %s" % str(e))
self.debug("Renaming temporary file over target.")
self.debug("Rename source: %s" % temp_path)
self.debug("Rename target: %s" % target_path)
os.rename(temp_path, target_path)
self.debug("Atomic rename complete: %s" % target_path)
self.chmod_executable(target_path)
def validate_shell_content(self, target_path, content):
temp_fd = None
temp_path = ""
try:
temp_fd, temp_path = tempfile.mkstemp(prefix="tt_shell_check_", suffix=".sh")
os.write(temp_fd, content)
os.close(temp_fd)
temp_fd = None
self.debug("Validating shell syntax with bash -n: %s" % temp_path)
p = subprocess.Popen(
["bash", "-n", temp_path],
stdout=subprocess.PIPE,
stderr=subprocess.PIPE)
stdout_data, stderr_data = p.communicate()
if p.returncode != 0:
raise RuntimeError("Patched shell file failed syntax validation: %s" % stderr_data.strip())
self.debug("Shell syntax validation passed for: %s" % target_path)
finally:
try:
if temp_fd is not None:
os.close(temp_fd)
except:
pass
try:
if temp_path and os.path.exists(temp_path):
os.remove(temp_path)
except:
pass
def patch_content(self, content):
if MARKER in content:
self.debug("Patch marker already present.")
return content, False
expected_header = (
"#!/bin/bash\n"
"# coding=utf-8\n"
"#-----------------------------------------------------------------------\n"
"# Copyright © 2014-2019 Tormach® Inc. All rights reserved.\n"
"# License: GPL Version 2\n"
"#-----------------------------------------------------------------------\n")
mod_block = (
"\n"
+ MARKER + "\n"
+ PATCH_START + "\n"
+ "if pgrep -f \"tormach_mill_ui.py\" > /dev/null; then\n"
+ " echo\n"
+ " echo\n"
+ " echo \"PathPilot is already running, exiting.\"\n"
+ " echo\n"
+ " echo\n"
+ " exit 0\n"
+ "fi\n"
+ PATCH_END + "\n"
+ "\n")
original_content = content
if content.startswith(expected_header):
self.debug("Expected operator_login header found; inserting patch after header.")
content = expected_header + mod_block + content[len(expected_header):]
else:
self.debug("Expected operator_login header not found; inserting patch at top after shebang if possible.")
if content.startswith("#!"):
first_newline = content.find("\n")
if first_newline >= 0:
content = content[:first_newline + 1] + mod_block + content[first_newline + 1:]
else:
content = content + mod_block
else:
content = mod_block.lstrip("\n") + content
if content == original_content:
raise RuntimeError("Patch produced no content changes.")
return content, True
def try_patch(self):
backup_path = ""
try:
target_path = TARGET_FILE
self.debug("Target file: %s" % target_path)
self.debug("Patch marker: %s" % MARKER)
with TTFilePatchLock(SCRIPT_NAME, self.error_handler, PATCH_LOCK_PATH, PATCH_LOCK_TIMEOUT_SECONDS):
self.debug("Inside locked patch section.")
if not os.path.exists(target_path):
self.error_handler.write("[%s] File not found: %s" % (SCRIPT_NAME, target_path), constants.ALARM_LEVEL_LOW)
return False
content = self.read_file(target_path)
patched_content, changed = self.patch_content(content)
if not changed:
self.error_handler.write("[%s] Already patched. No changes made." % SCRIPT_NAME, constants.ALARM_LEVEL_QUIET)
self.debug("Marker found; exiting without backup or write.")
if not os.access(target_path, os.X_OK):
self.debug("Patch marker exists, but file is not executable. Restoring executable bit.")
self.chmod_executable(target_path)
return False
self.validate_shell_content(target_path, patched_content)
self.debug("Patch content generated.")
self.debug("Original byte count: %d" % len(content))
self.debug("Patched byte count: %d" % len(patched_content))
backup_path = self.make_chronological_backup(target_path)
self.atomic_write_file(target_path, patched_content)
self.debug("Leaving locked patch section after successful write.")
self.error_handler.write(" ", constants.ALARM_LEVEL_QUIET)
self.error_handler.write("[%s] Patch applied to: %s" % (SCRIPT_NAME, target_path), constants.ALARM_LEVEL_MEDIUM)
self.error_handler.write("[%s] Backup created: %s" % (SCRIPT_NAME, backup_path), constants.ALARM_LEVEL_MEDIUM)
self.error_handler.write("[%s] Executable permission confirmed on patched file." % SCRIPT_NAME, constants.ALARM_LEVEL_MEDIUM)
self.error_handler.write("[%s] Restart PathPilot to ensure the change is active." % SCRIPT_NAME, constants.ALARM_LEVEL_MEDIUM)
self.error_handler.write(" ", constants.ALARM_LEVEL_QUIET)
except Exception as e:
self.error_handler.write("[%s] Error: %s" % (SCRIPT_NAME, str(e)), constants.ALARM_LEVEL_LOW)
return False
DESCRIPTION_LONG = """If you open
PathPilot in a terminal, and are always tinkering around, you may often try
to open PathPilot twice.
PathPilot hates that and will crash.
This attempts to fix double-runs.
Plugin is
available on this regularly-updated page.
"""
"""