import os
import json
from PySide6.QtCore import Property, QTimer
from qtpyvcp.widgets.input_widgets.line_edit import VCPLineEdit
from qtpyvcp.widgets.base_widgets import VarWidgetMixin
from qtpyvcp.utilities import logger
from qtpyvcp.utilities.misc import cnc_float
from qtpyvcp.utilities.qt_safety import safe_qt_callback
from qtpyvcp.plugins import getPlugin
from qtpyvcp.actions.machine_actions import issue_mdi
LOG = logger.getLogger(__name__)
IN_DESIGNER = os.getenv('DESIGNER') != None
_INI_CONFIG_CACHE = None
_INI_CONFIG_CACHE_KEY = None
_WARNED_MISSING_INI_ENV = False
_WARNED_MISSING_PARAMETER_FILE = False
def _safe_import_linuxcnc():
"""Import linuxcnc with intelligent error handling"""
if IN_DESIGNER:
# Expected in designer mode - don't log
return None
try:
import linuxcnc
return linuxcnc
except ImportError as e:
# Unexpected in runtime mode - log the error
LOG.error(f"Failed to import linuxcnc in runtime mode: {e}")
return None
def _get_cached_ini_configuration():
"""Resolve LinuxCNC ini + parameter file path once per process/env."""
global _INI_CONFIG_CACHE
global _INI_CONFIG_CACHE_KEY
global _WARNED_MISSING_INI_ENV
global _WARNED_MISSING_PARAMETER_FILE
ini_file_name = os.getenv('INI_FILE_NAME')
config_dir = os.getenv('CONFIG_DIR', os.path.dirname(ini_file_name) if ini_file_name else None)
cache_key = (ini_file_name, config_dir)
if _INI_CONFIG_CACHE is not None and _INI_CONFIG_CACHE_KEY == cache_key:
return _INI_CONFIG_CACHE
if not ini_file_name:
if not _WARNED_MISSING_INI_ENV:
LOG.warning("VCPVarLineEdit: INI_FILE_NAME environment variable not set")
_WARNED_MISSING_INI_ENV = True
return None
linuxcnc = _safe_import_linuxcnc()
if linuxcnc is None:
return None
ini_file = linuxcnc.ini(ini_file_name)
parameter_file = ini_file.find('RS274NGC', 'PARAMETER_FILE')
if not parameter_file:
if not _WARNED_MISSING_PARAMETER_FILE:
LOG.warning("VCPVarLineEdit: PARAMETER_FILE not found in [RS274NGC] section of ini file")
_WARNED_MISSING_PARAMETER_FILE = True
return None
if not os.path.isabs(parameter_file):
parameter_file_path = os.path.join(config_dir, parameter_file)
else:
parameter_file_path = parameter_file
_INI_CONFIG_CACHE = {
'ini_file': ini_file,
'config_dir': config_dir,
'parameter_file_path': parameter_file_path,
}
_INI_CONFIG_CACHE_KEY = cache_key
return _INI_CONFIG_CACHE
[docs]
class VCPVarLineEdit(VCPLineEdit, VarWidgetMixin):
"""Var Parameter Line Edit
A number entry that reads and writes a LinuxCNC numbered parameter, such
as ``#3014``, so the value is stored in the var file.
A value typed in is written when editing finishes, on Enter, Tab or
leaving the field, as the MDI command ``#<varParameterNumber> = <value>``
to 6 decimal places, the precision the var file stores. Text set any
other way, for example by a widget rule, is written ``writeDelay`` ms
later.
At startup, and whenever the var file changes, the field shows the
parameter's value. It is not overwritten while the field has focus, or
when a widget rule sets its ``Text``.
Setup:
* Set ``varParameterNumber`` in Qt Designer. Nothing is read or written
while it is 0.
* The var file is found automatically from ``[RS274NGC] PARAMETER_FILE``
in the INI.
* ``displayDecimals`` sets how many decimals are shown, 4 by default and
6 at most. The value is always stored to 6.
Safety:
* With ``requireHomed`` on, the default, the field is disabled unless the
machine is on, homed and idle. This is checked at startup and whenever
homing changes, and a write is refused unless the machine is in that
state.
"""
def __init__(self, parent=None):
VCPLineEdit.__init__(self, parent)
VarWidgetMixin.__init__(self)
# Widget properties for var parameter functionality
self._auto_write_enabled = True
self._write_delay = 500 # ms delay before writing to prevent excessive writes
self._display_decimals = 4 # User-configurable display formatting
self._require_homed = True # Safety feature: require machine to be homed
# Status monitoring for safety
self._status = None
self._status_homed_callback = None
self._original_enabled_state = True
# Internal 6-decimal storage (LinuxCNC var parameter limit)
self._internal_value = None
self._user_just_edited = False
self._pending_user_commit = False
# LinuxCNC configuration (legacy - will be removed)
self._ini_file = None
self._parameter_file_path = None
self._config_dir = None
self._loadIniConfiguration()
# Timer for delayed writing
self._write_timer = QTimer()
self._write_timer.setSingleShot(True)
self._write_timer.timeout.connect(self._writeToLinuxCNC)
# Connect to value changes
self.textChanged.connect(self._onTextChanged)
self.editingFinished.connect(self.onEditingFinished)
def _loadIniConfiguration(self):
"""Load LinuxCNC ini file and extract parameter file path"""
config = _get_cached_ini_configuration()
if config is None:
return
self._ini_file = config['ini_file']
self._config_dir = config['config_dir']
self._parameter_file_path = config['parameter_file_path']
[docs]
def getParameterFilePath(self):
"""Get the automatically detected parameter file path"""
return self._parameter_file_path
[docs]
def getConfigurationInfo(self):
"""
Get debugging information about the current configuration.
Returns:
dict: Configuration information including paths and settings
"""
return {
'ini_file_env': os.getenv('INI_FILE_NAME'),
'config_dir': self._config_dir,
'parameter_file_path': self._parameter_file_path,
'effective_parameter_file_path': self.getParameterFilePath(),
'var_parameter_number': self.var_parameter_number,
'auto_write_enabled': self._auto_write_enabled,
'write_delay': self._write_delay,
'display_decimals': self._display_decimals,
'var_monitoring_enabled': self.var_monitoring_enabled,
'parameter_file_exists': os.path.exists(self.getParameterFilePath()) if self.getParameterFilePath() else False
}
@Property(int)
def varParameterNumber(self):
"""LinuxCNC parameter number to write to (e.g., 3014 for #3014)"""
return self.var_parameter_number
@varParameterNumber.setter
def varParameterNumber(self, param_num):
self.var_parameter_number = param_num
# Load the current value from VarFileManager when parameter number is set
if param_num > 0 and self.getParameterFilePath():
# Use a timer to defer loading until after widget is fully initialized
QTimer.singleShot(100, self.loadParameterValue)
@Property(bool)
def autoWriteEnabled(self):
"""Whether to automatically write changes to LinuxCNC parameters"""
return self._auto_write_enabled
@autoWriteEnabled.setter
def autoWriteEnabled(self, enabled):
self._auto_write_enabled = bool(enabled)
@Property(int)
def writeDelay(self):
"""Delay in milliseconds before writing to LinuxCNC after text changes"""
return self._write_delay
@writeDelay.setter
def writeDelay(self, delay):
self._write_delay = int(delay)
@Property(int)
def displayDecimals(self):
"""Number of decimal places to display in the widget"""
return self._display_decimals
@displayDecimals.setter
def displayDecimals(self, decimals):
self._display_decimals = max(0, min(decimals, 6)) # Cap at 6 decimals max
# Update display immediately when decimals setting changes
if self._internal_value is not None:
self.setDisplayValue(self._internal_value)
@Property(bool)
def requireHomed(self):
"""Whether to require machine to be homed before enabling the widget"""
return self._require_homed
@requireHomed.setter
def requireHomed(self, require):
self._require_homed = bool(require)
# Re-evaluate enabled state when this property changes
self._updateEnabledState()
def _connectStatusPlugin(self):
"""Connect to the status plugin for monitoring machine state"""
self._status = getPlugin('status')
if not self._status:
LOG.warning("VCPVarLineEdit: Status plugin not available")
return
# Connect to all homed status signal using a tracked safe callback so
# terminate() can disconnect the exact callable that was connected.
self._status_homed_callback = safe_qt_callback(self, self._updateEnabledState)
self._status.all_axes_homed.signal.connect(self._status_homed_callback)
LOG.debug("VCPVarLineEdit: Connected to status plugin for homing monitoring")
# Update initial state
self._updateEnabledState()
def _updateEnabledState(self):
"""Update widget enabled state based on machine state"""
if not self._require_homed:
# If homing not required, use original enabled state
super().setEnabled(self._original_enabled_state)
return
if self._status is None:
# If no status plugin, default to disabled for safety
super().setEnabled(False)
self.setToolTip("Status plugin not available - widget disabled for safety")
return
# Check if machine is in safe state for parameter editing
linuxcnc = _safe_import_linuxcnc()
if linuxcnc is None:
# Disable widget when linuxcnc unavailable
super().setEnabled(False)
self.setToolTip("LinuxCNC not available - widget disabled")
return
stat = linuxcnc.stat()
stat.poll()
# Check machine state: ON, HOMED, and IDLE (same as MDI button safety)
is_machine_on = stat.task_state == linuxcnc.STATE_ON
is_all_homed = self._status.allHomed()
is_idle = stat.interp_state == linuxcnc.INTERP_IDLE
is_safe = is_machine_on and is_all_homed and is_idle
if is_safe:
super().setEnabled(self._original_enabled_state)
self.setToolTip("") # Clear any safety tooltip
else:
super().setEnabled(False)
if not is_machine_on:
self.setToolTip("Widget disabled: Machine must be ON")
elif not is_all_homed:
self.setToolTip("Widget disabled: Machine must be HOMED")
elif not is_idle:
self.setToolTip("Widget disabled: Machine must be IDLE")
[docs]
def setEnabled(self, enabled):
"""Override setEnabled to track original state and respect safety requirements"""
self._original_enabled_state = enabled
self._updateEnabledState()
[docs]
def setValue(self, value):
"""Set the value with 6-decimal internal storage and formatted display"""
float_value = cnc_float(value)
# Always store with 6-decimal precision (LinuxCNC var limit)
self._internal_value = round(float_value, 6)
# Format for display using user-configurable decimals
self.setDisplayValue(self._internal_value)
[docs]
def value(self):
"""Return the stored 6-decimal precision value"""
if self._internal_value is not None:
return self._internal_value
else:
return round(cnc_float(self.text()), 6) if self.text() else 0.0
[docs]
def setDisplayValue(self, value):
"""Set display value with consistent formatting using displayDecimals"""
# Skip settings notifications if user just edited to prevent overriding
if self._user_just_edited:
return
self.blockSignals(True)
# Always format using displayDecimals setting
float_value = cnc_float(value)
display_text = self.formatValue(float_value)
self.setText(display_text)
self.blockSignals(False)
[docs]
def onEditingFinished(self):
"""Handle user editing with 6-decimal precision storage and display formatting"""
user_text = self.text()
if not user_text.strip():
self._pending_user_commit = False
return
try:
user_value = cnc_float(user_text)
except ValueError:
LOG.warning("VCPVarLineEdit: invalid numeric input '%s'", user_text)
return
# Set flag to prevent settings notification from overriding
self._user_just_edited = True
# Store with 6-decimal precision (LinuxCNC var limit)
self._internal_value = round(user_value, 6)
# Format display using displayDecimals setting
formatted_text = self.formatValue(self._internal_value)
self.blockSignals(True)
self.setText(formatted_text)
self.setModified(False)
self.blockSignals(False)
self._user_just_edited = False
# Commit user edits only when editing is finished (Enter/Tab/focus-out).
if self._auto_write_enabled and self._var_parameter_number > 0:
self._write_timer.stop()
self._writeToLinuxCNC()
def _onTextChanged(self):
"""Handle text changes and schedule LinuxCNC parameter update if auto-write is enabled"""
if self.hasFocus():
self._pending_user_commit = True
# Update internal value when user types
if self.text():
try:
self._internal_value = round(cnc_float(self.text()), 6)
except ValueError:
# Invalid input - don't update internal value yet
pass
if self._auto_write_enabled and self._var_parameter_number > 0:
# Reset the timer to delay the write
self._write_timer.stop()
self._write_timer.start(self._write_delay)
[docs]
def writeToLinuxCNC(self, force=False):
"""
Public method to manually trigger writing to LinuxCNC parameters.
Args:
force (bool): If True, write immediately without delay
"""
if force:
self._writeToLinuxCNC()
else:
self._onTextChanged()
def _writeToLinuxCNC(self):
"""Write the current 6-decimal precision value to LinuxCNC via MDI command"""
if IN_DESIGNER:
LOG.debug("VCPVarLineEdit: skipping write in Designer mode")
return
# While the user is actively editing this field, defer writes until
# editingFinished (Enter/Tab/focus-out) to avoid partial-value commits.
if self.hasFocus() and self._pending_user_commit:
if self._auto_write_enabled and self._var_parameter_number > 0:
self._write_timer.stop()
self._write_timer.start(self._write_delay)
return
if not self.var_parameter_number > 0:
LOG.warning("VCPVarLineEdit: parameter number not set")
return
# Safety check: ensure machine is in safe state if required
if self._require_homed and self._status:
linuxcnc = _safe_import_linuxcnc()
if linuxcnc is None:
return # Skip writing when linuxcnc unavailable
stat = linuxcnc.stat()
stat.poll()
# Check machine state: ON, HOMED, and IDLE
is_machine_on = stat.task_state == linuxcnc.STATE_ON
is_all_homed = self._status.allHomed()
is_idle = stat.interp_state == linuxcnc.INTERP_IDLE
if not (is_machine_on and is_all_homed and is_idle):
LOG.warning("VCPVarLineEdit: Cannot write parameter - machine not in safe state")
return
# Get the 6-decimal precision value
if self._internal_value is not None:
value = self._internal_value
LOG.debug(f"VCPVarLineEdit: Using internal 6-decimal value: {value}")
else:
value = round(cnc_float(self.text()), 6)
LOG.debug(f"VCPVarLineEdit: Using text value rounded to 6 decimals: {value}")
# Use LinuxCNC MDI command to set parameter with 6-decimal precision
mdi_command = f"#{self.var_parameter_number} = {value:.6f}"
LOG.debug(f"VCPVarLineEdit: Issuing MDI command: {mdi_command}")
issue_mdi(mdi_command)
self._pending_user_commit = False
LOG.debug(f"VCPVarLineEdit: Set parameter #{self.var_parameter_number} = {value:.6f} via MDI")
[docs]
def readParameterFromVarFile(self, parameter_number=None):
"""
Read a parameter value directly from the var file.
Args:
parameter_number (int): Parameter number to read. If None, uses self._var_parameter_number
Returns:
float or None: The parameter value with 6-decimal precision, or None if not found
"""
param_num = parameter_number or self._var_parameter_number
if not param_num:
LOG.warning("VCPVarLineEdit: No parameter number specified for reading")
return None
var_file_path = self.getParameterFilePath()
if not var_file_path or not os.path.exists(var_file_path):
LOG.warning(f"VCPVarLineEdit: Parameter file not found: {var_file_path}")
return None
with open(var_file_path, 'r') as f:
for line in f:
line = line.strip()
if not line or line.startswith('#'):
continue
# Parse parameter line: "parameter_number<tab>value"
parts = line.split('\t')
if len(parts) >= 2:
file_param_num = int(parts[0])
if file_param_num == param_num:
# Round to 6 decimals to match LinuxCNC var precision
value = round(cnc_float(parts[1]), 6)
LOG.debug(f"VCPVarLineEdit: Read parameter #{param_num} = {value:.6f} from var file")
return value
LOG.debug(f"VCPVarLineEdit: Parameter #{param_num} not found in var file")
return None
[docs]
def loadParameterValue(self):
"""Load the parameter value from the var file into the widget"""
if self._var_parameter_number > 0:
value = self.readParameterFromVarFile()
if value is not None:
# Temporarily disable auto-write to prevent feedback loop
old_auto_write = self._auto_write_enabled
self._auto_write_enabled = False
self.setValue(value)
self._auto_write_enabled = old_auto_write
LOG.debug(f"VCPVarLineEdit: Loaded parameter #{self._var_parameter_number} = {value:.6f}")
else:
LOG.debug(f"VCPVarLineEdit: Could not load parameter #{self._var_parameter_number}")
[docs]
def onReturnPressed(self):
"""Override return-press behavior and let editingFinished perform commit"""
# Call parent onReturnPressed for VCP rules functionality
super(VCPVarLineEdit, self).onReturnPressed()
[docs]
def initialize(self):
"""Initialize the widget - called by VCP system"""
# Initialize VarWidgetMixin (sets up var file monitoring)
self._setup_var_monitoring()
# Connect to status plugin for safety monitoring
self._connectStatusPlugin()
LOG.debug(f"VCPVarLineEdit initialized: param #{self.var_parameter_number}, auto_write={self._auto_write_enabled}, require_homed={self._require_homed}")
[docs]
def terminate(self):
"""Cleanup when widget is destroyed"""
# Cleanup var file monitoring
self._cleanup_var_monitoring()
# Disconnect from status plugin
if self._status and self._status_homed_callback is not None:
try:
self._status.all_axes_homed.signal.disconnect(self._status_homed_callback)
except (RuntimeError, TypeError):
# Signal may already be disconnected during shutdown ordering.
pass
self._status_homed_callback = None
# Restore original enabled state
if hasattr(self, '_original_enabled_state'):
self.setEnabled(self._original_enabled_state)
if self._write_timer.isActive():
self._write_timer.stop()
LOG.debug("VCPVarLineEdit terminated")
# VarWidgetMixin abstract method implementations
def _load_parameter_value(self, value):
"""Load a parameter value from var file into the line edit widget"""
# Temporarily disable auto-write to prevent feedback loops
old_auto_write = self._auto_write_enabled
self._auto_write_enabled = False
# Set the value using existing setValue method
self.setValue(value)
# Restore auto-write setting
self._auto_write_enabled = old_auto_write
LOG.debug(f"VCPVarLineEdit: Loaded parameter value {value} from var file")
def _get_widget_value(self):
"""Get the current value from the line edit widget"""
return self.value()
def _has_text_rule(self):
"""Return True when this widget has a rule that sets the Text property."""
try:
rules = json.loads(self.rules or '[]')
except Exception:
return False
for rule in rules:
if isinstance(rule, dict) and rule.get('property') == 'Text':
return True
return False
def _on_parameter_changed(self, param_number, new_value):
"""Handle parameter change notifications from VarFileManager"""
if self.hasFocus():
return
# If Text is controlled by widget rules, do not overwrite the displayed
# expression result with var-file updates.
if self._has_text_rule():
return
if param_number == self.var_parameter_number and new_value is not None:
LOG.debug(f"VCPVarLineEdit: Parameter #{param_number} changed to {new_value}")
self._load_parameter_value(new_value)