#!/usr/bin/env python3 import sys,os,time,pprint,multiprocessing import concurrent.futures import traceback as tb import logging import hal import gi gi.require_version('Gtk', '3.0') from gi.repository import Gtk as gtk import gladevcp import gladevcp.makepins import linuxcnc LOG = logging.getLogger(__name__) STAT = linuxcnc.stat() CMD = linuxcnc.command() ERR = linuxcnc.error_channel() MODES = ['MODE_MANUAL', 'MODE_AUTO', 'MODE_MDI'] CPUCOUNT = multiprocessing.cpu_count IN_DESIGNER = os.getenv('DESIGNER', False) def get_handlers(halcomp,builder,useropts,gscreen): return [HandlerClass(halcomp,builder,useropts,gscreen)] class HandlerClass: def spawn_workers(self, num_workers = 1, cmd_one = False, cmd_two = False, cmd_three = False, cmd_four = False): if (CPUCOUNT >= 3): if (num_workers > 4): num_workers = 4 # Using this code, operations can be performed without blocking the main thread. Use with caution with concurrent.futures.ThreadPoolExecutor(max_workers=int(num_workers)) as e: if cmd_one: e.submit(cmd_one) if cmd_two: e.submit(cmd_two) if cmd_three: e.submit(cmd_three) if cmd_four: e.submit(cmd_four) else: # In the rare event that a system does not support multithreading, this code is used instead if cmd_one: exec(cmd_one) if cmd_two: exec(cmd_two) if cmd_three: exec(cmd_three) if cmd_four: exec(cmd_four) # This will be pretty standard to gain access to everything # emc is for control and status of linuxcnc # data is important data from gscreen and linuxcnc # widgets is all the widgets from the glade files # gscreen is for access to gscreens methods def __init__(self, halcomp,builder,useropts,gscreen): if IN_DESIGNER: return self.gscreen = gscreen self.data = gscreen.data self.emc = gscreen.emc self.widgets = gscreen.widgets self.halcomp = halcomp self.widgets['window1'].props.decorated = False self.widgets['window1'].set_keep_above = True self.park_pos = (20,20,0) self.jog_mode = 'continuous' self.jog_speed = 'Slow' self.z_jog_enabled = False self.z_jog_speed = 'Slow' self.xy_increment = 20 self.z_increment = 10 self.jog_vels = {'Slow': 0, 'Fast': 0} try: # Try to open ini file self.ini = self.emc.emc.ini(sys.argv[2]) self.jog_vels['Slow'] = float(self.ini.find("DISPLAY", "DEFAULT_LINEAR_VELOCITY") or "unknown") self.jog_vels['Fast'] = self.jog_vels['Slow'] * 2.5 self.jog_vel = self.jog_vels['Slow'] self.z_jog_vel = self.jog_vels['Slow'] * 0.5 except: tb.print_exc() # This connects signals without using glade's autoconnect method def connect_signals(self,handlers): signal_list = [ #Connect custom signals #Simple buttons ['home_all', 'on_home_all_clicked'], ['mode_manual_button', 'on_mode_manual_button_clicked'], ['mode_auto_button', 'on_mode_auto_button_clicked'], ['gremlin_zoom_in', 'on_gremlin_zoom_in'], ['gremlin_zoom_out', 'on_gremlin_zoom_out'], ['gremlin_view_x', 'on_gremlin_view_x'], ['gremlin_view_y', 'on_gremlin_view_z'], ['gremlin_view_z', 'on_gremlin_view_z'], #Connect Gscreen signals #2 arg connects ['window1', 'destroy', 'on_window1_destroy'], ['run_halshow', 'clicked', 'on_halshow'], #['run_ladder', 'clicked', 'on_ladder'], ['feed_override', 'clicked', 'set_feed_override'], ] for sig in signal_list: if len(sig) == 2: self.widgets[sig[0]].connect('pressed', getattr(self,sig[1]),True) self.widgets[sig[0]].connect('released', getattr(self,sig[1]),False) elif len(sig) == 3: self.widgets[sig[0]].connect(sig[1], self.gscreen[sig[2]]) elif len(sig) == 4: self.widgets[sig[0]].connect(sig[1], self.gscreen[sig[2]],sig[3]) self.widgets['open_file_button'].connect('file-set', self.on_open_file_button_clicked) self.widgets['jog_speed_button'].connect('toggled', self.on_jog_speed_button_toggled) self.widgets['jog_mode_button'].connect('toggled', self.on_jog_mode_button_toggled) self.widgets['z_jog_toggle'].connect('toggled', self.on_z_jog_toggled) self.widgets['z_speed_toggle'].connect('toggled', self.on_z_jog_speed_toggled) self.widgets['xy_incr_adj'].connect('value-changed', self.on_xy_incr_adj_changed) self.widgets['z_incr_adj'].connect('value-changed', self.on_z_incr_adj_changed) axes = ('x','y','z') for a in range(3): axis = axes[a] self.widgets[axis+'_neg_button'].connect('pressed', self.jog_axis,a,-1,True) self.widgets[axis+'_neg_button'].connect('released', self.jog_axis,a,-1,False) self.widgets[axis+'_pos_button'].connect('pressed', self.jog_axis,a,1,True) self.widgets[axis+'_pos_button'].connect('released', self.jog_axis,a,1,False) ords = ('upl','upr','downl','downr') for ord in ords: self.widgets[ord+'_button'].connect('pressed', self.on_diagonal_jog,True,ord) self.widgets[ord+'_button'].connect('released', self.on_diagonal_jog,False,ord) #self.show_gscreen_info() #self.show_joint_info() # Gscreen built-in override def initialize_widgets(self): self.gscreen.init_show_windows() # If we need extra HAL pins here is where we do it. # Gscreen built-in override def initialize_pins(self): pass # every 100 milli seconds this gets called # add pass so gscreen doesn't try to update it's regular widgets or # add the individual function names that you would like to call. # Gscreen built-in override def periodic(self): STAT.poll() def show_hal_info(self): print("showing hal info") pin_dicts = hal.get_info_pins() print("\nHAL Pins:") for pin in pin_dicts: #LOG.info(pin.get('NAME')) print(pin.get('NAME')) sig_dicts = hal.get_info_signals() print("\nHAL Signals:") for sig in sig_dicts: print(sig.get('NAME')) def show_gscreen_info(self): for item in dir(self.data): print(item) def show_joint_info(self): STAT.poll() params = ['g5x_index', 'g5x_offset', 'joint_actual_position'] for p in params: print(p, getattr(STAT, p)) def debug_led(self, state): if state: hal.set_p('hm2_7i96s.0.ssr.00.out-02', '1') else: hal.set_p('hm2_7i96s.0.ssr.00.out-02', '0') #*/---------- Handler Functions ------------------------------------/*# def on_window1_destroy(self, object, data=None): gtk.main_quit() def on_gtk_quit_activate(self, menuitem, data=None): gtk.main_quit() def on_open_file_button_clicked(self,widget): file = widget.get_file() path = file.get_path() try: CMD.program_open(path) CMD.wait_complete() except: tb.print_exc() def on_home_all_clicked(self,widget,state): self.debug_led(state) if state: hal.set_p('halui.mode.joint', '1') hal.set_p('halui.home-all', '1') else: hal.set_p('halui.mode.joint', '0') hal.set_p('halui.home-all', '0') def on_mode_manual_button_clicked(self,widget,state): self.emc.set_manual_mode() def on_mode_auto_button_clicked(self,widget,state): self.emc.set_auto_mode() #def on_mode_mdi_button_clicked(self,widget,state): #self.emc.set_mdi_mode() def on_gremlin_view_x(self,widget,state): if state: self.widgets['gremlin'].set_view_x() def on_gremlin_view_y(self,widget,state): if state: self.widgets['gremlin'].set_view_y() def on_gremlin_view_z(self,widget,state): if state: self.widgets['gremlin'].set_view_z() def on_gremlin_zoom_in(self,widget,state): if state: self.widgets['gremlin'].zoom_in() def on_gremlin_zoom_out(self,widget,state): if state: self.widgets['gremlin'].zoom_out() def on_jog_speed_button_toggled(self,widget): if self.jog_speed == 'Slow': self.jog_speed = 'Fast' else: self.jog_speed = 'Slow' self.jog_vel = self.jog_vels[self.jog_speed] widget.set_label(self.jog_speed) def on_jog_mode_button_toggled(self,widget): if self.jog_mode == 'continuous': self.jog_mode = 'incremental' widget.set_label('Incremental on') else: self.jog_mode = 'continuous' widget.set_label('Incremental off') def on_z_jog_toggled(self,widget): self.z_jog_enabled = not self.z_jog_enabled lab = 'On' if self.z_jog_enabled else 'Off' widget.set_label(lab) def on_z_jog_speed_toggled(self,widget): if self.z_jog_speed == 'Slow': self.z_jog_speed = 'Fast' else: self.z_jog_speed = 'Slow' self.z_jog_vel = self.jog_vels[self.z_jog_speed] * 0.5 widget.set_label(self.z_jog_speed) def on_diagonal_jog(self,widget,state,dir): match dir: case 'upl': self.jog_axis(widget,0,-1,state) self.jog_axis(widget,1,1,state) case 'upr': self.jog_axis(widget,0,1,state) self.jog_axis(widget,1,1,state) case 'downl': self.jog_axis(widget,0,-1,state) self.jog_axis(widget,1,-1,state) case 'downr': self.jog_axis(widget,0,1,state) self.jog_axis(widget,1,-1,state) def on_xy_incr_adj_changed(self,widget): self.xy_increment = widget.get_value() def on_z_incr_adj_changed(self,widget): self.z_increment = widget.get_value() def jog_axis(self,widget,axis,direction,state): if self.emc.get_mode() == 1: if self.jog_mode == 'continuous': try: if not state: CMD.jog(linuxcnc.JOG_STOP, 0, axis) else: if axis == 2: if self.z_jog_enabled: CMD.jog(linuxcnc.JOG_CONTINUOUS, 0, axis, self.jog_vel*direction*0.5) else: CMD.jog(linuxcnc.JOG_CONTINUOUS, 0, axis, self.jog_vel*direction) except Exception as e: print(str(e)) tb.print_exc() elif state: try: if axis == 2: CMD.jog(linuxcnc.JOG_INCREMENT, 0, axis, self.jog_vel*direction, self.z_increment) else: CMD.jog(linuxcnc.JOG_INCREMENT, 0, axis, self.jog_vel*direction, self.xy_increment) except: tb.print_exc()