import os
import sys
import linuxcnc
import tempfile
from time import perf_counter
from PySide6.QtCore import Qt, QTimer
# Set up logging
from qtpyvcp.utilities import logger
LOG = logger.getLogger(__name__)
from qtpyvcp.utilities.info import Info
from qtpyvcp.utilities.qt_safety import safe_qt_callback
from qtpyvcp.plugins import getPlugin
from qtpyvcp.utilities.load_perf_summary import PROGRAM_LOAD_PERF_SUMMARY
IN_DESIGNER = os.getenv('DESIGNER', False)
if not IN_DESIGNER:
STATUS = getPlugin('status')
STAT = STATUS.stat
INFO = Info()
CMD = linuxcnc.command()
_PRECLEAR_BLANK_FILE = None
from qtpyvcp.actions.base_actions import setTaskMode
#==============================================================================
# Program actions
#==============================================================================
def load(fname, add_to_recents=True, isreload=False, _skip_preclear=False):
if not fname:
# load a blank file. Maybe should load [DISPLAY] OPEN_FILE
clear()
return
if not _skip_preclear:
preclear_fname = _preclear_before_load(fname)
if preclear_fname:
_queue_load_after_preclear_signal(
target_fname=fname,
preclear_fname=preclear_fname,
add_to_recents=add_to_recents,
isreload=isreload,
)
return
PROGRAM_LOAD_PERF_SUMMARY.start(fname)
abs_fname = os.path.abspath(fname) if fname else None
file_event_seen = {'done': False}
def _disconnect_file_signal():
try:
STATUS.file.signal.disconnect(_on_linuxcnc_file_loaded)
except Exception:
pass
def _on_linuxcnc_file_loaded(loaded_fname):
try:
loaded_abs = os.path.abspath(str(loaded_fname)) if loaded_fname else None
except Exception:
loaded_abs = None
if abs_fname and loaded_abs == abs_fname and not file_event_seen['done']:
file_event_seen['done'] = True
PROGRAM_LOAD_PERF_SUMMARY.mark_linuxcnc_file_loaded_event(fname)
_disconnect_file_signal()
try:
STATUS.file.signal.connect(_on_linuxcnc_file_loaded)
except Exception:
pass
#setTaskMode(linuxcnc.MODE_AUTO)
if not isreload:
STATUS.addLock()
filter_prog = INFO.getFilterProgram(fname)
interp_start = perf_counter()
PROGRAM_LOAD_PERF_SUMMARY.mark_phase(fname, phase='linuxcnc-open-wait-start', percent=10)
try:
if not filter_prog:
LOG.debug(f"Loading NC program: {fname}")
CMD.program_open(fname.encode('utf-8'))
CMD.wait_complete()
else:
LOG.debug(f"Loading file with filter program: {fname}")
openFilterProgram(fname, filter_prog)
try:
STAT.poll()
stat_file = os.path.abspath(str(STAT.file)) if STAT.file else None
except Exception:
stat_file = None
if abs_fname and stat_file == abs_fname and not file_event_seen['done']:
file_event_seen['done'] = True
PROGRAM_LOAD_PERF_SUMMARY.mark_linuxcnc_file_loaded_event(fname)
interp_ms = (perf_counter() - interp_start) * 1000.0
PROGRAM_LOAD_PERF_SUMMARY.add_linuxcnc_interp_time(fname, interp_ms=interp_ms)
finally:
if not file_event_seen['done']:
QTimer.singleShot(500, _disconnect_file_signal)
if add_to_recents:
addToRecents(fname)
STATUS.removeLock()
load.ok = lambda *args, **kwargs: True
load.bindOk = lambda *args, **kwargs: True
def _get_preclear_blank_file():
global _PRECLEAR_BLANK_FILE
if _PRECLEAR_BLANK_FILE and os.path.exists(_PRECLEAR_BLANK_FILE):
return _PRECLEAR_BLANK_FILE
fd, path = tempfile.mkstemp(prefix="qtpyvcp_preclear_", suffix=".ngc")
try:
with os.fdopen(fd, 'w') as fp:
fp.write("(QtPyVCP pre-clear)\nM30\n")
except Exception:
try:
os.close(fd)
except Exception:
pass
raise
_PRECLEAR_BLANK_FILE = path
return _PRECLEAR_BLANK_FILE
def _queue_load_after_preclear_signal(*, target_fname, preclear_fname, add_to_recents, isreload):
target_abs = os.path.abspath(target_fname) if target_fname else None
preclear_abs = os.path.abspath(preclear_fname) if preclear_fname else None
fired = {'done': False}
def _trigger_target_load(reason='signal'):
if fired['done']:
return
fired['done'] = True
try:
STATUS.file.signal.disconnect(_on_preclear_file_loaded)
except Exception:
pass
LOG.debug("Proceeding with target load after pre-clear (%s): %s", reason, target_abs)
# Brief dwell lets the cleared state paint before loading the target.
QTimer.singleShot(
75,
lambda: load(
target_fname,
add_to_recents=add_to_recents,
isreload=isreload,
_skip_preclear=True,
),
)
def _on_preclear_file_loaded(loaded_fname):
try:
loaded_abs = os.path.abspath(str(loaded_fname)) if loaded_fname else None
except Exception:
loaded_abs = None
if preclear_abs and loaded_abs == preclear_abs:
LOG.debug("Pre-clear completion hook matched: %s", preclear_abs)
_trigger_target_load(reason='hook')
try:
STATUS.file.signal.connect(_on_preclear_file_loaded)
except Exception:
_trigger_target_load(reason='no-hook')
return
# Fallback in case the file signal is delayed/missed.
def _timeout_fallback():
if not fired['done']:
LOG.debug("Pre-clear hook timeout; continuing with target load after 700ms")
_trigger_target_load(reason='timeout')
QTimer.singleShot(700, _timeout_fallback)
def _preclear_before_load(target_fname):
try:
target_abs = os.path.abspath(target_fname) if target_fname else None
STAT.poll()
current_abs = os.path.abspath(str(STAT.file)) if STAT.file else None
# Skip pre-clear when no current file or reloading the same file.
if not target_abs or not current_abs or current_abs == target_abs:
return None
blank_file = _get_preclear_blank_file()
LOG.debug(
"Pre-clearing current program before load: current=%s target=%s blank=%s",
current_abs,
target_abs,
blank_file,
)
CMD.program_open(blank_file.encode('utf-8'))
CMD.wait_complete()
return blank_file
except Exception:
# Continue with normal load even if pre-clear fails.
LOG.warning("Pre-clear before program load failed; continuing", exc_info=True)
return None
[docs]
def reload():
"""Reload the currently loaded NC program
ActionButton syntax::
program.reload
"""
stat = linuxcnc.stat()
stat.poll()
fname = stat.file
if os.path.exists(fname):
load(stat.file, add_to_recents=False, isreload=True)
reload.ok = lambda *args, **kwargs: True
reload.bindOk = lambda *args, **kwargs: True
[docs]
def clear():
"""Clear the loaded NC program
ActionButton syntax::
program.clear
"""
_, blankfile = tempfile.mkstemp(prefix="new_program_", suffix=".ngc")
with open(blankfile, 'w') as fp:
fp.write("(New Program)\n\n\nM30")
load(blankfile, add_to_recents=False)
clear.ok = lambda *args, **kwargs: True
clear.bindOk = lambda *args, **kwargs: True
def addToRecents(fname):
files = STATUS.recent_files.getValue()
if fname in files:
files.remove(fname)
files.insert(0, fname)
STATUS.recent_files.setValue(files[:STATUS.max_recent_files])
# -------------------------------------------------------------------------
# program RUN action
# -------------------------------------------------------------------------
[docs]
def run(start_line=0):
"""Runs the loaded program, optionally starting from a specific line.
ActionButton syntax::
program.run
program.run:line
Args:
start_line (int, optional) : The line to start program from. Defaults to 0.
"""
interp_paused = (STAT.interp_state == linuxcnc.INTERP_PAUSED)
mdi_exec = STAT.state == linuxcnc.RCS_EXEC and STAT.task_mode == linuxcnc.MODE_MDI
LOG.debug(
"program.run requested: start_line=%s state=%s mode=%s interp=%s paused=%s feed_hold=%s",
start_line,
STAT.state,
STAT.task_mode,
STAT.interp_state,
STAT.paused,
STAT.feed_hold_enabled,
)
# MDI exception path: cycle-start is often used by operators as "resume" while
# running queued MDI lines; in this case prefer clearing feed hold over AUTO start.
if STAT.task_mode == linuxcnc.MODE_MDI and STAT.feed_hold_enabled:
LOG.debug("program.run: clearing feed hold in MDI mode")
CMD.set_feed_hold(0)
if mdi_exec and (STAT.paused or interp_paused):
CMD.auto(linuxcnc.AUTO_RESUME)
return
if mdi_exec and (STAT.paused or interp_paused):
LOG.debug("program.run: attempting MDI resume path")
CMD.auto(linuxcnc.AUTO_RESUME)
elif STAT.state == linuxcnc.RCS_EXEC and (STAT.paused or interp_paused):
LOG.debug("program.run: attempting AUTO resume path")
CMD.auto(linuxcnc.AUTO_RESUME)
elif STAT.file == "":
LOG.warning("program.run: no file loaded, skipping AUTO start")
elif setTaskMode(linuxcnc.MODE_AUTO):
LOG.debug("program.run: starting AUTO run at line %s", start_line)
CMD.auto(linuxcnc.AUTO_RUN, start_line)
else:
LOG.warning(
"program.run: no action taken; state=%s mode=%s interp=%s paused=%s feed_hold=%s",
STAT.state,
STAT.task_mode,
STAT.interp_state,
STAT.paused,
STAT.feed_hold_enabled,
)
def _run_ok(widget=None):
"""Checks if it is OK to run a program.
Args:
widget (QWidget, optional) : If a widget is supplied it will be
enabled/disabled according to the result, and will have it's
statusTip property set to the reason the action is disabled.
Returns:
bool : True if Ok, else False.
"""
if IN_DESIGNER:
return
interp_paused = STAT.interp_state == linuxcnc.INTERP_PAUSED
mdi_resume_ready = (
STAT.task_mode == linuxcnc.MODE_MDI
and (STAT.feed_hold_enabled or STAT.paused or interp_paused)
)
if STAT.estop:
ok = False
msg = "Can't run program when in E-Stop"
elif not STAT.enabled:
ok = False
msg = "Can't run program when not enabled"
elif not STATUS.allHomed():
ok = False
msg = "Can't run program when not homed"
elif mdi_resume_ready:
ok = True
msg = "Resume MDI command"
elif not STAT.paused and STAT.interp_state not in (linuxcnc.INTERP_IDLE, linuxcnc.INTERP_PAUSED):
ok = False
msg = "Can't run program when already running"
elif STAT.file == "":
ok = False
msg = "Can't run program when no file loaded"
else:
ok = True
msg = "Run program"
_run_ok.msg = msg
if widget is not None:
widget.setEnabled(ok)
widget.setStatusTip(msg)
widget.setToolTip(msg)
return ok
def _run_bindOk(widget):
STATUS.estop.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _run_ok(widget)))
STATUS.enabled.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _run_ok(widget)))
STATUS.all_axes_homed.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _run_ok(widget)))
STATUS.interp_state.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _run_ok(widget)))
STATUS.file.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _run_ok(widget)))
run.ok = _run_ok
run.bindOk = _run_bindOk
# -------------------------------------------------------------------------
# program RUN from LINE action
# -------------------------------------------------------------------------
def run_from_line(line=None):
# TODO: This might should show a popup to select start line,
# or it could get the start line from the gcode view or
# even from the backplot.
LOG.error('Run from line not implemented yet.')
run_from_line.ok = _run_ok
run_from_line.bindOk = _run_bindOk
# -------------------------------------------------------------------------
# program STEP action
# -------------------------------------------------------------------------
[docs]
def step():
"""Steps program line by line
ActionButton syntax::
program.step
"""
if STAT.state == linuxcnc.RCS_EXEC and STAT.paused:
CMD.auto(linuxcnc.AUTO_STEP)
elif setTaskMode(linuxcnc.MODE_AUTO):
CMD.auto(linuxcnc.AUTO_STEP)
step.ok = _run_ok
step.bindOk = _run_bindOk
# -------------------------------------------------------------------------
# program PAUSE action
# -------------------------------------------------------------------------
[docs]
def pause():
"""Pause executing program
ActionButton syntax::
program.pause
"""
LOG.debug(
"program.pause requested: state=%s mode=%s interp=%s paused=%s feed_hold=%s",
STAT.state,
STAT.task_mode,
STAT.interp_state,
STAT.paused,
STAT.feed_hold_enabled,
)
CMD.auto(linuxcnc.AUTO_PAUSE)
def _pause_ok(widget=None):
"""Checks if it is OK to pause the program.
Args:
widget (QWidget, optional) : If a widget is supplied it will be
enabled/disabled according to the result, and will have it's
statusTip property set to the reason the action is disabled.
Returns:
bool : True if Ok, else False.
"""
if STAT.state == linuxcnc.RCS_EXEC and not STAT.paused:
msg = "Pause program execution"
ok = True
elif STAT.paused:
msg = "Program is already paused"
ok = False
else:
msg = "No program running to pause"
ok = False
_pause_ok.msg = msg
if widget is not None:
widget.setEnabled(ok)
widget.setStatusTip(msg)
widget.setToolTip(msg)
return ok
def _pause_bindOk(widget):
STATUS.state.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _pause_ok(widget)))
STATUS.paused.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _pause_ok(widget)))
pause.ok = _pause_ok
pause.bindOk = _pause_bindOk
# -------------------------------------------------------------------------
# program RESUME action
# -------------------------------------------------------------------------
[docs]
def resume():
"""Resume a previously paused program
ActionButton syntax::
program.resume
"""
LOG.debug(
"program.resume requested: state=%s mode=%s interp=%s paused=%s feed_hold=%s",
STAT.state,
STAT.task_mode,
STAT.interp_state,
STAT.paused,
STAT.feed_hold_enabled,
)
CMD.auto(linuxcnc.AUTO_RESUME)
def _resume_ok(widget):
"""Checks if it is OK to resume a paused program.
Args:
widget (QWidget, optional) : If a widget is supplied it will be
enabled/disabled according to the result, and will have it's
statusTip property set to the reason the action is disabled.
Returns:
bool : True if Ok, else False.
"""
if STAT.state == linuxcnc.RCS_EXEC and STAT.paused:
ok = True
msg = "Resume program execution"
else:
ok = False
msg = "No paused program to resume"
_resume_ok.msg = msg
if widget is not None:
widget.setEnabled(ok)
widget.setStatusTip(msg)
widget.setToolTip(msg)
return ok
def _resume_bindOk(widget):
STATUS.paused.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _resume_ok(widget)))
STATUS.state.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _resume_ok(widget)))
resume.ok = _resume_ok
resume.bindOk = _resume_bindOk
# -------------------------------------------------------------------------
# program ABORT action
# -------------------------------------------------------------------------
[docs]
def abort():
"""Aborts any currently executing program, MDI command or homing operation.
ActionButton syntax::
program.abort
"""
LOG.debug("Aborting program")
CMD.abort()
def _abort_ok(widget=None):
"""Checks if it is OK to abort current operation.
Args:
widget (QWidget, optional) : If a widget is supplied it will be
enabled/disabled according to the result, and will have it's
statusTip property set to the reason the action is disabled.
Returns:
bool : True if Ok, else False.
"""
if STAT.state == linuxcnc.RCS_EXEC or STAT.state == linuxcnc.RCS_ERROR:
ok = True
msg = ""
else:
ok = False
msg = "Nothing to abort"
_abort_ok.msg = msg
if widget is not None:
widget.setEnabled(ok)
widget.setStatusTip(msg)
widget.setToolTip(msg)
return ok
def _abort_bindOk(widget):
STATUS.state.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _abort_ok(widget)))
abort.ok = _abort_ok
abort.bindOk = _abort_bindOk
# -------------------------------------------------------------------------
# BLOCK DELETE actions
# -------------------------------------------------------------------------
[docs]
class block_delete:
"""Block Delete Group"""
[docs]
@staticmethod
def on():
"""Start ignoring lines beginning with '/'.
ActionButton syntax::
program.block-delete.on
"""
LOG.debug("Setting block delete green<ON>")
CMD.set_block_delete(True)
[docs]
@staticmethod
def off():
"""Stop ignoring lines beginning with '/'.
ActionButton syntax::
program.block-delete.off
"""
LOG.debug("Setting block delete red<OFF>")
CMD.set_block_delete(False)
[docs]
@staticmethod
def toggle():
"""Toggle ignoring lines beginning with '/'.
ActionButton syntax::
program.block-delete.toggle
"""
if STAT.block_delete == True:
block_delete.off()
else:
block_delete.on()
def _block_delete_ok(widget=None):
"""Checks if it is OK to set block_delete.
Args:
widget (QWidget, optional) : If a widget is supplied it will be
enabled/disabled according to the result, and will have it's
statusTip property set to the reason the action is disabled.
Returns:
bool : True if Ok, else False.
"""
if STAT.task_state == linuxcnc.STATE_ON:
ok = True
msg = ""
else:
ok = False
msg = "Machine must be ON to set Block Del"
_block_delete_ok.msg = msg
if widget is not None:
widget.setEnabled(ok)
widget.setStatusTip(msg)
widget.setToolTip(msg)
return ok
def _block_delete_bindOk(widget):
widget.setChecked(STAT.block_delete)
STATUS.task_state.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _block_delete_ok(widget)))
STATUS.block_delete.onValueChanged(safe_qt_callback(widget, lambda s: widget.setChecked(s)))
block_delete.on.ok = block_delete.off.ok = block_delete.toggle.ok = _block_delete_ok
block_delete.on.bindOk = block_delete.off.bindOk = block_delete.toggle.bindOk = _block_delete_bindOk
# -------------------------------------------------------------------------
# OPTIONAL STOP actions
# -------------------------------------------------------------------------
[docs]
class optional_stop:
"""Optional Stop Group"""
[docs]
@staticmethod
def on():
"""Pause when a line beginning with M1 is encountered
ActionButton syntax::
program.optional-stop.on
"""
LOG.debug("Setting optional stop green<ON>")
CMD.set_optional_stop(True)
[docs]
@staticmethod
def off():
"""Don't pause when a line beginning with M1 is encountered
ActionButton syntax::
program.option-stop.off
"""
LOG.debug("Setting optional stop red<OFF>")
CMD.set_optional_stop(False)
[docs]
@staticmethod
def toggle():
"""Toggle pause when a line beginning with M1 is encountered
ActionButton syntax::
program.optional-stop.toggle
"""
if STAT.optional_stop == True:
optional_stop.off()
else:
optional_stop.on()
def _optional_stop_ok(widget=None):
"""Checks if it is OK to set optional_stop.
Args:
widget (QWidget, optional) : If a widget is supplied it will be
enabled/disabled according to the result, and will have it's
statusTip property set to the reason the action is disabled.
Returns:
bool : True if Ok, else False.
"""
if STAT.task_state == linuxcnc.STATE_ON:
ok = True
msg = ""
else:
ok = False
msg = "Machine must be ON to set Opt Stop"
_optional_stop_ok.msg = msg
if widget is not None:
widget.setEnabled(ok)
widget.setStatusTip(msg)
widget.setToolTip(msg)
return ok
def _optional_stop_bindOk(widget):
widget.setChecked(STAT.block_delete)
STATUS.task_state.onValueChanged(safe_qt_callback(widget, lambda *args, **kwargs: _optional_stop_ok(widget)))
STATUS.optional_stop.onValueChanged(safe_qt_callback(widget, lambda s: widget.setChecked(s)))
optional_stop.on.ok = optional_stop.off.ok = optional_stop.toggle.ok = _optional_stop_ok
optional_stop.on.bindOk = optional_stop.off.bindOk = optional_stop.toggle.bindOk = _optional_stop_bindOk
optional_skip = block_delete
#==============================================================================
# Program preprocessing handlers
#==============================================================================
import os, sys, time, select, re
import tempfile, atexit, shutil
FILTER_TEMP = None
def openFilterProgram(infile, prog_name):
temp_dir = _mktemp()
outfile = os.path.join(temp_dir, os.path.basename(infile))
#FilterProgram(prog_name, infile, outfile, lambda r: r or _loadFilterResult(outfile))
FilterProgram(prog_name, infile, outfile, None)
CMD.program_open(outfile)
LOG.debug('Linuxcnc Command - program_open')
def _loadFilterResult(fname):
if fname:
CMD.program_open(fname)
def _mktemp():
global FILTER_TEMP
if FILTER_TEMP is not None:
return FILTER_TEMP
FILTER_TEMP = tempfile.mkdtemp(prefix='emcflt-', suffix='.d')
atexit.register(lambda: shutil.rmtree(FILTER_TEMP))
return FILTER_TEMP
# slightly reworked code from gladevcp
# loads a filter program and collects the result
progress_re = re.compile("^FILTER_PROGRESS=(\\d*)$")
class FilterProgram:
def __init__(self, prog_name, infile, outfile, callback=None):
import subprocess
outfile = open(outfile, "w")
infile = infile.replace("'", "'\\''")
env = dict(os.environ)
env['AXIS_PROGRESS_BAR'] = '1'
self.p = subprocess.run(["sh", "-c", "%s '%s'" % (prog_name, infile)],
stdin=subprocess.PIPE,
stdout=outfile,
stderr=subprocess.PIPE,
env=env,
text=True)
self.stderr_text = []
self.program_filter = prog_name
self.callback = callback
#self.gid = STATUS.onValueChanged('periodic', self.update)
#progress = Progress(1, 100)
#progress.set_text(_("Filtering..."))
# force file load until know what to do about update/finish
def update(self, w):
if self.p.poll() is not None:
self.finish()
STATUS.disconnect(self.gid)
return False
r, w, x = select.select([self.p.stderr], [], [], 0)
if not r:
return True
stderr_line = self.p.stderr.readline()
m = progress_re.match(stderr_line)
if m:
pass #progress.update(int(m.group(1)), 1)
else:
self.stderr_text.append(stderr_line)
sys.stderr.write(stderr_line)
return True
def finish(self):
# .. might be something left on stderr
for line in self.p.stderr:
m = progress_re.match(line)
if not m:
self.stderr_text.append(line)
sys.stderr.write(line)
r = self.p.returncode
if r:
self.error(r, "".join(self.stderr_text))
if self.callback:
self.callback(r)
def error(self, exitcode, stderr):
LOG.error("Error loading filter program!")
# dialog = gtk.MessageDialog(None, 0, gtk.MESSAGE_ERROR, gtk.BUTTONS_CLOSE,
# _("The program %(program)r exited with code %(code)d. "
# "Any error messages it produced are shown below:")
# % {'program': self.program_filter, 'code': exitcode})
# diaLOG.format_secondary_text(stderr)
# diaLOG.run()
# diaLOG.destroy()