Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
self.value = self.getValue()
self.valueChanged[self.type].emit(self.value)
def setLogChange(self, log_change):
self.log_change = log_change
def getLogChange(self):
return self.log_change
def convertType(self, value):
if self.type == bool:
return value.lower() in ['true', '1']
return self.type(value)
class HalStatus(DataPlugin):
"""docstring for StatusPoller"""
def __init__(self):
super(HalStatus, self).__init__()
self.cycle_time = 100
self.linuxcnc_is_alive = False
self.status_items = {}
self.pin_dict = {}
# Create a thread for checking the HAL pins and signals
self.hal_mutex = threading.Lock()
self.hal_thread = threading.Thread(target=self.hal_poll_thread)
self.hal_thread.daemon = True
self.hal_thread.start()
# You should have received a copy of the GNU General Public License
# along with QtPyVCP. If not, see .
from opcua import Server
from opcua.common.events import Event
# from IPython import embed
from qtpyvcp import PLUGINS
from qtpyvcp.plugins import DataPlugin
from qtpyvcp.utilities.logger import getLogger
LOG = getLogger(__name__)
class OpcUA(DataPlugin):
def __init__(self):
super(OpcUA, self).__init__()
# https://github.com/FreeOpcUa/freeopcua/blob/master/python/examples/server.py
# setup our server
self.server = Server()
address = '0.0.0.0'
port = 4840
self.endpoint = "opc.tcp://{}:{}/qtpyvcp/server/".format(address, port)
self.server.set_endpoint(self.endpoint)
self.server.set_server_name("QtPyVCP OpcUa Server")
class Channel:
def __init__(self, plugin):
self.plugin = plugin
def on_get(self, req, resp):
"""Handles GET requests"""
channel = req.query_string
resp_data = self.plugin.getChannel(channel)[0]
resp.status = falcon.HTTP_200
resp.body = str(resp_data)
class VcpApi(DataPlugin):
def __init__(self):
super(VcpApi, self).__init__()
api = falcon.API()
clock = getPlugin('clock')
status = getPlugin('status')
offsets = getPlugin('offsettable')
positions = getPlugin('position')
tools = getPlugin('tooltable')
self.clock = Channel(clock)
self.status = Channel(status)
self.offsets = Channel(offsets)
self.positions = Channel(positions)
self.tools = Channel(tools)
import time
import linuxcnc
from qtpy.QtWidgets import QApplication
from qtpyvcp.utilities.logger import getLogger
from qtpyvcp.plugins import DataPlugin, DataChannel, getPlugin
from qtpyvcp.lib.native_notification import NativeNotification
from qtpyvcp.lib.dbus_notification import DBusNotification
LOG = getLogger(__name__)
STATUS = getPlugin('status')
class Notifications(DataPlugin):
"""
Notification data plugin
Args:
enabled (bool, optional): Enable or disable notification popups (Default = True)
mode (str, optional): native or dbus (Default = 'native')
max_messages (int, optional) Max number of notification popups to show.
persistent (bool, optional): Save notifications on shutdown (Default = True)
"""
def __init__(self, enabled=True, mode="native", max_messages=5,
persistent=True, **kwargs):
super(Notifications, self).__init__()
self.enabled = enabled
self.mode = mode
self.max_messages = max_messages
'Z': 'Z Offset',
}
# Column formats when writing tool table
INT_COLUMN_WIDTH = 6
FLOAT_COLUMN_WIDTH = 12
FLOAT_DECIMAL_PLACES = 6
def makeLorumIpsumToolTable():
return {i: merge(DEFAULT_TOOL,
{'T': i, 'P': i, 'R': 'Lorum Ipsum ' + str(i)})
for i in range(10)}
class ToolTable(DataPlugin):
TOOL_TABLE = {0: NO_TOOL}
DEFAULT_TOOL = DEFAULT_TOOL
COLUMN_LABELS = COLUMN_LABELS
tool_table_changed = Signal(dict)
def __init__(self, columns='TPXYZABCUVWDIJQR', file_header_template=None,
remember_tool_in_spindle=True):
super(ToolTable, self).__init__()
self.fs_watcher = None
self.orig_header_lines = []
self.file_header_template = file_header_template or ''
self.remember_tool_in_spindle = remember_tool_in_spindle
self.columns = self.validateColumns(columns) or [c for c in 'TPXYZABCUVWDIJQR']
def __init__(self, parent=None):
super(CompleterDelegate, self).__init__(parent)
items = []
for plugin, obj in iterPlugins():
if isinstance(obj, DataPlugin):
for chan_name in obj.channels:
items.append('{}:{}'.format(plugin, chan_name))
self.completer = QtWidgets.QCompleter(sorted(items))
self.completer.setCompletionColumn(0)
self.completer.setCompletionRole(QtCore.Qt.EditRole)
self.completer.setCaseSensitivity(QtCore.Qt.CaseInsensitive)
.. code-block:: yaml
data_plugins:
clock:
provider: qtpyvcp.plugins.clock:Clock
"""
from datetime import datetime
from qtpy.QtCore import QTimer
from qtpyvcp.plugins import DataPlugin, DataChannel
class Clock(DataPlugin):
"""Clock Plugin"""
def __init__(self):
super(Clock, self).__init__()
# set initial values
self.time.setValue(datetime.now())
self.date.setValue(datetime.now())
# make the clock tick
self.timer = QTimer()
self.timer.timeout.connect(self.tick)
@DataChannel
def time(self, chan):
"""The current time, updated every second.
CMD = linuxcnc.command()
LOG = getLogger(__name__)
STATUS = getPlugin('status')
STAT = STATUS.stat
INFO = Info()
def merge(a, b):
"""Shallow merge two dictionaries"""
r = a.copy()
r.update(b)
return r
class OffsetTable(DataPlugin):
DEFAULT_OFFSET = {
0: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
1: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
2: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
3: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
4: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
5: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
6: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
7: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0],
8: [0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0, 0.0]
}
NO_TOOL = merge(DEFAULT_OFFSET, {'T': 0, 'R': 'No Tool Loaded'}) # FIXME Requires safe removal
COLUMN_LABELS = [
'X',
from qtpy.QtCore import QTimer, QFileSystemWatcher
from qtpyvcp.utilities.logger import getLogger
from qtpyvcp.app.runtime_config import RuntimeConfig
from qtpyvcp.plugins import DataPlugin, DataChannel
from qtpyvcp.utilities.info import Info
INFO = Info()
LOG = getLogger(__name__)
STAT = linuxcnc.stat()
CMD = linuxcnc.command()
class Status(DataPlugin):
stat = STAT
def __init__(self, cycle_time=100):
super(Status, self).__init__()
self.no_force_homing = INFO.noForceHoming()
self.file_watcher = None
# recent files
self.max_recent_files = 10
with RuntimeConfig('~/.axis_preferences') as rc:
files = rc.get('DEFAULT', 'recentfiles', default=[])
files = [file for file in files if os.path.exists(file)]
self.recent_files.setValue(files)
class MultiOrderedDict(OrderedDict):
def __setitem__(self, key, value):
# print(key, value)
if key in self.keys():
items = self.get(key)
new = None
if isinstance(value, list):
new = value[0]
if new and new not in items:
items.append(new)
else:
super(MultiOrderedDict, self).__setitem__(key, value)
class IniConfig(DataPlugin):
def __init__(self, ini_file=None):
super(IniConfig, self).__init__()
self.config = ConfigParser(dict_type=MultiOrderedDict)
self.ini_file = os.path.abspath(ini_file or
os.environ.get("INI_FILE_NAME"))
self.sections = list()
def initialise(self):
self.config.read(self.ini_file)
for section in self.config.sections():
print(section)
for option in self.config.items(section):