# -*- coding: utf-8 -*- # import native python modules import time import math import sys import platform import io import os.path import traceback import tables import xml.parsers.expat import threading import webbrowser import xdg.BaseDirectory import tempfile import gi gi.require_version('Gtk', '3.0') gi.require_version('GtkSource', '3.0') # import 3rd party modules # gui graphics from gi.repository import GObject as gobject import sqlite3 import uuid from gi.repository import Gtk as gtk gtk_version_missmatch = gtk.check_version( 3, 0, 0 ) if gtk_version_missmatch: raise Exception( "insufficient gtk version: " + gtk_version_missmatch ) from gi.repository import Gdk as gdk from gi.repository import GtkSource as gtksourceview2 gobject.type_register(gtksourceview2.View) from gi.repository import Pango as pango # array math import numpy # math graphics import matplotlib # force noninteractive matplotlib.rcParams[ "interactive" ] = "False" matplotlib.rcParams[ "text.usetex" ] = "False" matplotlib.rcParams[ "axes.formatter.limits" ] = "-3,3" from matplotlib.backends.backend_gtk3agg import FigureCanvasGTK3Agg as FigureCanvas #from matplotlib.backends.backend_gtk3cairo import FigureCanvasGTK3Cairo as FigureCanvas max_points_to_display = 0 import matplotlib.axes import matplotlib.figure # import our own stuff from damaris.gui import ExperimentWriter, ExperimentHandling from damaris.gui import ResultReader, ResultHandling from damaris.gui import BackendDriver #from damaris.gui.gtkcodebuffer import CodeBuffer, SyntaxLoader #from damaris.data import Drawable # this is a base class, it should be used... from damaris.data import DataPool, Accumulation, ADC_Result, MeasurementResult # default, can be set to true from start script debug = False # version info from damaris import __version__ class logstream: gui_log = None text_log = sys.__stdout__ def write( self, message ): # default for stdout and stderr if self.gui_log is not None: self.gui_log( message ) if debug or self.gui_log is None: self.text_log.write( message ) self.text_log.flush( ) def __call__( self, message ): self.write( message ) def flush(self): pass def __del__( self ): pass global log log = logstream( ) sys.stdout = log sys.stderr = log ExperimentHandling.log = log ResultHandling.log = log ResultReader.log = log ExperimentWriter.log = log BackendDriver.log = log DataPool.log = log class LockFile: def __init__(self): self.index = 0 self.file = os.path.expanduser("~/.damaris.lockdb") self.conn = sqlite3.connect(self.file) self.cursor = self.conn.cursor() self.id = None print("Connected to db..") def add_experiment(self): print("Add experiment..") self.id = str(uuid.uuid4()) self.cursor.execute("INSERT INTO damaris VALUES (?, ?)",(self.id, "started")) self.conn.commit() return self.id def del_experiment(self, id): print("Delete experiment..") self.cursor.execute('DELETE FROM damaris WHERE uuid=?', (self.id,)) self.conn.commit() return True def am_i_next(self): first = self.cursor.execute("SELECT * FROM damaris ORDER BY ROWID ASC LIMIT 1") current_running = first.fetchone()[0] #print current_running,self.id if current_running == self.id: return True else: return False class DamarisGUI: ExpScript_Display = 0 ResScript_Display = 1 Monitor_Display = 2 Log_Display = 3 Config_Display = 4 Edit_State = 0 Run_State = 1 Pause_State = 2 Stop_State = 3 Quit_State = 4 def __init__(self, exp_script_filename=None, res_script_filename=None, start_immediately=False): # state: edit, run, stop, quit # state transitions: # edit -> run|quit # run -> pause|stop # pause -> run|stop # stop -> edit self.state = DamarisGUI.Edit_State # script execution engines and backend driver self.si = None # produced and displayed data self.data = None self.glade_layout_init( ) # my notebook self.main_notebook = self.xml_gui.get_object( "main_notebook" ) self.log = LogWindow( self.xml_gui , self) self.sw = ScriptWidgets( self.xml_gui ) self.toolbar_init( ) self.documentation_init( ) self.monitor = MonitorWidgets( self.xml_gui ) self.config = ConfigTab( self.xml_gui ) # lock file to prevent other DAMARIS to start immediatly self.lockfile = LockFile() # = os.path.expanduser("~/.damaris.lock") self.id = None #to stop queued experiments self.stop_experiment_flag = threading.Event() exp_script = "" if exp_script_filename is not None and exp_script_filename != "": self.sw.exp_script_filename = exp_script_filename[ : ] if os.path.isfile( exp_script_filename ) and os.access( exp_script_filename, os.R_OK ): exp_script = self.sw.load_file_as_unicode( exp_script_filename ) res_script = "" if res_script_filename is not None and res_script_filename != "": self.sw.res_script_filename = res_script_filename[ : ] if os.path.isfile( res_script_filename ) and os.access( res_script_filename, os.R_OK ): res_script = self.sw.load_file_as_unicode( res_script_filename ) self.sw.set_scripts( exp_script, res_script ) if start_immediately: if exp_script and res_script: self.start_immediately = start_immediately else: raise "RuntimeError" else: self.start_immediately = False self.statusbar_init( ) #connect event handlers eventhandlers = {"on_toolbar_run_button_clicked": self.start_experiment, "on_toolbar_pause_button_toggled": self.pause_experiment, "on_toolbar_stop_button_clicked": self.stop_experiment, "on_doc_menu_activate": self.show_doc_menu, "on_toolbar_manual_button_clicked": self.show_manual, "on_data_handling_column_textfield_value_changed": self.sw.column_line_widgets_changed_event, "on_data_handling_line_textfield_value_changed": self.sw.column_line_widgets_changed_event, "on_experiment_script_line_textfield_value_changed": self.sw.column_line_widgets_changed_event, "on_experiment_script_column_textfield_value_changed": self.sw.column_line_widgets_changed_event, "on_toolbar_open_file_button_clicked": self.sw.open_file, "on_toolbar_new_button_clicked": self.sw.new_file, "on_toolbar_save_as_button_clicked": self.sw.save_file_as, "on_toolbar_save_file_button_clicked": self.sw.save_file, "on_toolbar_save_all_button_clicked": self.sw.save_all_files, "on_toolbar_check_scripts_button_clicked": self.sw.check_script, "on_toolbar_undo_button_clicked": self.sw.undo, "on_toolbar_redo_button_clicked": self.sw.redo, "on_toolbar_search_button_clicked": self.sw.search, "on_config_save_button_clicked": self.config.save_config_handler, "on_config_load_button_clicked": self.config.load_config_handler, "on_backend_executable_browse_button_clicked": self.config.browse_backend_executable_dialog, "on_fontbutton_font_set": self.config.set_script_font_handler, "on_display_autoscaling_checkbutton_toggled": self.monitor.display_autoscaling_toggled, "on_display_statistics_checkbutton_toggled": self.monitor.display_statistics_toggled, "on_display_x_scaling_combobox_changed": self.monitor.display_scaling_changed, "on_display_y_scaling_combobox_changed": self.monitor.display_scaling_changed, "on_display_save_data_as_text_button_clicked": self.monitor.save_display_data_as_text, "on_display_copy_data_to_clipboard_button_clicked": self.monitor.copy_display_data_to_clipboard, "on_data_handling_textview_event": self.sw.textviews_modified, "on_experiment_script_textview_event": self.sw.textviews_modified, "on_main_notebook_switch_page": self.nopevent, "main_window_close": self.nopevent } self.xml_gui.connect_signals(eventhandlers) self.main_window.maximize() self.main_window.show_all( ) self.main_window.present( ) def glade_layout_init( self ): glade_file = os.path.join( os.path.dirname( __file__ ), "damaris3.glade" ) self.xml_gui = gtk.Builder() self.xml_gui.add_from_file(glade_file) self.main_window = self.xml_gui.get_object( "main_window" ) self.main_window.connect( "delete-event", self.quit_event ) for icon_path in ["/usr/share/pixmaps/DAMARIS3.png", os.path.join( os.path.dirname( __file__ ), "DAMARIS3.png" )]: if os.path.exists(icon_path): self.main_window.set_icon_from_file(icon_path) break self.main_window.set_title( "DAMARIS-%s" % __version__ ) def statusbar_init( self ): """ experiment and result thread status, backend state """ self.experiment_script_statusbar_label = self.xml_gui.get_object( "statusbar_experiment_script_label" ) self.data_handling_statusbar_label = self.xml_gui.get_object( "statusbar_data_handling_label" ) self.backend_statusbar_label = self.xml_gui.get_object( "statusbar_core_label" ) def toolbar_init( self ): """ buttons like save and run... """ self.toolbar_stop_button = self.xml_gui.get_object( "toolbar_stop_button" ) self.toolbar_run_button = self.xml_gui.get_object( "toolbar_run_button" ) self.toolbar_pause_button = self.xml_gui.get_object( "toolbar_pause_button" ) # prepare for edit state self.toolbar_run_button.set_sensitive( True ) self.toolbar_stop_button.set_sensitive( False ) self.toolbar_pause_button.set_sensitive( False ) def run( self ): # prolong lifetime of clipboard till the very end (avoid error message) self.main_clipboard = self.sw.main_clipboard gdk.threads_enter( ) if self.start_immediately: self.toolbar_run_button.emit("clicked") gtk.main( ) gdk.threads_leave( ) self.si = None self.sw = None self.config = None self.xml_gui = None # event handling: the real acitons in gui programming # first global events def nopevent(self, widget, data=None, otherargument=None): #print(widget) #print(data) #print(otherargument) pass def quit_event( self, widget, data=None ): """ expecting quit event for main application """ if self.state in [ DamarisGUI.Edit_State, DamarisGUI.Quit_State ]: self.state = DamarisGUI.Quit_State # do a cleanup... if self.sw.experiment_script_textbuffer.get_modified( ): question = gtk.MessageDialog(parent=self.xml_gui.get_object( "main_window"), type=gtk.MessageType.WARNING, flags=gtk.DialogFlags.MODAL, buttons=("_Cancel", gtk.ResponseType.CANCEL, "_No", gtk.ResponseType.NO, "_Yes", gtk.ResponseType.YES)) question.set_markup("The Experiment Script has been changed. Save changes?") response = question.run() question.destroy() if response == gtk.ResponseType.CANCEL: return True elif response == gtk.ResponseType.YES: current_page = self.main_notebook.get_current_page( ) self.main_notebook.set_current_page(0) if self.sw.exp_script_filename is None: self.sw.save_file_as() else: self.sw.save_file() self.main_notebook.set_current_page(current_page) if self.sw.data_handling_textbuffer.get_modified( ): question = gtk.MessageDialog(parent=self.xml_gui.get_object( "main_window"), type=gtk.MessageType.WARNING, flags=gtk.DialogFlags.MODAL, buttons=("_Cancel", gtk.ResponseType.CANCEL, "_No", gtk.ResponseType.NO, "_Yes", gtk.ResponseType.YES)) question.set_markup("The Result Script has been changed. Save changes?") response = question.run() question.destroy() if response == gtk.ResponseType.CANCEL: return True elif response == gtk.ResponseType.YES: current_page = self.main_notebook.get_current_page( ) self.main_notebook.set_current_page(1) if self.sw.res_script_filename is None: self.sw.save_file_as() else: self.sw.save_file() self.main_notebook.set_current_page(current_page) self.config = None self.sw = None self.monitor = None self.log = None # and quit gtk.main_quit( ) return True else: question = gtk.MessageDialog(parent=self.xml_gui.get_object( "main_window"), type=gtk.MessageType.ERROR, flags=gtk.DialogFlags.MODAL, buttons=("_Understood!", gtk.ResponseType.OK)) question.set_markup("Experiment still running, can not close DAMARIS!") response = question.run() question.destroy() return True # toolbar related events: def start_experiment( self, widget, data=None ): # something running? if self.si is not None: print("Last Experiment is not clearly stopped!") self.si = None # get config values: actual_config = self.config.get( ) # get scripts and start script interface self.sw.disable_editing( ) exp_script, res_script = self.sw.get_scripts( ) if not actual_config[ "start_result_script" ]: res_script = "" if not actual_config[ "start_experiment_script" ]: exp_script = "" backend = actual_config[ "backend_executable" ] if not actual_config[ "start_backend" ]: backend = "" if (backend == "" and exp_script == "" and res_script == ""): print("nothing to do...so doing nothing!") self.sw.enable_editing( ) return # check whether scripts are syntactically valid # should be merged with check script function exp_code = None if exp_script != "": try: exp_code = compile( exp_script, "Experiment Script", "exec" ) except SyntaxError as e: ln = e.lineno lo = e.offset if type( ln ) is not int: ln = 0 if type( lo ) is not int: lo = 0 print("Experiment Script: %s at line %d, col %d:" % (e.__class__.__name__, ln, lo)) if e.text != "": print(e.text.rstrip()) print(" "*(e.offset-1)+"^") print(e) res_code = None if res_script != "": try: res_code = compile( res_script, "Result Script", "exec" ) except SyntaxError as e: ln = e.lineno lo = e.offset if type( ln ) is not int: ln = 0 if type( lo ) is not int: lo = 0 print("Result script: %s at line %d, col %d:" % (e.__class__.__name__, ln, lo)) if e.text != "": print("\"%s\"" % e.text) # print " "*(e.offset+1)+"^" # nice idea, but needs monospaced fonts pass print(e) # detect error if (exp_script != "" and exp_code is None) or \ (res_script != "" and res_code is None): self.main_notebook.set_current_page( DamarisGUI.Log_Display ) self.sw.enable_editing( ) return # prepare to run self.state = DamarisGUI.Run_State self.toolbar_run_button.set_sensitive( False ) self.toolbar_stop_button.set_sensitive( True ) self.toolbar_pause_button.set_sensitive( True ) self.toolbar_pause_button.set_active( False ) # delete old data self.data = None self.monitor.observe_data_pool( self.data ) # set the text mark for hdf logging if self.log.textbuffer.get_mark( "lastdumped" ) is None: self.log.textbuffer.create_mark( "lastdumped", self.log.textbuffer.get_end_iter( ), left_gravity=True ) loop_run=0 #user shorter interval because this blocks user events from the GUI interval = 0.1 self.stop_experiment_flag.clear() self.id = self.lockfile.add_experiment() while not self.lockfile.am_i_next(): while gtk.events_pending(): gtk.main_iteration_do(False) if self.stop_experiment_flag.wait(interval): break if loop_run > 1: self.experiment_script_statusbar_label.set_text("Waiting for other experiment to finish") loop_run = 0 loop_run += interval if self.stop_experiment_flag.isSet(): #Experiment has been stopped, clean up and leave self.experiment_script_statusbar_label.set_text("Experiment stopped") self.state = DamarisGUI.Edit_State self.sw.enable_editing( ) self.toolbar_run_button.set_sensitive( True ) self.toolbar_stop_button.set_sensitive( False ) self.toolbar_pause_button.set_sensitive( False ) self.toolbar_pause_button.set_active( False ) #delete experiment from queue so that next experiment may start self.lockfile.del_experiment(self.id) return # start experiment try: self.spool_dir = os.path.abspath( actual_config[ "spool_dir" ] ) # setup script engines self.si = ScriptInterface( exp_code, res_code, backend, self.spool_dir, clear_jobs=actual_config[ "del_jobs_after_execution" ], clear_results=actual_config[ "del_results_after_processing" ] ) self.data = self.si.data # run frontend and script engines self.monitor.observe_data_pool( self.data ) self.si.runScripts( ) except Exception as e: #print "ToDo evaluate exception",str(e), "at",traceback.extract_tb(sys.exc_info()[2])[-1][1:3] #print "Full traceback:" traceback_file = io.StringIO( ) traceback.print_tb( sys.exc_info( )[ 2 ], None, traceback_file ) self.main_notebook.set_current_page( DamarisGUI.Log_Display ) print("Error while executing scripts: %s\n" % str( e ) + traceback_file.getvalue( )) traceback_file = None self.data = None if self.si is not None: still_running = [_f for _f in [ self.si.exp_handling, self.si.res_handling, self.si.back_driver ] if _f] for r in still_running: r.quit_flag.set( ) print("waiting for threads stoping...", end=' ') still_running = [x for x in [ self.si.exp_handling, self.si.res_handling, self.si.back_driver ] if x is not None and x.is_alive( )] for t in still_running: while t.is_alive(): while gtk.events_pending(): gtk.main_iteration_do(False) t.join( 0.05 ) print("done") # cleanup self.si = None self.state = DamarisGUI.Edit_State self.sw.enable_editing( ) self.toolbar_run_button.set_sensitive( True ) self.toolbar_stop_button.set_sensitive( False ) self.toolbar_pause_button.set_sensitive( False ) self.toolbar_pause_button.set_active( False ) #delete Experiment from queue self.lockfile.del_experiment(self.id) return # switch to grapics self.main_notebook.set_current_page( DamarisGUI.Monitor_Display ) # set running if self.si.exp_handling is not None: self.experiment_script_statusbar_label.set_text( "Experiment Script Running (0)" ) else: self.experiment_script_statusbar_label.set_text( "Experiment Script Idle" ) if self.si.res_handling is not None: self.data_handling_statusbar_label.set_text( "Result Script Running (0)" ) else: self.data_handling_statusbar_label.set_text( "Result Script Idle" ) if self.si.back_driver is not None: self.backend_statusbar_label.set_text( "Backend Running" ) else: self.backend_statusbar_label.set_text( "Backend Idle" ) # start data dump self.dump_thread = None self.save_thread = None self.dump_filename = "" if actual_config[ "data_pool_name" ] != "": self.dump_states( init=True ) gobject.timeout_add( 200, self.observe_running_experiment ) def observe_running_experiment( self ): """ periodically look at running threads """ # look at components and update them # test whether backend and scripts are done r = self.si.data.get( "__recentresult", -1 ) + 1 b = self.si.data.get( "__resultsinadvance", -1 ) + 1 e = self.si.data.get( "__recentexperiment", -1 ) + 1 experimentstarttime = self.si.data.get( "__experimentstarted", 0 ) expectedexperimentruntime = self.si.data.get( "__totalexperimentlength", 0 ) experimentsfinishedtime = 0.0 for i in range(0, b): experimentsfinishedtime += self.si.data.get("__experimentlengths", {}).get(i, 0.0) experimentruntime = time.time() - experimentstarttime experimenttimetext = "" if experimentstarttime > 0 and expectedexperimentruntime > 0: runtimemin, runtimesec = divmod(math.floor(experimentruntime), 60) runtimehours, runtimemin = divmod(runtimemin, 60) experimenttimetext += " ({:02d}:{:02d}:{:02d} / ".format(int(runtimehours), int(runtimemin), int(runtimesec)) expectedmin, expectedsec = divmod(math.ceil(expectedexperimentruntime), 60) expectedhours, expectedmin = divmod(expectedmin, 60) experimenttimetext += "{:02d}:{:02d}:{:02d})".format(int(expectedhours), int(expectedmin), int(expectedsec)) backendtimetext = "" if experimentsfinishedtime > 0: finexperimmin, finexperimsec = divmod(round(experimentsfinishedtime), 60) finexperimhours, finexperimmin = divmod(finexperimmin, 60) backendtimetext = " ({:02d}:{:02d}:{:02d})".format(int(finexperimhours), int(finexperimmin), int(finexperimsec)) e_text = None r_text = None b_text = None if self.si.exp_handling is not None: if not self.si.exp_handling.is_alive( ): self.si.exp_handling.join( ) if self.si.exp_handling.raised_exception: print("experiment script failed at line %d (function %s): %s" % (self.si.exp_handling.location[ 0 ], self.si.exp_handling.location[ 1 ], self.si.exp_handling.raised_exception)) print("Full traceback", self.si.exp_handling.traceback) e_text = "Experiment Script Failed (%d)" % e else: e_text = "Experiment Script Finished (%d)" % e print("experiment script finished") self.si.exp_handling = None else: #print self.si.back_driver.get_messages() e_text = "Experiment Script Running (%d)" % e e_text += experimenttimetext if self.si.res_handling is not None: if not self.si.res_handling.is_alive( ): self.si.res_handling.join( ) if self.si.res_handling.raised_exception: print("result script failed at line %d (function %s): %s" % (self.si.res_handling.location[ 0 ], self.si.res_handling.location[ 1 ], self.si.res_handling.raised_exception)) print("Full traceback", self.si.res_handling.traceback) r_text = "Result Script Failed (%d)" % r else: r_text = "Result Script Finished (%d)" % r self.si.res_handling = None else: r_text = "Result Script Running (%d)" % r if self.si.back_driver is not None: if not self.si.back_driver.is_alive( ): if self.si.back_driver.raised_exception: b_text = "Backend Failed" else: b_text = "Backend Finished" self.si.back_driver.join( ) self.si.back_driver = None else: b_text = "Backend Running" if b != 0: b_text += " (%d)" % b b_text += backendtimetext if self.dump_thread is not None: if self.dump_thread.is_alive( ): sys.stdout.write( "." ) self.dump_dots += 1 if self.dump_dots > 80: print() self.dump_dots = 0 else: self.dump_thread.join( ) self.dump_thread = None dump_size = os.stat( self.dump_filename ).st_size / 1e6 print("done (%.1f s, %.1f MB)" % (time.time( ) - self.dump_start_time, dump_size)) gdk.threads_enter( ) if e_text: self.experiment_script_statusbar_label.set_text( e_text ) if r_text: self.data_handling_statusbar_label.set_text( r_text ) if b_text: self.backend_statusbar_label.set_text( b_text ) gdk.threads_leave( ) still_running = [_f for _f in [ self.si.exp_handling, self.si.res_handling, self.si.back_driver, self.dump_thread ] if _f] if len( still_running ) == 0: if self.save_thread is None and self.dump_filename != "": print("all subprocesses ended, saving data pool") # thread to save data... self.save_thread = threading.Thread( target=self.dump_states, name="dump states" ) self.save_thread.start( ) self.dump_start_time = time.time( ) self.dump_dots = 0 self.state = DamarisGUI.Stop_State if self.state == DamarisGUI.Stop_State: gdk.threads_enter( ) self.toolbar_pause_button.set_sensitive( False ) self.toolbar_stop_button.set_sensitive( False ) gdk.threads_leave( ) if len( still_running ) != 0: print("subprocess(es) still running: " + ', '.join( [s.getName( ) for s in still_running] )) return True else: if self.save_thread is not None: if self.save_thread.is_alive( ): sys.stdout.write( "." ) self.dump_dots += 1 if self.dump_dots > 80: print() self.dump_dots = 0 return True self.save_thread.join( ) self.save_thread = None dump_size = os.stat( self.dump_filename ).st_size / 1e6 print("done (%.1f s, %.1f MB)" % (time.time( ) - self.dump_start_time, dump_size)) # now everything is stopped self.state = DamarisGUI.Edit_State gdk.threads_enter( ) self.sw.enable_editing( ) self.toolbar_run_button.set_sensitive( True ) self.toolbar_stop_button.set_sensitive( False ) self.toolbar_pause_button.set_sensitive( False ) gdk.threads_leave( ) # keep data to display but throw away everything else self.si = None # delete lock file so that other experiment can start self.lockfile.del_experiment(self.id) return False # dump states? if self.dump_thread is None and self.dump_filename != "" and \ self.dump_timeinterval != 0 and self.last_dumped + self.dump_timeinterval < time.time( ): print("dumping data pool") self.dump_start_time = time.time( ) self.dump_dots = 0 # thread to save data... self.dump_thread = threading.Thread( target=self.dump_states, name="dump states" ) self.dump_thread.start( ) # or look at them again return True def dump_states( self, init=False ): """ init: constructs basic structure of this file compress: optional argument for zlib compression 0-9 """ if init: # calculate new settings for hdf files actual_config = self.config.get( ) # provide name tags for dump file name extensions_dict = { "date": time.strftime( "%Y-%m-%d" ), "datetime": time.strftime( "%Y-%m-%d_%H%M" ) } if self.sw.exp_script_filename is None: extensions_dict[ "exp" ] = "unnamed" else: extensions_dict[ "exp" ] = os.path.splitext( os.path.basename( self.sw.exp_script_filename ) )[ 0 ] if self.sw.res_script_filename is None: extensions_dict[ "res" ] = "unnamed" else: extensions_dict[ "res" ] = os.path.splitext( os.path.basename( self.sw.res_script_filename ) )[ 0 ] # create new dump file name try: self.dump_filename = actual_config.get( "data_pool_name" ) % extensions_dict except ValueError: print("invalid formating character in '" + actual_config.get( "data_pool_name" ) + "'") self.dump_filename = "DAMARIS_data_pool.h5" print("reseting dumpfile to " + self.dump_filename) del extensions_dict if os.path.split( self.dump_filename )[ 1 ] == "" or os.path.isdir( self.dump_filename ): print("The dump filename is a directory, using filename 'DAMARIS_data_pool.h5'") self.dump_filename += os.sep + "DAMARIS_data_pool.h5" # compression self.dump_complib = actual_config.get( "data_pool_complib", None ) if self.dump_complib == "None": self.dump_complib = None if self.dump_complib is not None: self.dump_complevel = int( actual_config.get( "data_pool_comprate", 0 ) ) else: self.dump_complevel = 0 # time interval self.dump_timeinterval = 60 * 60 * 10 try: self.dump_timeinterval = int( 60 * float( actual_config[ "data_pool_write_interval" ] ) ) except ValueError as e: print("configuration provides non-number for dump interval: " + str( e )) # if existent, move away old files if os.path.isfile( self.dump_filename ): # create bakup name pattern dump_filename_pattern = None (filename, ext) = os.path.splitext( self.dump_filename ) if ext in [ ".h5", ".hdf", ".hdf5" ]: dump_filename_pattern = filename.replace( "%", "%%" ) + "_%d" + ext else: dump_filename_pattern = self.dump_filename.replace( "%", "%%" ) + "_%d" last_backup = 0 cummulated_size = os.stat( self.dump_filename ).st_size while os.path.isfile( dump_filename_pattern % last_backup ): cummulated_size += os.stat( dump_filename_pattern % last_backup ).st_size last_backup += 1 while last_backup > 0: os.rename( dump_filename_pattern % (last_backup - 1), dump_filename_pattern % last_backup ) last_backup -= 1 os.rename( self.dump_filename, dump_filename_pattern % 0 ) if cummulated_size > (1 << 30): print("Warning: the cumulated backups size of '%s' is %d MByte" % (self.dump_filename, cummulated_size / (1 << 20))) # init is finnished now # now it's time to create the hdf file dump_file = None if not os.path.isfile( self.dump_filename ): if not init: print("dump file \"%s\" vanished unexpectedly, creating new one" % self.dump_filename) # have a look to the path and create necessary directories dir_stack = [ ] dir_trunk = os.path.dirname( os.path.abspath( self.dump_filename ) ) while dir_trunk != "" and not os.path.isdir( dir_trunk ): dir_stack.append( os.path.basename( dir_trunk ) ) dir_trunk = os.path.dirname( dir_trunk ) try: while len( dir_stack ): dir_trunk = os.path.join( dir_trunk, dir_stack.pop( ) ) if os.path.isdir( dir_trunk ): continue os.mkdir( dir_trunk ) except OSError as e: print(e) print("could not create dump directory '%s', so hdf5 dumps disabled" % dir_trunk) self.dump_filename = "" self.dump_timeinterval = 0 return True # create new dump file dump_file = tables.open_file( self.dump_filename, mode="w", title="DAMARIS experiment data" ) # write scripts and other useful information scriptgroup = dump_file.create_group( "/", "scripts", "Used Scripts" ) exp_text, res_text = self.sw.get_scripts( ) if self.si.exp_script: dump_file.create_array( scriptgroup, "experiment_script", exp_text.encode('utf-8') ) if self.si.res_script: dump_file.create_array( scriptgroup, "result_script", res_text.encode('utf-8') ) if self.si.backend_executable: dump_file.create_array( scriptgroup, "backend_executable", self.si.backend_executable.encode('utf-8') ) if self.spool_dir: dump_file.create_array( scriptgroup, "spool_directory", self.spool_dir.encode('utf-8') ) timeline_tablecols = numpy.recarray( 0, dtype=([ ("time", "S17"), ("experiments", "int64"), ("results", "int64") ]) ) timeline_table = dump_file.create_table( "/", "timeline", timeline_tablecols, title="Timeline of Experiment" ) if tables.__version__[ 0 ] == "1": logarray = dump_file.create_vlarray( where=dump_file.root, name="log", atom=tables.StringAtom( length=120 ), title="log messages", filters=tables.Filters( complevel=9, complib='zlib' ) ) else: logarray = dump_file.create_earray( where=dump_file.root, name="log", atom=tables.StringAtom( itemsize=120 ), shape=(0,), title="log messages", filters=tables.Filters( complevel=9, complib='zlib' ) ) if dump_file is None and os.path.isfile( self.dump_filename ) and tables.is_pytables_file( self.dump_filename ): # take some data from dump file and repack os.rename( self.dump_filename, self.dump_filename + ".bak" ) old_dump_file = tables.open_file( self.dump_filename + ".bak", mode="r+" ) if "data_pool" in old_dump_file.root: old_dump_file.remove_node( where="/", name="data_pool", recursive=True ) old_dump_file.copy_file( self.dump_filename ) old_dump_file.close( ) del old_dump_file os.remove( self.dump_filename + ".bak" ) # prepare for update dump_file = tables.open_file( self.dump_filename, mode="r+" ) if dump_file is None: # exit! print("could not create dump directory '%s', so hdf5 dumps disabled" % dir_trunk) self.dump_filename = "" self.dump_timeinterval = 0 return True # no undo please! if dump_file.is_undo_enabled( ): dump_file.disable_undo( ) # save the data! self.data.write_hdf5( dump_file, where="/", name="data_pool", complib=self.dump_complib, complevel=self.dump_complevel ) # now save additional information e = self.si.data.get( "__recentexperiment", -1 ) + 1 r = self.si.data.get( "__recentresult", -1 ) + 1 timeline_table = dump_file.root.timeline timeline_row = timeline_table.row timeline_row[ "time" ] = time.strftime( "%Y%m%d %H:%M:%S" ) timeline_row[ "experiments" ] = e timeline_row[ "results" ] = r timeline_row.append( ) timeline_table.flush( ) # save log window information: # also save backend's logfile information? logtextbuffer = self.log.textbuffer last_end = logtextbuffer.get_mark( "lastdumped" ) if last_end is None: last_end = logtextbuffer.create_mark( "lastdumped", logtextbuffer.get_start_iter( ), left_gravity=True ) #logtext_start = logtextbuffer.get_iter_at_mark( last_end ) logtext_end = logtextbuffer.get_end_iter( ) logtextbuffer.move_mark( last_end, logtext_end ) # Create new iterators after the mark move logtext_start = logtextbuffer.get_iter_at_mark( last_end ) logtext_end = logtextbuffer.get_end_iter( ) # recode from unicode logtext = logtextbuffer.get_text( logtext_start, logtext_end, True ) # avoid circular references (seems to be necessary with gtk-2.12) # del logtextbuffer, logtext_start, logtext_end, last_end for l in logtext.splitlines( ): dump_file.root.log.append( numpy.array( [ l ], dtype="S120" ) ) dump_file.flush( ) dump_file.close( ) self.last_dumped = time.time( ) del dump_file return True def pause_experiment( self, widget, data=None ): """ pause experiment execution (that means delay backend and let others run) """ if self.si is None: return False pause_state = self.toolbar_pause_button.get_active( ) if pause_state: if self.state != DamarisGUI.Run_State: return False if self.spool_dir is None: return False no = self.si.data.get( "__recentresult", -1 ) + 1 result_pattern = os.path.join( self.spool_dir, "job.%09d.result" ) job_pattern = os.path.join( self.spool_dir, "job.%09d" ) while os.path.isfile( result_pattern % no ): no += 1 i = 0 self.pause_files = [ ] while i < 3 and os.path.isfile( job_pattern % (no + i) ): pause_file = (job_pattern % (no + i)) + ".pause" os.rename( job_pattern % (no + i), pause_file ) self.pause_files.append( pause_file ) i += 1 self.state = DamarisGUI.Pause_State self.backend_statusbar_label.set_text( "Backend Paused" ) else: if self.state != DamarisGUI.Pause_State: return False self.state = DamarisGUI.Run_State for f in self.pause_files: os.rename( f, f[ :-6 ] ) self.pause_files = None self.backend_statusbar_label.set_text( "Backend Running" ) def stop_experiment( self, widget, data=None ): if self.state in [ DamarisGUI.Run_State, DamarisGUI.Pause_State ]: self.stop_experiment_flag.set() if self.si is None: return still_running = [_f for _f in [ self.si.exp_handling, self.si.res_handling, self.si.back_driver ] if _f] for r in still_running: r.quit_flag.set( ) self.state = DamarisGUI.Stop_State def documentation_init( self ): self.doc_urls = { "DAMARIS Manual": "file:///usr/share/doc/python3-damaris/doc/_build/html/index.html", } if os.path.isdir( "/usr/share/doc/python%d.%d-doc/html" % (sys.version_info[ :2 ]) ): self.doc_urls[ "Python Docs" ] = "file:///usr/share/doc/python%d.%d-doc/html/index.html" % ( sys.version_info[ :2 ]) doc_index_url = None self.doc_browser = None def show_doc_menu( self, widget, data=None ): """ offer a wide variety of docs, prefer local installations """ if type( widget ) is gtk.MenuItem: requested_doc = widget.get_child( ).get_text( ) else: requested_doc = "Python DAMARIS" if requested_doc in self.doc_urls and self.doc_urls[ requested_doc ] is not None: if self.doc_browser is not None: if not self.doc_browser.is_alive( ): self.doc_browser.join( ) if self.doc_browser.my_webbrowser is not None: print("new browser tab") self.doc_browser.my_webbrowser.open_new_tab( self.doc_urls[ requested_doc ] ) else: del self.doc_browser self.doc_browser = start_browser( self.doc_urls[ requested_doc ] ) self.doc_browser.start( ) else: self.doc_browser = start_browser( self.doc_urls[ requested_doc ] ) self.doc_browser.start( ) else: print("missing docs for '%s'" % (requested_doc)) show_manual = show_doc_menu class start_browser( threading.Thread ): def __init__( self, url ): threading.Thread.__init__( self, name="manual browser" ) self.my_webbrowser = None self.my_webbrowser_process = None self.start_url = url def run( self ): """ start a webbrowser """ if sys.hexversion >= 0x02050000: if sys.platform == "linux2" and self.my_webbrowser is None: # try for debian linux self.my_webbrowser = webbrowser.get( "x-www-browser" ) if self.my_webbrowser is None: # this is what it should be everywhere! self.my_webbrowser = webbrowser.get( ) if self.my_webbrowser is not None: print("starting web browser (module webbrowser)") self.my_webbrowser.open( self.start_url ) return True # last resort print("starting web browser (webbrowser.py)") self.my_webbrowser_process = os.spawnl( os.P_NOWAIT, sys.executable, os.path.basename( sys.executable ), "-c", "import webbrowser\nwebbrowser.open('%s')" % self.start_url ) return True class LogWindow: """ writes messages to the log window """ def __init__( self, xml_gui, damaris_gui ): self.xml_gui = xml_gui self.damaris_gui = damaris_gui self.textview = self.xml_gui.get_object( "messages_textview" ) self.textview.connect( "key-press-event", self.textview_keypress) font_desc = pango.FontDescription("Monospace 10") self.textview.modify_font(font_desc) self.textbuffer = self.textview.get_buffer( ) self.logstream = log self.logstream.gui_log = self self.last_timetag = None self( "Started in directory %s\n" % os.getcwd( ) ) def __call__( self, message ): timetag = time.time( ) gobject.idle_add( self.add_message_callback, timetag, message, priority=gobject.PRIORITY_LOW ) def add_message_callback( self, timetag, message ): date_tag = "" if self.last_timetag is None or (message != "\n" and self.last_timetag + 60 < timetag): self.last_timetag = timetag timetag_int = math.floor( timetag ) timetag_ms = int( (timetag - timetag_int) * 1000 ) date_tag = time.strftime( "%Y%m%d %H:%M:%S.%%03d:\n", time.localtime( timetag_int ) ) % timetag_ms gdk.threads_enter( ) self.textbuffer.place_cursor( self.textbuffer.get_end_iter( ) ) self.textbuffer.insert_at_cursor( date_tag + str( message ) ) self.textview.scroll_to_mark( self.textbuffer.get_insert( ), 0.1 , False, 0.5, 0.5) gdk.threads_leave( ) def textview_keypress( self, widget, event, data=None ): if event.state & gdk.ModifierType.CONTROL_MASK != 0: if event.keyval == gdk.keyval_from_name("f"): self.damaris_gui.sw.search(None, None) return True return False def __del__( self ): self.logstream.gui_log = None self.logstream = None class ScriptWidgets: def __init__( self, xml_gui ): """ initialize text widgets with text """ self.xml_gui = xml_gui # my states # editing enabled/disabled self.editing_state = True # keep in mind which filename was used for experiment script self.exp_script_filename = None # keep in mind which filename was used for result script self.res_script_filename = None # load syntax description for syntax highlighting (defined in python.xml) #sl = SyntaxLoader( "python" ) # script buffers: self.experiment_script_textview = self.xml_gui.get_object( "experiment_script_textview" ) self.data_handling_textview = self.xml_gui.get_object( "data_handling_textview" ) # create and set syntax-highlighting text-buffers as textview backends self.experiment_script_textbuffer = gtksourceview2.Buffer() self.data_handling_textbuffer = gtksourceview2.Buffer() lm = gtksourceview2.LanguageManager() langpython = lm.get_language("python") self.experiment_script_textbuffer.set_language(langpython) self.experiment_script_textbuffer.set_highlight_syntax(True) self.data_handling_textbuffer.set_language(langpython) self.data_handling_textbuffer.set_highlight_syntax(True) fontdesc = pango.FontDescription("monospace") self.experiment_script_textview.set_buffer( self.experiment_script_textbuffer ) self.experiment_script_textview.set_show_line_numbers(True) self.experiment_script_textview.set_auto_indent(True) self.experiment_script_textview.set_insert_spaces_instead_of_tabs(True) self.experiment_script_textview.set_tab_width(4) self.experiment_script_textview.set_smart_home_end(True) self.experiment_script_textview.modify_font(fontdesc) self.data_handling_textview.set_buffer( self.data_handling_textbuffer ) self.data_handling_textview.set_show_line_numbers(True) self.data_handling_textview.set_auto_indent(True) self.data_handling_textview.set_insert_spaces_instead_of_tabs(True) self.data_handling_textview.set_tab_width(4) self.data_handling_textview.set_smart_home_end(True) self.data_handling_textview.modify_font(fontdesc) keywords = """for if else elif in print try finally except global lambda not or pass def class import from as return yield while continue break assert None True False AccumulatedValue Accumulation MeasurementResult ADC_Result Experiment synchronize sleep result data issubclass min max abs pow range xrange log_range lin_range staggered_range file combine_ranges interleaved_range get_sampling_rate uses_statistics write_to_csv write_to_hdf get_job_id get_description set_description get_xdata get_ydata set_xdata set_ydata ttl_pulse ttls rf_pulse state_start state_end wait record loop_start loop_end set_frequency set_pfg set_dac set_phase set_pts_local get_length set_title get_title add_lineplotdata add_errorplotdata get_number_of_channels get_style set_style get_xlabel set_xlabel get_ylabel set_ylabel""" self.keywordsbuffer = gtksourceview2.Buffer() self.keywordsbuffer.set_text(keywords) compexp = self.experiment_script_textview.get_completion() compres = self.data_handling_textview.get_completion() compexp.set_property("show-headers", True) compres.set_property("show-headers", True) self.compwgen = gtksourceview2.CompletionWords.new("Keywords", None) self.compwgen.register(self.keywordsbuffer) self.compwgen.set_property("minimum-word-size", 1) self.compwexp = gtksourceview2.CompletionWords.new("Words", None) self.compwexp.register(self.experiment_script_textbuffer) self.compwexp.set_property("minimum-word-size", 2) self.compwres = gtksourceview2.CompletionWords.new("Words", None) self.compwres.register(self.data_handling_textbuffer) self.compwres.set_property("minimum-word-size", 2) compexp.add_provider(self.compwgen) compres.add_provider(self.compwgen) compexp.add_provider(self.compwexp) compres.add_provider(self.compwres) # clipboard self.main_clipboard = gtk.Clipboard( ) # statusbar self.experiment_script_statusbar_label = self.xml_gui.get_object( "statusbar_experiment_script_label" ) self.data_handling_statusbar_label = self.xml_gui.get_object( "statusbar_data_handling_label" ) # footer with line and col indicators self.experiment_script_line_indicator = self.xml_gui.get_object( "experiment_script_line_textfield" ) self.experiment_script_column_indicator = self.xml_gui.get_object( "experiment_script_column_textfield" ) self.data_handling_line_indicator = self.xml_gui.get_object( "data_handling_line_textfield" ) self.data_handling_column_indicator = self.xml_gui.get_object( "data_handling_column_textfield" ) # some event handlers self.experiment_script_textbuffer.connect( "modified-changed", self.textviews_modified ) self.experiment_script_textview.connect_after( "move-cursor", self.textviews_moved ) self.experiment_script_textview.connect( "key-press-event", self.textviews_keypress ) self.experiment_script_textview.connect( "button-release-event", self.textviews_clicked ) self.data_handling_textbuffer.connect( "modified-changed", self.textviews_modified ) self.data_handling_textview.connect_after( "move-cursor", self.textviews_moved ) self.data_handling_textview.connect( "key-press-event", self.textviews_keypress ) self.data_handling_textview.connect( "button-release-event", self.textviews_clicked ) # and the editing toolbar buttons self.toolbar_new_button = self.xml_gui.get_object( "toolbar_new_button" ) self.toolbar_open_button = self.xml_gui.get_object( "toolbar_open_file_button" ) self.toolbar_save_button = self.xml_gui.get_object( "toolbar_save_file_button" ) self.toolbar_save_as_button = self.xml_gui.get_object( "toolbar_save_as_button" ) self.toolbar_save_all_button = self.xml_gui.get_object( "toolbar_save_all_button" ) self.toolbar_check_scripts_button = self.xml_gui.get_object( "toolbar_check_scripts_button" ) self.toolbar_undo_button = self.xml_gui.get_object( "toolbar_undo_button" ) self.toolbar_redo_button = self.xml_gui.get_object( "toolbar_redo_button" ) self.toolbar_search_button = self.xml_gui.get_object("toolbar_search_button") # main notebook self.main_notebook = self.xml_gui.get_object( "main_notebook" ) # config toolbar self.main_notebook.connect_after( "switch_page", self.notebook_page_switched ) # start with empty scripts self.set_scripts( "", "" ) self.enable_editing( ) # public methods def set_scripts( self, exp_script=None, res_script=None ): # load buffers and set cursor to front if exp_script is not None: self.experiment_script_textbuffer.begin_not_undoable_action() self.experiment_script_textbuffer.set_text( str( exp_script ) ) self.experiment_script_textbuffer.place_cursor( self.experiment_script_textbuffer.get_start_iter( ) ) self.experiment_script_textbuffer.set_modified( False ) self.experiment_script_textbuffer.end_not_undoable_action() self.textviews_moved( self.experiment_script_textview ) if res_script is not None: self.data_handling_textbuffer.begin_not_undoable_action() self.data_handling_textbuffer.set_text( str( res_script ) ) self.data_handling_textbuffer.place_cursor( self.data_handling_textbuffer.get_start_iter( ) ) self.data_handling_textbuffer.set_modified( False ) self.data_handling_textbuffer.end_not_undoable_action() self.textviews_moved( self.data_handling_textview ) self.set_toolbuttons_status( ) def get_scripts( self ): """ returns scripts """ exp_script = self.experiment_script_textbuffer.get_text( self.experiment_script_textbuffer.get_start_iter( ), self.experiment_script_textbuffer.get_end_iter( ) , True).rstrip( ) res_script = self.data_handling_textbuffer.get_text( self.data_handling_textbuffer.get_start_iter( ), self.data_handling_textbuffer.get_end_iter( ) , True).rstrip( ) return (exp_script, res_script) def disable_editing( self ): """ disable editing (for running experiments) """ # disable buffers self.editing_state = False self.experiment_script_textview.set_sensitive( False ) self.data_handling_textview.set_sensitive( False ) self.set_toolbuttons_status( ) def enable_editing( self ): """ returns to editable state """ self.editing_state = True self.experiment_script_textview.set_sensitive( True ) self.data_handling_textview.set_sensitive( True ) self.set_toolbuttons_status( ) # methods to update status and appearance def set_toolbuttons_status( self ): """ ToDo: care about associated file names """ current_page = self.main_notebook.get_current_page( ) if self.editing_state and current_page in [ 0, 1 ]: self.toolbar_new_button.set_sensitive( True ) self.toolbar_open_button.set_sensitive( True ) # find visible tab exp_modified = self.experiment_script_textbuffer.get_modified( ) res_modified = self.data_handling_textbuffer.get_modified( ) enable_save = True canundo = False canredo = False if current_page == 0: enable_save = exp_modified and self.exp_script_filename is not None canundo = self.experiment_script_textbuffer.can_undo() canredo = self.experiment_script_textbuffer.can_redo() elif current_page == 1: enable_save = res_modified and self.res_script_filename is not None canundo = self.data_handling_textbuffer.can_undo() canredo = self.data_handling_textbuffer.can_redo() self.toolbar_save_button.set_sensitive( enable_save ) self.toolbar_save_as_button.set_sensitive( True ) self.toolbar_save_all_button.set_sensitive( exp_modified or res_modified ) self.toolbar_check_scripts_button.set_sensitive( True ) self.toolbar_undo_button.set_sensitive(canundo) self.toolbar_redo_button.set_sensitive(canredo) self.toolbar_search_button.set_sensitive(True) else: # disable toolbar self.toolbar_new_button.set_sensitive( False ) self.toolbar_open_button.set_sensitive( False ) self.toolbar_save_button.set_sensitive( False ) self.toolbar_save_as_button.set_sensitive( False ) self.toolbar_save_all_button.set_sensitive( False ) self.toolbar_check_scripts_button.set_sensitive( False ) self.toolbar_undo_button.set_sensitive(False) self.toolbar_redo_button.set_sensitive(False) self.toolbar_search_button.set_sensitive(current_page == 3) if self.exp_script_filename: exp_titlename = str( os.path.splitext( os.path.basename( self.exp_script_filename ) )[ 0 ] ) else: exp_titlename = "unnamed" if self.res_script_filename: res_titlename = str( os.path.splitext( os.path.basename( self.res_script_filename ) )[ 0 ] ) else: res_titlename = "unnamed" window_title = "DAMARIS-%s %s,%s" % (__version__, exp_titlename, res_titlename) self.xml_gui.get_object( "main_window" ).set_title( window_title ) # text widget related events def check_script( self, widget, data=None ): if not self.editing_state: return 0 current_page = self.main_notebook.get_current_page( ) if current_page not in [ 0, 1 ]: return 0 script = self.get_scripts( )[ current_page ] try: compile( script, ("Experiment Script", "Result Script")[ current_page ], "exec" ) except SyntaxError as se: if current_page == 0: tb = self.experiment_script_textbuffer tv = self.experiment_script_textview elif current_page == 1: tb = self.data_handling_script_textbuffer tv = self.data_handling_script_textview ln = se.lineno lo = se.offset # reguard http://bugs.python.org/issue1778 if type( ln ) is not int: ln = 0 if type( lo ) is not int: lo = 0 print("Syntax Error:\n%s in %s at line %d, offset %d" % ( str( se ), se.filename, ln, lo) + "\n(ToDo: Dialog)") if ln > 0 and ln <= tb.get_line_count( ): new_place = tb.get_iter_at_line_offset( ln - 1, 0 ) if lo > 0 and lo <= new_place.get_chars_in_line( ): new_place.set_line_offset( lo ) tb.place_cursor( new_place ) tv.scroll_to_iter( new_place, 0.2, False, 0, 0 ) except Exception as e: print("Compilation Error:\n" + str( e ) + "\n(ToDo: Dialog)") def notebook_page_switched( self, notebook, page, pagenumber ): self.set_toolbuttons_status( ) def column_line_widgets_changed_event( self, data=None ): page = self.main_notebook.get_current_page( ) if page == 0: textview = self.experiment_script_textview column_indicator = self.experiment_script_column_indicator newline = self.experiment_script_line_indicator.get_value_as_int( ) - 1 newcol = column_indicator.get_value_as_int( ) - 1 elif page == 1: textview = self.data_handling_textview column_indicator = self.data_handling_column_indicator newline = self.data_handling_line_indicator.get_value_as_int( ) - 1 newcol = column_indicator.get_value_as_int( ) - 1 else: print("unknown line/column selector") return False textbuffer = textview.get_buffer( ) new_place = textbuffer.get_iter_at_line( newline ) if not newcol > new_place.get_chars_in_line( ): new_place.set_line_offset( newcol ) else: column_indicator.set_value( 1 ) if len( textbuffer.get_selection_bounds( ) ) != 0: textbuffer.move_mark_by_name( "insert", new_place ) else: textbuffer.place_cursor( new_place ) textview.scroll_mark_onscreen( textbuffer.get_insert( ) ) textview.grab_focus( ) return True def textviews_modified( self, widget = None, event=None ): # mix into toolbar affairs if type(widget) is gtksourceview2.View: return self.textviews_moved( widget ) else: return False def textviews_clicked( self, widget, event ): return self.textviews_moved( widget ) def textviews_moved( self, widget, text=None, count=None, ext_selection=None, data=None ): textbuffer = widget.get_buffer( ) cursor_mark = textbuffer.get_insert( ) cursor_iter = textbuffer.get_iter_at_mark( cursor_mark ) if textbuffer is self.experiment_script_textbuffer: line_indicator = self.experiment_script_line_indicator column_indicator = self.experiment_script_column_indicator if textbuffer is self.data_handling_textbuffer: line_indicator = self.data_handling_line_indicator column_indicator = self.data_handling_column_indicator # do only necessary updates! li_range_new = textbuffer.get_end_iter( ).get_line( ) + 1 if line_indicator.get_range( )[ 1 ] != li_range_new: line_indicator.set_range( 1, li_range_new ) ci_range_new = cursor_iter.get_chars_in_line( ) + 1 if column_indicator.get_range( )[ 1 ] != ci_range_new: column_indicator.set_range( 1, ci_range_new ) cursor_line = cursor_iter.get_line( ) + 1 cursor_lineoffset = cursor_iter.get_line_offset( ) + 1 if line_indicator.get_value( ) != cursor_line: line_indicator.set_value( cursor_line ) if column_indicator.get_value( ) != cursor_lineoffset: column_indicator.set_value( cursor_lineoffset ) self.set_toolbuttons_status( ) return False def textviews_keypress( self, widget, event, data=None ): """ helpful tab and return key functions """ if event.state & gdk.ModifierType.CONTROL_MASK != 0: if event.keyval == gdk.keyval_from_name( "s" ): # save buffer page = self.main_notebook.get_current_page( ) if (self.exp_script_filename, self.res_script_filename)[ page ] is None: self.save_file_as( ) else: self.save_file( ) return True elif event.keyval == gdk.keyval_from_name( "S" ): # save both buffers self.save_all_files( None, None ) return True elif event.keyval == gdk.keyval_from_name("f"): self.search(None, None) return True return False #Handle Backspace (delete more than one space on line beginning) if event.keyval == 0xFF08: textbuffer = widget.get_buffer( ) if textbuffer.get_has_selection(): return False cursor_mark = textbuffer.get_insert( ) cursor_iter = textbuffer.get_iter_at_mark( cursor_mark ) linestart_iter = cursor_iter.copy( ) linestart_iter.set_line_offset( 0 ) linebegin = textbuffer.get_text( linestart_iter, cursor_iter, True ).expandtabs(4) if linebegin.isspace() and len(linebegin) > 0: linebegin = ' ' * int((len( linebegin ) - 1) / 4) * 4 textbuffer.delete( linestart_iter, cursor_iter ) textbuffer.insert( linestart_iter, linebegin ) return True elif event.keyval == 0xFF0D: textbuffer = widget.get_buffer( ) if textbuffer.get_has_selection(): return False cursor_mark = textbuffer.get_insert( ) cursor_iter = textbuffer.get_iter_at_mark( cursor_mark ) lastchar_iter = cursor_iter.copy() lastchar = lastchar_iter.backward_char( ) if not lastchar_iter.get_char( ) == ":": return False linestart_iter = cursor_iter.copy( ) linestart_iter.set_line_offset( 0 ) spaceend_iter = linestart_iter.copy( ) while (not spaceend_iter.ends_line( ) and not spaceend_iter.is_end( ) and spaceend_iter.get_char( ).isspace( )): spaceend_iter.forward_char( ) linebegin = textbuffer.get_text( linestart_iter, spaceend_iter, True ).expandtabs(4) indent_length = int((int(len(linebegin)/4)+1)*4) intext = "\n" + (" " * indent_length) textbuffer.insert(cursor_iter, intext) widget.scroll_to_mark( cursor_mark, 0.0, False, 0.5, 0.5 ) return True #self.textviews_moved(widget) return False def load_file_as_unicode( self, script_filename ): script_file = open( script_filename, "r", encoding="iso-8859-15" ) script_string = "" for line in script_file: script_string += line # str( line, encoding="iso-8859-15", errors="replace" ) script_file.close( ) return script_string def open_file( self, widget, Data=None ): """ do the open file dialog, if necessary ask for save """ # ignore if not self.editing_state: return 0 # Determining the tab which is currently open current_page = self.main_notebook.get_current_page( ) if current_page == 0: open_dialog_title = "Open Experiment Script..." modified = self.experiment_script_textbuffer.get_modified( ) elif current_page == 1: open_dialog_title = "Open Result Script..." modified = self.data_handling_textbuffer.get_modified( ) else: return 0 if modified: question = gtk.MessageDialog(parent=self.xml_gui.get_object( "main_window"), type=gtk.MessageType.WARNING, flags=gtk.DialogFlags.MODAL, buttons=("_Yes", gtk.ResponseType.YES, "_No", gtk.ResponseType.NO, "_Cancel", gtk.ResponseType.CANCEL)) question.set_markup("The file has been changed. Save changes?") response = question.run() question.destroy() if response == gtk.ResponseType.YES: self.save_file() elif response == gtk.ResponseType.CANCEL: return 0 def response( self, response_id, script_widget ): if response_id == gtk.ResponseType.OK: file_name = dialog.get_filename( ) if file_name is None: return script_filename = os.path.abspath( file_name ) if not os.access( script_filename, os.R_OK ): self.outer_space.show_error_dialog( "File I/O Error", "Cannot read from file %s" % script_filename ) return True script_string = script_widget.load_file_as_unicode( script_filename ) if script_widget.main_notebook.get_current_page( ) == 0: script_widget.exp_script_filename = script_filename script_widget.set_scripts( script_string, None ) elif script_widget.main_notebook.get_current_page( ) == 1: script_widget.res_script_filename = script_filename script_widget.set_scripts( None, script_string ) return True parent_window = self.xml_gui.get_object( "main_window" ) dialog = gtk.FileChooserDialog( title=open_dialog_title, parent=parent_window, action=gtk.FileChooserAction.OPEN, buttons=("_Open", gtk.ResponseType.OK, "_Cancel", gtk.ResponseType.CANCEL) ) dialog.set_default_response( gtk.ResponseType.OK ) dialog.set_select_multiple( False ) # Event-Handler for responce-signal (when one of the button is pressed) dialog.connect( "response", response, self ) dialog.run( ) dialog.destroy( ) # update title and so on... return True def save_file( self, widget=None, Data=None ): """ save file to associated filename """ # ignore if not self.editing_state: return 0 # Determining the tab which is currently open current_page = self.main_notebook.get_current_page( ) if current_page == 0: filename = self.exp_script_filename elif current_page == 1: filename = self.res_script_filename else: return 0 if filename is None: return 0 # save file if current_page == 0: script = self.get_scripts( )[ 0 ] elif current_page == 1: script = self.get_scripts( )[ 1 ] else: return 0 # encode from unicode to iso-8859-15 filecontents = script openfile = open( filename, "w", encoding = "iso-8859-15" ) openfile.write( filecontents ) openfile.close() if current_page == 0: self.experiment_script_textbuffer.set_modified( False ) elif current_page == 1: self.data_handling_textbuffer.set_modified( False ) self.set_toolbuttons_status( ) def save_file_as( self, widget=None, Data=None ): def response( self, response_id, script_widget ): if response_id == gtk.ResponseType.OK: file_name = dialog.get_filename( ) if file_name is None: return True absfilename = os.path.abspath( file_name ) if os.access( file_name, os.F_OK ): question = gtk.MessageDialog(parent=script_widget.xml_gui.get_object( "main_window"), type=gtk.MessageType.WARNING, flags=gtk.DialogFlags.MODAL, buttons=("_Yes", gtk.ResponseType.YES, "_No", gtk.ResponseType.NO)) question.set_markup("The file already exists. Do you want to overwrite the existing file and delete its contents?") response = question.run() question.destroy() if response != gtk.ResponseType.YES: return True current_page = script_widget.main_notebook.get_current_page( ) if current_page == 0: script_widget.exp_script_filename = absfilename elif current_page == 1: script_widget.res_script_filename = absfilename script_widget.save_file( ) return True # Determining the tab which is currently open current_page = self.main_notebook.get_current_page( ) if current_page == 0: dialog_title = "Save Experiment Script As..." elif current_page == 1: dialog_title = "Save Result Script As..." else: return parent_window = self.xml_gui.get_object( "main_window" ) dialog = gtk.FileChooserDialog( title=dialog_title, parent=parent_window, action=gtk.FileChooserAction.SAVE, buttons=("_Save", gtk.ResponseType.OK, "_Cancel", gtk.ResponseType.CANCEL) ) dialog.set_default_response( gtk.ResponseType.OK ) dialog.set_select_multiple( False ) # Event-Handler for responce-signal (when one of the button is pressed) dialog.connect( "response", response, self ) dialog.run( ) dialog.destroy( ) return True def save_all_files( self, widget, Data=None ): current_page = self.main_notebook.get_current_page( ) # change page and call save dialog self.main_notebook.set_current_page( 0 ) if self.exp_script_filename is None: self.save_file_as( ) else: self.save_file( ) self.main_notebook.set_current_page( 1 ) if self.res_script_filename is None: self.save_file_as( ) else: self.save_file( ) self.main_notebook.set_current_page( current_page ) def undo(self, widget=None, Data=None ): if not self.editing_state: return 0 current_page = self.main_notebook.get_current_page( ) if current_page == 0: self.experiment_script_textview.do_undo(self.experiment_script_textview) elif current_page == 1: self.data_handling_textview.do_undo(self.data_handling_textview) self.set_toolbuttons_status() def redo(self, widget=None, Data=None ): if not self.editing_state: return 0 current_page = self.main_notebook.get_current_page( ) if current_page == 0: self.experiment_script_textview.do_redo(self.experiment_script_textview) elif current_page == 1: self.data_handling_textview.do_redo(self.data_handling_textview) self.set_toolbuttons_status() def search(self, widget=None, Data=None): def response(script_widget, response_id, damsw, tb ): if response_id != -1 and response_id != 1: script_widget.destroy() return True current_page = damsw.main_notebook.get_current_page( ) startiter = None enditer = None logview = None logbuffer = None if current_page == 0: if damsw.experiment_script_textbuffer.get_has_selection(): startiter, enditer = damsw.experiment_script_textbuffer.get_selection_bounds() else: mark = damsw.experiment_script_textbuffer.get_insert() startiter = damsw.experiment_script_textbuffer.get_iter_at_mark(mark) enditer = startiter elif current_page == 1: if damsw.data_handling_textbuffer.get_has_selection(): startiter, enditer = damsw.data_handling_textbuffer.get_selection_bounds() else: mark = damsw.data_handling_textbuffer.get_insert() startiter = damsw.data_handling_textbuffer.get_iter_at_mark(mark) enditer = startiter elif current_page == 3: logview = damsw.xml_gui.get_object( "messages_textview" ) logbuffer = logview.get_buffer() if logbuffer.get_has_selection(): startiter, enditer = logbuffer.get_selection_bounds() else: mark = logbuffer.get_insert() startiter = logbuffer.get_iter_at_mark(mark) enditer = startiter else: script_widget.destroy() return True searchstring = tb.get_text() if len(searchstring) > 0: match_start = None match_end = None try: if response_id == -1: match_start, match_end = startiter.backward_search(searchstring, gtk.TextSearchFlags.TEXT_ONLY) elif response_id == 1: match_start, match_end = enditer.forward_search(searchstring, gtk.TextSearchFlags.TEXT_ONLY) except: pass if match_start is not None and match_end is not None: if current_page == 0: damsw.experiment_script_textbuffer.select_range(match_start, match_end) damsw.experiment_script_textview.scroll_to_iter(match_start, 0.2) damsw.textviews_moved(damsw.experiment_script_textview) elif current_page == 1: damsw.data_handling_textbuffer.select_range(match_start, match_end) damsw.data_handling_textview.scroll_to_iter(match_start, 0.2) damsw.textviews_moved(damsw.data_handling_textview) elif current_page == 3: logbuffer.select_range(match_start, match_end) logview.scroll_to_iter(match_start, 0.2) dialog = gtk.Dialog(title="Search", parent=self.xml_gui.get_object("main_window"), flags=0, buttons=("_Close", gtk.ResponseType.CLOSE, "Find last", -1, "Find next", 1)) cont = dialog.get_content_area() markedstring = "" current_page = self.main_notebook.get_current_page( ) startit = None endit = None logbuffer = None if current_page == 0: if self.experiment_script_textbuffer.get_has_selection(): startit, endit = self.experiment_script_textbuffer.get_selection_bounds() markedstring = self.experiment_script_textbuffer.get_text(startit, endit, True) elif current_page == 1: if self.data_handling_textbuffer.get_has_selection(): startit, endit = self.data_handling_textbuffer.get_selection_bounds() markedstring = self.data_handling_textbuffer.get_text(startit, endit, True) elif current_page == 3: logview = self.xml_gui.get_object( "messages_textview" ) logbuffer = logview.get_buffer() if logbuffer.get_has_selection(): startit, endit = logbuffer.get_selection_bounds() markedstring = logbuffer.get_text(startit, endit, True) markedstring = markedstring.strip() textbox = gtk.Entry() textbox.set_text(markedstring) cont.add(textbox) dialog.set_default_response(1) dialog.connect("response", response, self, textbox) dialog.set_resizable(False) dialog.show_all() if startit is not None and endit is not None: if current_page == 0: self.experiment_script_textbuffer.select_range(startit, endit) elif current_page == 1: self.data_handling_textbuffer.select_range(startit, endit) elif current_page == 3: logbuffer.select_range(startit, endit) def new_file( self, widget, Data=None ): if not self.editing_state: return 0 current_page = self.main_notebook.get_current_page( ) if current_page == 0: if self.experiment_script_textbuffer.get_modified( ): question = gtk.MessageDialog(parent=self.xml_gui.get_object( "main_window"), type=gtk.MessageType.WARNING, flags=gtk.DialogFlags.MODAL, buttons=("_Yes", gtk.ResponseType.YES, "_No", gtk.ResponseType.NO, "_Cancel", gtk.ResponseType.CANCEL)) question.set_markup("The file has been changed. Save changes?") response = question.run() question.destroy() if response == gtk.ResponseType.YES: self.save_file() elif response == gtk.ResponseType.CANCEL: return 0 self.set_scripts( "", None ) self.exp_script_filename = None elif current_page == 1: if self.data_handling_textbuffer.get_modified( ): question = gtk.MessageDialog(parent=self.xml_gui.get_object( "main_window"), type=gtk.MessageType.WARNING, flags=gtk.DialogFlags.MODAL, buttons=("_Yes", gtk.ResponseType.YES, "_No", gtk.ResponseType.NO, "_Cancel", gtk.ResponseType.CANCEL)) question.set_markup("The file has been changed. Save changes?") response = question.run() question.destroy() if response == gtk.ResponseType.YES: self.save_file() elif response == gtk.ResponseType.CANCEL: return 0 self.set_scripts( None, "" ) self.res_script_filename = None self.set_toolbuttons_status( ) class ConfigTab: """ by now all values are saved in the GUI widgets """ def __init__( self, xml_gui ): self.xml_gui = xml_gui self.configname = "damaris/python-damaris.xml" self.system_default_filename = None self.system_backend_folder = "/usr/lib/damaris/backends/" if sys.platform[ :5 ] == "linux" or "darwin": xdg_dirs = xdg.BaseDirectory.xdg_config_dirs xdg_dirs.remove( xdg.BaseDirectory.xdg_config_home ) self.user_default_filename = os.path.join( xdg.BaseDirectory.xdg_config_home, self.configname ) self.system_default_filenames = [ os.path.join( syscfg, self.configname ) for syscfg in xdg_dirs ] self.config_start_backend_checkbutton = self.xml_gui.get_object( "start_backend_checkbutton" ) self.config_backend_executable_entry = self.xml_gui.get_object( "backend_executable_entry" ) self.config_spool_dir_entry = self.xml_gui.get_object( "spool_dir_entry" ) self.config_start_experiment_script_checkbutton = self.xml_gui.get_object( "start_experiment_script_checkbutton" ) self.config_start_result_script_checkbutton = self.xml_gui.get_object( "start_result_script_checkbutton" ) self.config_del_results_after_processing_checkbutton = self.xml_gui.get_object( "del_results_after_processing_checkbutton" ) self.config_del_jobs_after_execution_checkbutton = self.xml_gui.get_object( "del_jobs_after_execution_checkbutton" ) self.config_data_pool_name_entry = self.xml_gui.get_object( "data_pool_name_entry" ) self.config_data_pool_name_entry.set_tooltip_text("Use %(date)s for ISO date or %(datetime)s for ISO date+time") self.config_data_pool_write_interval_entry = self.xml_gui.get_object( "data_pool_write_interval_entry" ) self.config_data_pool_complib = self.xml_gui.get_object( "CompLibs" ) self.config_data_pool_complib.set_tooltip_text("Use zlib for best compatibility with other tools") self.config_data_pool_comprate = self.xml_gui.get_object( "CompRatio" ) self.config_data_pool_comprate.set_tooltip_text("Use 1 for best compression, 9 for best speed; 3 is very usable") self.config_info_textview = self.xml_gui.get_object( "info_textview" ) self.config_script_font_button = self.xml_gui.get_object( "script_fontbutton" ) # insert version informations components_text = """ operating system %(os)s gtk version %(gtk)s glib version %(glib)s python version %(python)s matplotlib version %(matplotlib)s, %(matplotlib_backend)s numpy version %(numpy)s pytables version %(pytables)s, using: %(pytables_libs)s pygobject version %(pygobject)s """ if hasattr( gobject, "glib_version" ): glib_version = "%d.%d.%d" % gobject.glib_version else: glib_version = "? (no pygobject module)" if hasattr( gobject, "pygobject_version" ): pygobject_version = "%d.%d.%d" % gobject.pygobject_version else: pygobject_version = "? (no gobject version number)" numpy_version = "none" try: import numpy except ImportError: pass else: numpy_version = numpy.__version__ components_versions = { "os": platform.platform( ), "gtk": "{}.{}.{}".format(gtk.MAJOR_VERSION, gtk.MINOR_VERSION, gtk.MICRO_VERSION), "glib": glib_version, "python": sys.version, "matplotlib": matplotlib.__version__, "matplotlib_backend": FigureCanvas.__name__[ 12: ], "numpy": numpy_version, "pytables": tables.__version__, "pytables_libs": "", "pygobject": pygobject_version } # pytables modules: # find compression extensions for combo box and write version numbers # list is taken from ValueError output of tables.which_lib_version("") model = self.config_data_pool_complib.get_model( ) for libname in ('hdf5', 'zlib', 'lzo', 'blosc', 'bzip2', 'szip'): version_info = None try: version_info = tables.which_lib_version( libname ) except ValueError: continue if version_info: components_versions[ "pytables_libs" ] += "\n * %s: %s" % (libname, str( version_info[1] )) if libname != "hdf5": # a compression library, add it to combo box if isinstance( model, gtk.ListStore ): model.append( [ libname ] ) elif isinstance( model, gtk.TreeStore ): model.append( None, [ libname ] ) else: print("cannot append compression lib name to %s" % model.__class__.__name__) # debug message if debug: print("DAMARIS", __version__) print(components_text % components_versions) # set no compression as default... self.config_data_pool_complib.set_active( 0 ) info_textbuffer = self.config_info_textview.get_buffer( ) info_text = info_textbuffer.get_text( info_textbuffer.get_start_iter( ), info_textbuffer.get_end_iter( ), True ) info_text %= { "moduleversions": components_text % components_versions, "damarisversion": __version__ } info_textbuffer.set_text( info_text ) del info_textbuffer, info_text, components_text, components_versions for self.system_default_filename in self.system_default_filenames: if self.system_default_filename: if os.access( self.system_default_filename, os.R_OK ): self.load_config( self.system_default_filename ) else: print("can not read system defaults from %s, ask your instrument responsible if required" % self.system_default_filename) self.config_from_system = self.get( ) self.load_config( ) def get( self ): """ returns a dictionary of actual config values """ complib_iter = self.config_data_pool_complib.get_active_iter( ) complib = self.config_data_pool_complib.get_model( ).get_value( complib_iter, 0 ) return { "start_backend": self.config_start_backend_checkbutton.get_active( ), "start_result_script": self.config_start_result_script_checkbutton.get_active( ), "start_experiment_script": self.config_start_experiment_script_checkbutton.get_active( ), "spool_dir": self.config_spool_dir_entry.get_text( ), "backend_executable": self.config_backend_executable_entry.get_text( ), "data_pool_name": self.config_data_pool_name_entry.get_text( ), "del_results_after_processing": self.config_del_results_after_processing_checkbutton.get_active( ), "del_jobs_after_execution": self.config_del_jobs_after_execution_checkbutton.get_active( ), "data_pool_write_interval": self.config_data_pool_write_interval_entry.get_text( ), "data_pool_complib": complib, "data_pool_comprate": self.config_data_pool_comprate.get_value_as_int( ), "script_font": self.config_script_font_button.get_font_name( ), "adc_bit_depth": ADC_Result.default_bit_depth } def set( self, config ): if "start_backend" in config: self.config_start_backend_checkbutton.set_active( config[ "start_backend" ] ) if "start_experiment_script" in config: self.config_start_experiment_script_checkbutton.set_active( config[ "start_experiment_script" ] ) if "start_result_script" in config: self.config_start_result_script_checkbutton.set_active( config[ "start_result_script" ] ) if "spool_dir" in config: self.config_spool_dir_entry.set_text( config[ "spool_dir" ] ) if "backend_executable" in config: self.config_backend_executable_entry.set_text( config[ "backend_executable" ] ) if "del_results_after_processing" in config: self.config_del_results_after_processing_checkbutton.set_active( config[ "del_results_after_processing" ] ) if "del_jobs_after_execution" in config: self.config_del_jobs_after_execution_checkbutton.set_active( config[ "del_jobs_after_execution" ] ) if "data_pool_write_interval" in config: self.config_data_pool_write_interval_entry.set_text( config[ "data_pool_write_interval" ] ) if "data_pool_name" in config: self.config_data_pool_name_entry.set_text( config[ "data_pool_name" ] ) if "data_pool_comprate" in config: self.config_data_pool_comprate.set_value( float( config[ "data_pool_comprate" ] ) ) if "script_font" in config: self.config_script_font_button.set_font_name( config[ "script_font" ] ) self.set_script_font_handler( None ) if "adc_bit_depth" in config: try: ADC_Result.default_bit_depth = int(config["adc_bit_depth"]) except (ValueError, TypeError): pass if "data_pool_complib" in config: # find combo-box entry and make it active... model = self.config_data_pool_complib.get_model( ) iter = model.get_iter_first( ) while iter is not None: if model.get( iter, 0 )[ 0 ] == config[ "data_pool_complib" ]: self.config_data_pool_complib.set_active_iter( iter ) break iter = model.iter_next( iter ) # if this compression method is not supported, warn and do nothing if iter is None: print("compression method %s is not supported" % config[ "data_pool_complib" ]) def load_config_handler( self, widget ): if self.system_default_filename: self.load_config( self.system_default_filename ) self.load_config( ) def save_config_handler( self, widget ): self.save_config( ) def set_script_font_handler( self, widget ): """ handles changes in font name also sets the fonts to the text views (fast implementation: breaking encapsulation) """ font = self.config_script_font_button.get_font_name( ) experiment_script_textview = self.xml_gui.get_object( "experiment_script_textview" ) if experiment_script_textview: experiment_script_textview.modify_font( pango.FontDescription( font ) ) data_handling_textview = self.xml_gui.get_object( "data_handling_textview" ) if data_handling_textview: data_handling_textview.modify_font( pango.FontDescription( font ) ) def browse_backend_executable_dialog( self, widget ): """ do the open file dialog """ backend_filename_dialog_title = "find backend" def response( self, response_id, script_widget ): if response_id == gtk.ResponseType.OK: file_name = self.get_filename( ) if file_name is None: return script_widget.config_backend_executable_entry.set_text( file_name ) return True parent_window = self.xml_gui.get_object( "main_window" ) dialog = gtk.FileChooserDialog( title=backend_filename_dialog_title, parent=parent_window, action=gtk.FileChooserAction.OPEN, buttons=("_Open", gtk.ResponseType.OK, "_Cancel", gtk.ResponseType.CANCEL) ) dialog.set_default_response( gtk.ResponseType.OK ) dialog.set_select_multiple( False ) dialog.set_filename( os.path.abspath( self.config_backend_executable_entry.get_text( ) ) ) if os.path.isdir( self.system_backend_folder ) \ and os.access( self.system_backend_folder, os.R_OK ): dialog.add_shortcut_folder( self.system_backend_folder ) # Event-Handler for responce-signal (when one of the button is pressed) dialog.connect( "response", response, self ) #f = gtk.FileFilter( ) #f.add_custom( gtk.FileFilterFlags.FILENAME, lambda x: os.access( x[ 0 ], os.X_OK ) ) #dialog.set_filter( f ) dialog.run( ) dialog.destroy( ) return True def load_config( self, filename=None ): """ set config from an xml file """ if filename is None: filename = self.user_default_filename try: readfile = open( filename, "rb" ) except Exception as e: if debug: print("Could not open %s: %s" % (filename, str( e ))) return # parser functions def start_element( name, attrs, config ): if name == "config" and "key" in attrs: config[ "__this_key__" ] = attrs[ "key" ] if "type" in attrs: config[ "__this_type__" ] = attrs[ "type" ] config[ attrs[ "key" ] ] = "" def end_element( name, config ): if "__this_type__" in config and "__this_key__" in config: if config[ "__this_type__" ] == "Boolean": if config[ config[ "__this_key__" ] ] == "True": config[ config[ "__this_key__" ] ] = True else: config[ config[ "__this_key__" ] ] = False elif config[ "__this_type__" ] == "Integer": config[ config[ "__this_key__" ] ] = int( config[ config[ "__this_key__" ] ] ) if "__this_type__" in config: del config[ "__this_type__" ] if "__this_key__" in config: del config[ "__this_key__" ] def char_data( data, config ): if "__this_key__" in config: config[ config[ "__this_key__" ] ] += data # parse file contents to dictionary config = { } p = xml.parsers.expat.ParserCreate( ) p.StartElementHandler = lambda n, a: start_element( n, a, config ) p.EndElementHandler = lambda n: end_element( n, config ) p.CharacterDataHandler = lambda d: char_data( d, config ) p.ParseFile( readfile ) self.set( config ) def save_config( self, filename=None ): """ write config as an xml file """ config = self.get( ) if filename is None: filename = self.user_default_filename dirs = os.path.dirname( filename ) if not os.path.isdir( dirs ): os.makedirs( dirs ) print("save config to: "+filename) configfile = open( filename, "w" ) configfile.write( "\n" ) configfile.write( "\n" ) for k, v in config.items( ): if k in self.config_from_system \ and self.config_from_system[ k ] == v: if debug: print("Ignoring for write, because system value for %r is %r equal to %r" % \ (k, self.config_from_system[ k ], v)) continue print(k,v, type(v)) val = "" typename = "" if type( v ) is bool: typename = "Boolean" if v: val = "True" else: val = "False" elif type( v ) is bytes: typename = "Bytes" val = v elif type( v ) is str: typename = "String" val = v elif type( v ) is int: typename = "Integer" val = str( v ) configfile.write( " %s\n" % (k, typename, val) ) configfile.write( "\n" ) class MonitorWidgets: def __init__( self, xml_gui ): """ initialize matplotlib widgets and stuff around """ self.xml_gui = xml_gui self.main_window = self.xml_gui.get_object( "main_window" ) self.display_settings_frame = self.xml_gui.get_object( "display_settings_frame" ) # Display footer: self.display_x_scaling_combobox = self.xml_gui.get_object( "display_x_scaling_combobox" ) self.display_x_scaling_combobox.set_sensitive( False ) self.display_y_scaling_combobox = self.xml_gui.get_object( "display_y_scaling_combobox" ) self.display_y_scaling_combobox.set_sensitive( False ) self.display_autoscaling_checkbutton = self.xml_gui.get_object( "display_autoscaling_checkbutton" ) self.display_statistics_checkbutton = self.xml_gui.get_object( "display_statistics_checkbutton" ) self.display_descriptions_checkbutton = gtk.CheckButton(label="Show Desc.") self.display_descriptions_checkbutton.set_active(True) self.display_descriptions_checkbutton.show() self.display_settings_frame.set_property("n-columns", 5) self.display_settings_frame.attach(self.display_descriptions_checkbutton, 4, 5, 0, 1, gtk.AttachOptions.FILL, 0, 0, 0) self.display_descriptions_checkbutton.connect("toggled", self.display_descriptions_toggled) # Matplot (Display_Table, 1st Row) -------------------------------------------------------- # create new plot self.matplot_figure = matplotlib.figure.Figure( ) # the plot area surrounded by axes self.matplot_axes = self.matplot_figure.add_axes( [ 0.1, 0.15, 0.8, 0.7 ] ) self.matplot_axes.grid( True ) self.matplot_axes.set_ylim( [ 0.0, 1.0 ] ) self.matplot_axes.set_xlim( [ 0.0, 1.0 ] ) self.matplot_axes.set_autoscale_on( self.display_autoscaling_checkbutton.get_active( ) ) # linear y/x scaling self.matplot_axes.set_yscale( "linear" ) self.matplot_axes.set_xscale( "linear" ) # add figure to canvas self.matplot_canvas = FigureCanvas( self.matplot_figure ) # add ctrl + right-click context menu to canvas self.matplot_canvas.connect("button-press-event", self.on_canvas_button_press) self.display_table = self.xml_gui.get_object( "display_table" ) self.display_table.attach( self.matplot_canvas, 0, 6, 0, 1, gtk.AttachOptions.EXPAND | gtk.AttachOptions.FILL, gtk.AttachOptions.EXPAND | gtk.AttachOptions.FILL, 0, 0 ) self.matplot_canvas.show( ) # add matplotlib toolbar (display_table, 2nd row) self.matplot_toolbar = matplotlib.backends.backend_gtk3.NavigationToolbar2GTK3( self.matplot_canvas ) # remove subplots: #self.matplot_toolbar.DeleteToolByPos(1) #self.matplot_toolbar.remove(self.matplot_toolbar.get_nth_item(3)) self.matplot_toolbar.remove(self.matplot_toolbar.get_nth_item(6)) # now let's add a button to the toolbar self.matplot_cursor_button = gtk.ToggleButton(label='') try: image = gtk.Image.new_from_file("/usr/share/icons/breeze/actions/24/crosshairs-symbolic.svg") self.matplot_cursor_button.set_image(image) except: pass self.matplot_cursor_button.set_tooltip_text('Enable Crosshairs') self.matplot_cursor_button.show() self.matplot_cursor_button.connect('toggled', self.toggle_cursor) toolitem = gtk.ToolItem() toolitem.add(self.matplot_cursor_button) self.matplot_toolbar.insert(toolitem, 5) # Make cursor position text bigger and bold - access message label directly try: # The coordinate label is stored in the toolbar's message attribute if hasattr(self.matplot_toolbar, 'message'): self.matplot_toolbar.message.modify_font(pango.FontDescription("sans bold 16")) except: pass self.display_table.attach( self.matplot_toolbar, 0, 1, 1, 2, gtk.AttachOptions.FILL | gtk.AttachOptions.EXPAND, 0, 0, 0 ) self.matplot_toolbar.show( ) # select display source self.display_source_combobox = self.xml_gui.get_object( "display_source_combobox" ) self.display_source_treestore = gtk.TreeStore( gobject.TYPE_STRING ) self.display_source_combobox.set_model( self.display_source_treestore ) display_source_cell = gtk.CellRendererText( ) self.display_source_combobox.pack_start( display_source_cell, True ) self.display_source_combobox.add_attribute( display_source_cell, 'text', 0 ) # GTK3 fix: Prevent blank space from child tree nodes in popup self.display_source_combobox.set_wrap_width(1) self.source_list_reset( ) self.display_source_path_label = self.xml_gui.get_object( "display_source_path_label" ) # enable display scaling self.display_x_scaling_combobox.set_active( 0 ) self.display_y_scaling_combobox.set_active( 0 ) for i in ("lin", "log10"): self.display_x_scaling_combobox.append_text(i) self.display_y_scaling_combobox.append_text(i) self.display_x_scaling_combobox.append_text("ppm") self.display_x_scaling_combobox.set_active(0) self.display_y_scaling_combobox.set_active(0) self.display_x_scaling_combobox.set_sensitive( True ) self.display_y_scaling_combobox.set_sensitive( True ) # and events... self.display_source_combobox.connect( "changed", self.display_source_changed_event ) # data to observe self.data_pool = None # name of displayed data and reference to data self.displayed_data = [ None, None ] self.__rescale = True self.update_counter = 0 self.update_counter_lock = threading.Lock( ) self.description_text = None self.clipping_marker = None def source_list_reset( self ): self.display_source_treestore.clear( ) self.source_list_add( 'None' ) none_iter = self.source_list_find( [ 'None' ] ) if none_iter is not None: self.display_source_combobox.set_active_iter( none_iter ) def source_list_find_one( self, model, iter, what ): """find node in subcategory""" while iter is not None: if model.get( iter, 0 )[ 0 ] == what: return iter iter = model.iter_next( iter ) return iter def source_list_find( self, namelist ): """ namelist sequence of names, e.g. ["a", "b", "c" ] for "a/b/c" """ model = self.display_source_treestore retval = None #iter = model.get_iter_root( ) iter = model.get_iter_first( ) while iter is not None and len( namelist ) > 0: name = namelist[ 0 ] iter = self.source_list_find_one( model, iter, name ) if iter is not None: retval = iter namelist.pop( 0 ) iter = model.iter_children( retval ) return retval def source_list_add( self, source_name, parent=None ): namelist = source_name.split( "/" ) found = self.source_list_find( namelist ) if parent is None: parent = found for rest_name in namelist: # append() returns iter for the new row parent = self.display_source_treestore.append( parent, [ rest_name ] ) def source_list_remove( self, source_name ): namelist = source_name.split( "/" ) pwd = namelist[ : ] iter = self.source_list_find( namelist ) if iter is None or len( namelist ) > 0: print("source_list_remove: WARNING: Not found") return model = self.display_source_treestore if model.iter_has_child( iter ): print("source_list_remove: WARNING: Request to delete a tree") return while True: parent = model.iter_parent( iter ) model.remove( iter ) pwd.pop( ) # We now test, if we want to remove parent too if parent is None: break if model.iter_has_child( parent ): # The parent has other children break if "/".join( pwd ) in self.data_pool: # The parent has data connected to it break iter = parent def source_list_current( self ): ai = self.display_source_combobox.get_active_iter( ) namelist = [ ] while ai is not None: namelist.insert( 0, str( self.display_source_treestore.get( ai, 0 )[ 0 ] ) ) ai = self.display_source_treestore.iter_parent( ai ) cur_source_name = "/".join( namelist ) return cur_source_name def observe_data_pool( self, data_pool ): """ register a listener and save reference to data assume to be in gtk/gdk lock """ if self.data_pool is not None: # maybe some extra cleanup needed print("ToDo: cleanup widgets") if self.displayed_data[ 1 ] is not None and hasattr( self.displayed_data[ 1 ], "unregister_listener" ): self.displayed_data[ 1 ].unregister_listener( self.datastructures_listener ) self.displayed_data[ 1 ] = None self.displayed_data = [ None, None ] self.data_pool.unregister_listener( self.datapool_listener ) self.data_pool = None self.source_list_reset( ) self.update_counter_lock.acquire( ) self.update_counter = 0 self.update_counter_lock.release( ) # display states self.__rescale = True self.displayed_data = [ None, None ] self.display_source_path_label.set_label( "" ) self.clear_display( ) if data_pool is not None: # keep track of data self.data_pool = data_pool self.data_pool.register_listener( self.datapool_listener ) #################### observing data structures and produce idle events def datapool_listener( self, event ): """ sort data as fast as possible and get rid of non-interesting data """ if event.subject.startswith( "__" ): return if debug and self.update_counter < 0: print("negative event count!", self.update_counter) if self.update_counter > 5: if debug: print("sleeping to find time for grapics updates") threading.Event( ).wait( 0.05 ) while self.update_counter > 15: threading.Event( ).wait( 0.05 ) if event.what == DataPool.Event.updated_value: if self.displayed_data[ 0 ] is None or event.subject != self.displayed_data[ 0 ]: # do nothing, forget it return displayed_object = self.displayed_data[ 1 ] object_to_display = self.data_pool.get( event.subject ) if displayed_object is None or object_to_display is None: self.update_counter_lock.acquire( ) self.update_counter += 1 self.update_counter_lock.release( ) gobject.idle_add( self.datapool_idle_listener, event, priority=gobject.PRIORITY_DEFAULT_IDLE ) else: if event.what == DataPool.Event.updated_value and \ ( displayed_object is object_to_display or displayed_object.__class__ is object_to_display.__class__): # oh, another category self.update_counter_lock.acquire() self.update_counter += 1 self.update_counter_lock.release() gobject.idle_add( self.update_display_idle_event, self.displayed_data[ 0 ][ : ], priority=gobject.PRIORITY_DEFAULT_IDLE ) else: self.update_counter_lock.acquire( ) self.update_counter += 1 self.update_counter_lock.release( ) gobject.idle_add( self.datapool_idle_listener, event, priority=gobject.PRIORITY_DEFAULT_IDLE ) if event.what in [ DataPool.Event.updated_value, DataPool.Event.new_key ]: #print "Update hdf5 file ..." # TODO: incremental hdf5 update pass del displayed_object del object_to_display def datastructures_listener( self, event ): """ do fast work selecting important events """ if debug and self.update_counter < 0: print("negative event count!", self.update_counter) if self.update_counter > 5: if debug: print("sleeping to find time for graphics updates") threading.Event( ).wait( 0.05 ) while self.update_counter > 15: threading.Event( ).wait( 0.05 ) if event.origin is not self.displayed_data[ 1 ]: return self.update_counter_lock.acquire( ) self.update_counter += 1 self.update_counter_lock.release( ) gobject.idle_add( self.update_display_idle_event, self.displayed_data[ 0 ][ : ], priority=gobject.PRIORITY_DEFAULT_IDLE ) ################### consume idle events def datapool_idle_listener( self, event ): """ here dictionary changes are done """ self.update_counter_lock.acquire( ) self.update_counter -= 1 self.update_counter_lock.release( ) if event.what == DataPool.Event.updated_value: if (self.displayed_data[ 0 ] is not None and self.displayed_data[ 0 ] == event.subject): new_data_struct = self.data_pool[ self.displayed_data[ 0 ] ] if self.displayed_data[ 1 ] is new_data_struct: # update display only if self.update_counter > 10: print("update queue too long (%d>10): skipping one update" % self.update_counter) else: gdk.threads_enter( ) try: self.update_display( ) finally: gdk.threads_leave( ) else: # unregister old one if self.displayed_data[ 1 ] is not None and hasattr( self.displayed_data[ 1 ], "unregister_listener" ): self.displayed_data[ 1 ].unregister_listener( self.datastructures_listener ) self.displayed_data[ 1 ] = None # register new one if hasattr( new_data_struct, "register_listener" ): new_data_struct.register_listener( self.datastructures_listener ) self.displayed_data[ 1 ] = new_data_struct if self.update_counter > 10: print("update queue too long (%d>10): skipping one update" % self.update_counter) else: gdk.threads_enter( ) try: self.renew_display( ) finally: gdk.threads_leave( ) new_data_struct = None elif event.what == DataPool.Event.new_key: # update combo-box by inserting and rely on consistent information gdk.threads_enter( ) self.source_list_add( event.subject ) gdk.threads_leave( ) elif event.what == DataPool.Event.deleted_key: # update combo-box by removing and rely on consistent information gdk.threads_enter( ) if (self.displayed_data[ 0 ] is not None and self.displayed_data[ 0 ] == event.subject): self.displayed_data = [ None, None ] none_iter = self.source_list_find( [ 'None' ] ) if none_iter is not None: self.display_source_combobox.set_active_iter( none_iter ) # not necessary, because event will be submitted #self.clear_display() self.source_list_remove( event.subject ) gdk.threads_leave( ) elif event.what == DataPool.Event.destroy: gdk.threads_enter( ) self.source_list_reset( 'None' ) self.displayed_data = [ None, None ] self.clear_display( ) gdk.threads_leave( ) return def update_display_idle_event( self, subject=None ): self.update_counter_lock.acquire( ) self.update_counter -= 1 self.update_counter_lock.release( ) # print "update display", self.update_counter if self.update_counter > 10: print("update queue too long (%d>10): skipping one update" % self.update_counter) return if self.displayed_data[ 0 ] is None or subject != self.displayed_data[ 0 ]: return gdk.threads_enter( ) try: self.update_display( ) finally: gdk.threads_leave( ) ######################## events from buttons def display_source_changed_event( self, widget, data=None ): new_data_name = self.source_list_current( ) if (self.displayed_data[ 0 ] is None and new_data_name == "None"): return if (self.displayed_data[ 0 ] == new_data_name): return if self.displayed_data[ 1 ] is not None and hasattr( self.displayed_data[ 1 ], "unregister_listener" ): self.displayed_data[ 1 ].unregister_listener( self.datastructures_listener ) self.displayed_data[ 1 ] = None # register new one if new_data_name == "None": self.display_source_path_label.set_label( "" ) self.displayed_data = [ None, None ] self.clear_display( ) elif self.data_pool is None or new_data_name not in self.data_pool: none_iter = self.source_list_find( [ 'None' ] ) if none_iter is not None: self.display_source_combobox.set_active_iter( none_iter ) self.display_source_path_label.set_label( "" ) else: new_data_struct = self.data_pool[ new_data_name ] if hasattr( new_data_struct, "register_listener" ): new_data_struct.register_listener( self.datastructures_listener ) self.displayed_data = [ new_data_name, new_data_struct ] dirpart = new_data_name.rfind( "/" ) if dirpart >= 0: self.display_source_path_label.set_label( "in " + new_data_name[ :dirpart ] ) else: self.display_source_path_label.set_label( "" ) self.clear_display( ) # renew display via idle event self.update_counter_lock.acquire( ) self.update_counter += 1 self.update_counter_lock.release( ) event = DataPool.Event( DataPool.Event.updated_value, new_data_name ) gobject.idle_add( self.datapool_idle_listener, event, priority=gobject.PRIORITY_DEFAULT_IDLE ) def display_autoscaling_toggled( self, widget, data=None ): self.matplot_axes.set_autoscale_on( self.display_autoscaling_checkbutton.get_active( ) ) if self.displayed_data[ 0 ] is not None: self.update_display( self.displayed_data[ 0 ][ : ] ) def display_scaling_changed( self, widget, data=None ): self.__rescale = True if self.displayed_data[ 0 ] is not None: self.update_display( self.displayed_data[ 0 ][ : ] ) def display_statistics_toggled( self, widget, data=None ): if self.displayed_data[ 0 ] is not None: self.update_display( self.displayed_data[ 0 ][ : ] ) def display_descriptions_toggled( self, widget, data=None ): if self.displayed_data[ 0 ] is not None: self.update_display( self.displayed_data[ 0 ][ : ] ) def save_display_data_as_text( self, widget, data=None ): """ copy data to tmp file and show save dialog """ data_to_save = self.displayed_data[ : ] if self.displayed_data[ 1 ] is None: # nothing to save return if not hasattr( data_to_save[ 1 ], "write_to_csv" ): log( "do not know how to save %s of class/type %s" % (data_to_save[ 0 ], type( data_to_save[ 1 ] )) ) return # save them to a temporary file (in memory) tmpdata = tempfile.TemporaryFile(mode="w+" ) tmpdata.write("# saved from monitor as %s\n" % data_to_save[ 0 ] ) data_to_save[ 1 ].write_to_csv( tmpdata ) parentfordialog = self.main_window # show save dialog def response( self, response_id, tmpfile ): if response_id == gtk.ResponseType.OK: file_name = dialog.get_filename( ) if file_name is None: return True absfilename = os.path.abspath( file_name ) if os.access( file_name, os.F_OK ): question = gtk.MessageDialog(parent=parentfordialog, type=gtk.MessageType.WARNING, flags=gtk.DialogFlags.MODAL, buttons=("_Yes", gtk.ResponseType.YES, "_No", gtk.ResponseType.NO)) question.set_markup("The file already exists. Do you want to overwrite the existing file and delete its contents?") response = question.run() question.destroy() if response != gtk.ResponseType.YES: return True textfile = open( absfilename, "w" ) tmpfile.seek( 0 ) for l in tmpfile: textfile.write( l ) textfile.close( ) textfile = None tmpfile = None return True # Determining the tab which is currently open dialog_title = "Save %s in file" % data_to_save[ 0 ] dialog = gtk.FileChooserDialog( title=dialog_title, parent=self.main_window, action=gtk.FileChooserAction.SAVE, buttons=("_Save", gtk.ResponseType.OK, "_Cancel", gtk.ResponseType.CANCEL) ) dialog.set_default_response( gtk.ResponseType.OK ) dialog.set_current_name( data_to_save[ 0 ] ) dialog.set_select_multiple( False ) # Event-Handler for responce-signal (when one of the button is pressed) dialog.connect( "response", response, tmpdata ) del tmpdata, data_to_save dialog.run( ) dialog.destroy( ) return True def toggle_cursor(self, widget): """Toggle matplotlib crosshair cursor""" if widget.get_active(): # Activate cursor if not hasattr(self, 'matplot_cursor') or self.matplot_cursor is None: self.matplot_cursor = matplotlib.widgets.Cursor(self.matplot_axes, useblit=False, color='red', linewidth=1) self.matplot_canvas.draw() else: # Deactivate cursor if hasattr(self, 'matplot_cursor') and self.matplot_cursor is not None: self.matplot_cursor.disconnect_events() # Remove the lines from the plot if hasattr(self.matplot_cursor, 'lineh'): self.matplot_cursor.lineh.remove() if hasattr(self.matplot_cursor, 'linev'): self.matplot_cursor.linev.remove() self.matplot_cursor = None self.matplot_canvas.draw() def on_canvas_button_press(self, widget, event): """Handle button press events on matplotlib canvas""" if event.button == 2: # middle click self.show_canvas_context_menu(event) return True return False def show_canvas_context_menu(self, event): """Show context menu for matplotlib canvas""" menu = gtk.Menu() # toggle cursor toggle_cursor = gtk.CheckMenuItem(label="Crosshair") # Set the current state based on whether cursor exists if hasattr(self, 'matplot_cursor') and self.matplot_cursor is not None: toggle_cursor.set_active(True) else: toggle_cursor.set_active(False) toggle_cursor.connect("toggled", lambda w: self.matplot_cursor_button.set_active(w.get_active())) menu.append(toggle_cursor) # Copy as PNG item_copy_png = gtk.MenuItem(label="Copy as PNG") item_copy_png.connect("activate", self.copy_canvas_as_png) menu.append(item_copy_png) # Copy as SVG item_copy_svg = gtk.MenuItem(label="Copy as SVG") item_copy_svg.connect("activate", self.copy_canvas_as_svg) menu.append(item_copy_svg) menu.show_all() menu.popup(None, None, None, None, event.button, event.time) def copy_canvas_as_png(self, widget): """Copy canvas to clipboard as PNG""" self.copy_display_data_to_clipboard(widget) def copy_canvas_as_svg(self, widget): """Copy canvas as SVG to clipboard""" if self.matplot_figure is None: return try: # Save the original figure size original_size = self.matplot_figure.get_size_inches() # Save figure to a buffer as SVG import io as iomodule buf = iomodule.StringIO() self.matplot_figure.set_size_inches(6, 4) self.matplot_figure.savefig(buf, format='svg', bbox_inches='tight') buf.seek(0) # Copy SVG text to clipboard clipboard = gtk.Clipboard.get(gdk.SELECTION_CLIPBOARD) clipboard.set_text(buf.read(), -1) clipboard.store() self.matplot_figure.set_size_inches(original_size) log("Figure copied to clipboard as SVG") except Exception as e: log(f"Error copying figure as SVG: {e}") def copy_display_data_to_clipboard( self, widget, data=None ): """ Copy matplotlib figure as PNG image to clipboard Also keep text data copy functionality as fallback """ # First, try to copy the matplotlib figure as image # Save the original figure size original_size = self.matplot_figure.get_size_inches() if self.matplot_figure is not None and self.matplot_canvas is not None: try: # Save figure to a temporary buffer as PNG import io as iomodule buf = iomodule.BytesIO() self.matplot_figure.set_size_inches(6, 4) self.matplot_figure.savefig(buf, format='png', dpi=300, bbox_inches='tight') buf.seek(0) # Load the PNG data into a GdkPixbuf from gi.repository import GdkPixbuf loader = GdkPixbuf.PixbufLoader.new_with_type('png') loader.write(buf.read()) loader.close() pixbuf = loader.get_pixbuf() # Copy to clipboard clipboard = gtk.Clipboard.get(gdk.SELECTION_CLIPBOARD) clipboard.set_image(pixbuf) clipboard.store() # Restore original figure size self.matplot_figure.set_size_inches(original_size) # Redraw the canvas to ensure it stays visible self.matplot_canvas.draw() log("Figure copied to clipboard as PNG image") return except Exception as e: log("Error copying figure to clipboard: %s" % str(e)) ##################### functions to feed display def clear_display( self ): """ unconditionally throw away everything we are inside gtk/gdk lock """ self.display_x_scaling_combobox.set_sensitive( True ) self.display_y_scaling_combobox.set_sensitive( True ) x_scale = self.display_x_scaling_combobox.get_active_text() y_scale = self.display_y_scaling_combobox.get_active_text() print(x_scale) if x_scale == "lin": self.matplot_axes.set_xscale("linear") elif x_scale == "log10": self.matplot_axes.set_xscale("log", base=10, nonpositive="mask") elif x_scale == "ppm": self.matplot_axes.set_xscale("linear") self.matplot_axes.invert_xaxis() if y_scale == "lin": self.matplot_axes.set_yscale("linear") elif y_scale == "log10": self.matplot_axes.set_yscale("log", base=10, nonpositive="mask") if not hasattr( self, "__rescale" ): self.__rescale = True if not hasattr( self, "measurementresultline" ): self.measurementresultline = None elif self.measurementresultline is not None: # clear line plot self.matplot_axes.lines.remove( self.measurementresultline[ 0 ] ) self.measurementresultline = None if hasattr(self, 'matplot_cursor') and self.matplot_cursor is not None: self.matplot_cursor.disconnect_events() self.matplot_cursor = None if not hasattr( self, "measurementresultgraph" ): self.measurementresultgraph = None elif self.measurementresultgraph is not None: # clear errorbars self.measurementresultgraph[0].remove() for line in self.measurementresultgraph[ 1 ]: line.remove() for line in self.measurementresultgraph[ 2 ]: line.remove() self.measurementresultgraph = None self.matplot_axes.clear( ) self.matplot_axes.grid( True ) if not hasattr( self, "graphen" ): self.graphen = [ ] elif self.graphen: for line in self.graphen: line.remove() self.graphen = [ ] self.matplot_axes.clear( ) self.matplot_axes.grid( True ) if self.description_text is not None: try: self.description_text.remove() except: pass self.description_text = None if self.clipping_marker is not None: try: self.clipping_marker.remove() except: pass self.clipping_marker = None self.matplot_canvas.draw_idle( ) def update_display( self, subject=None ): """ try to recycle labels, data, lines.... assume, object is not changed we are inside gtk/gdk lock """ in_result = self.data_pool.get( self.displayed_data[ 0 ] ) if in_result is None: self.clear_display( ) return if isinstance( in_result, Accumulation ) or isinstance( in_result, ADC_Result ): xmin = in_result.get_xmin( ) xmax = in_result.get_xmax( ) ymin = in_result.get_ymin( ) ymax = in_result.get_ymax( ) # Initial rescaling needed? if self.__rescale: x_scale = self.display_x_scaling_combobox.get_active_text( ) y_scale = self.display_y_scaling_combobox.get_active_text( ) # x-axis if x_scale == "lin": self.matplot_axes.set_xscale( "linear" ) self.matplot_axes.set_xlim(xmin, xmax) elif x_scale == "log10": xminpos = in_result.get_xminpos( ) self.matplot_axes.set_xscale("log", base=10, nonpositive="mask") self.matplot_axes.set_xlim(xminpos, xmax) elif x_scale == "ppm": self.matplot_axes.set_xscale("linear") self.matplot_axes.set_xlabel("PPM") #self.display_autoscaling_checkbutton.set_active(True) #partial fix T141 self.matplot_axes.set_xlim(xmax, xmin) #self.matplot_axes.invert_xaxis() # y-axis if y_scale == "lin": self.matplot_axes.set_yscale( "linear" ) self.matplot_axes.set_ylim(ymin, ymax) elif y_scale == "log10": yminpos = in_result.get_yminpos() self.matplot_axes.set_yscale("log", base=10, nonpositive="mask") self.matplot_axes.set_ylim( yminpos, ymax ) self.__rescale = False # Autoscaling activated? elif self.display_autoscaling_checkbutton.get_active(): xlim_min, xlim_max = self.matplot_axes.get_xlim() if xlim_min != xmin or xlim_max != xmax: self.matplot_axes.set_xlim(xmin, xmax) ylim_min, ylim_max = self.matplot_axes.get_ylim() # Check if y-axis is log scale yscale = self.matplot_axes.get_yscale() if yscale == 'log' or yscale == 'symlog': # For log scales, work with log values to avoid issues with # small numbers (undetected rescale) # Use log-space calculations log_ymin, log_ymax = numpy.log10(ymin), numpy.log10(ymax) log_diff = log_ymax - log_ymin # Add a small margin in log space (about 0.08 in log10 units ~20% margin) new_ymin = 10**(log_ymin - 0.08 * max(log_diff, 1)) new_ymax = 10**(log_ymax + 0.08 * max(log_diff, 1)) # Only update if values would change significantly if (ylim_max < ymax or ylim_min > ymin or ylim_max > new_ymax * 1.2 or ylim_min < new_ymin / 1.2): self.matplot_axes.set_ylim(new_ymin, new_ymax) else: # Original linear scale logic ydiff = ymax - ymin if (ylim_max < ymax or ylim_min > ymin or ylim_max > ymax + 0.2 * ydiff or ylim_min < ymin - 0.2 * ydiff): self.matplot_axes.set_ylim(ymin, ymax) xdata = in_result.get_xdata( ) chans = in_result.get_number_of_channels( ) data = [ ] colors = [ (0, 0, 0.8, 1), (0.7, 0, 0, 1), (0, 0.7, 0, 1), (0.7, 0.5, 0, 1), (0, 0, 0, 1) ] # rgba tuples: blue, red, green, yellow for i in range( chans ): data.append( in_result.get_ydata( i ) ) moving_average = data_slice = None if max_points_to_display > 0 and len( xdata ) > max_points_to_display: print("decimating data to %d points by moving average (prevent crash of matplotlib)" % max_points_to_display) n = numpy.ceil( len( xdata ) / max_points_to_display ) moving_average = numpy.ones( n, dtype="float" ) / n data_slice = numpy.array( numpy.floor( numpy.arange( max_points_to_display, dtype="float" ) \ / max_points_to_display * len( xdata ) ), dtype="int" ) xdata = xdata.take( data_slice ) # no average !? for i in range( chans ): data[ i ] = numpy.convolve( data[ i ], moving_average, "same" ).take( data_slice ) if len( self.graphen ) == 0: for i in range( chans ): self.graphen.extend( self.matplot_axes.plot( xdata, data[ i ], linestyle="-", color=colors[ i ], linewidth=2 ) ) for i in range( chans ): # initialize error bars self.graphen.extend( self.matplot_axes.plot( [ 0.0 ], [ 0.0 ], linestyle="-", color=colors[ i ], linewidth=0.5 ) ) self.graphen.extend( self.matplot_axes.plot( [ 0.0 ], [ 0.0 ], linestyle="-", color=colors[ i ], linewidth=0.5 ) ) else: for i in range( chans ): self.graphen[ i ].set_data( xdata, data[ i ] ) # Statistics activated? if (self.display_statistics_checkbutton.get_active( ) and in_result.uses_statistics( ) and in_result.ready_for_drawing_error( )): for i in range( chans ): err = in_result.get_yerr( i ) if moving_average is not None: err = numpy.convolve( err, moving_average, "same" ).take( data_slice ) self.graphen[ chans + 2 * i ].set_data( xdata, data[ i ] + err ) self.graphen[ chans + 2 * i + 1 ].set_data( xdata, data[ i ] - err ) else: for i in range( chans ): self.graphen[ chans + 2 * i ].set_data( [ 0.0 ], [ 0.0 ] ) self.graphen[ chans + 2 * i + 1 ].set_data( [ 0.0 ], [ 0.0 ] ) moving_average = data_mask = None data = None # Any title to be set? in_result_title = in_result.get_title( ) if in_result_title is not None: col = 101 while len( in_result_title ) - col > 10: in_result_title = in_result_title[ :col ] + "\n" + in_result_title[ col: ] col += 101 self.matplot_axes.set_title( in_result_title ) else: self.matplot_axes.set_title( "" ) # Any labels to be set? if in_result.get_xlabel( ) is not None: self.matplot_axes.set_xlabel( in_result.get_xlabel( ) ) else: self.matplot_axes.set_xlabel( "" ) if in_result.get_ylabel( ) is not None: self.matplot_axes.set_ylabel( in_result.get_ylabel( ) ) else: self.matplot_axes.set_ylabel( "" ) # Any variables to be drawn on plot description_lines = [] if self.display_descriptions_checkbutton.get_active(): if isinstance(in_result, Accumulation): description_lines.append("n = %d" % in_result.n) if in_result.common_descriptions is not None: description_lines.extend(["%s = %s" % (key, in_result.common_descriptions[key]) for key in in_result.common_descriptions.keys()]) elif isinstance(in_result, ADC_Result): if hasattr(in_result, "job_id") and in_result.job_id is not None: description_lines.append("job_id = %s" % str(in_result.job_id)) if hasattr(in_result, "description") and in_result.description is not None: description_lines.extend(["%s = %s" % (key, in_result.description[key]) for key in in_result.description.keys()]) if description_lines: description_string = "\n".join(description_lines) if self.description_text is None: self.description_text = self.matplot_axes.text(0.98, 0.95, description_string, size=10, transform=self.matplot_axes.transAxes, va='top', ha='right', ma='left', bbox=dict(facecolor='lightgray', alpha=0.5, boxstyle='round,pad=0.2')) else: self.description_text.set_text(description_string) elif self.description_text is not None: try: self.description_text.remove() except: pass self.description_text = None if getattr(in_result, "is_clipped", False): if self.clipping_marker is None: self.clipping_marker = self.matplot_axes.text(0.5, 0.95, "ADC OVERFLOW", transform=self.matplot_axes.transAxes, color="red", fontsize=20, fontweight="bold", ha="center", va="top", bbox=dict(facecolor='white', alpha=0.7, edgecolor='red')) elif self.clipping_marker is not None: try: self.clipping_marker.remove() except: pass self.clipping_marker = None # Draw it! self.matplot_canvas.draw_idle( ) del in_result elif isinstance( in_result, MeasurementResult ): # remove lines and error bars if self.measurementresultgraph is not None: self.matplot_axes.lines.remove( self.measurementresultgraph[ 0 ] ) # remove caps for l in self.measurementresultgraph[ 1 ]: self.matplot_axes.lines.remove( l ) # and columns for l in self.measurementresultgraph[ 2 ]: self.matplot_axes.collections.remove( l ) self.measurementresultgraph = None if self.measurementresultline is not None: self.matplot_axes.lines.remove( self.measurementresultline[ 0 ] ) self.measurementresultline = None [ k, v, e ] = in_result.get_errorplotdata( ) if k.shape[ 0 ] != 0: xmin = k.min( ) ymin = (v - e).min( ) if xmin > 0: self.display_x_scaling_combobox.set_sensitive( True ) else: # force switch to lin scale self.display_x_scaling_combobox.set_sensitive( False ) if self.display_x_scaling_combobox.get_active_text( ) != "lin": self.__rescale = True # and reset to linear self.display_x_scaling_combobox.set_active( 0 ) if ymin > 0: self.display_y_scaling_combobox.set_sensitive( True ) else: # force switch to lin scale self.display_y_scaling_combobox.set_sensitive( False ) if self.display_y_scaling_combobox.get_active_text( ) != "lin": self.__rescale = True # and reset to linear self.display_y_scaling_combobox.set_active( 0 ) x_scale = self.display_x_scaling_combobox.get_active_text( ) y_scale = self.display_y_scaling_combobox.get_active_text( ) # Initial rescaling needed? if self.__rescale: # or self.display_autoscaling_checkbutton.get_active(): xmax = k.max( ) ymax = (v + e).max( ) # is there a range problem? if xmin == xmax or ymin == ymax: # fix range and scaling problems if xmin == xmax: if xmin != 0: (xmin, xmax) = (xmin - abs( xmin / 10.0 ), xmin + abs( xmin / 10.0 )) else: (xmin, xmax) = (-1, 1) if ymin == ymax: if ymin == 0: (ymin, ymax) = (ymin - abs( ymin / 10.0 ), ymin + abs( ymin / 10.0 )) else: (ymin, ymax) = (-1, 1) if xmin <= 0 or x_scale == "lin": self.matplot_axes.set_xscale( "linear" ) if ymin <= 0 or y_scale == "lin": self.matplot_axes.set_yscale( "linear" ) self.matplot_axes.set_xlim( xmin, xmax ) self.matplot_axes.set_ylim( ymin, ymax ) # finally decide about x log plot if x_scale == "log10" and xmin > 0: self.matplot_axes.set_xscale( "log", basex=10.0 ) self.matplot_axes.fmt_xdata = lambda x: "%g" % x #elif x_scale=="log" and xmin>0: # e scaling implementation not really useful # self.matplot_axes.set_xscale("log", basex=numpy.e) if y_scale == "log10" and ymin > 0: self.matplot_axes.set_yscale( "log", basex=10.0 ) self.matplot_axes.fmt_ydata = lambda x: "%g" % x self.__rescale = False self.measurementresultgraph = self.matplot_axes.errorbar( x=k, y=v, yerr=e, fmt="bx" ) [ k, v ] = in_result.get_lineplotdata( ) if k.shape[ 0 ] != 0 and v.shape == k.shape: self.measurementresultline = self.matplot_axes.plot( k, v, 'r-' ) # Any title to be set? title = in_result.get_title( ) + "" if title is not None: self.matplot_axes.set_title( title ) else: self.matplot_axes.set_title( "" ) self.matplot_canvas.draw_idle( ) del k, v, e del in_result def renew_display( self ): """ set all properties of display we are inside gtk/gdk lock """ self.clear_display( ) to_draw = self.data_pool[ self.displayed_data[ 0 ] ] if to_draw is None: return self.update_display( ) class ScriptInterface: """ texts or code objects are executed as experiment and result script the backend is started with sufficient arguments """ def __init__( self, exp_script=None, res_script=None, backend_executable=None, spool_dir="spool", clear_jobs=True, clear_results=True ): """ run experiment scripts and result scripts """ self.exp_script = exp_script self.res_script = res_script self.backend_executable = str( backend_executable ) self.spool_dir = os.path.abspath( spool_dir ) self.clear_jobs = clear_jobs self.clear_results = clear_results self.exp_handling = self.res_handling = None self.exp_writer = self.res_reader = self.back_driver = None if self.backend_executable is not None and self.backend_executable != "": self.back_driver = BackendDriver.BackendDriver( self.backend_executable, spool_dir, clear_jobs, clear_results ) if self.exp_script: self.exp_writer = self.back_driver.get_exp_writer( ) if self.res_script: self.res_reader = self.back_driver.get_res_reader( ) elif self.exp_script and self.res_script: self.back_driver = None self.res_reader = ResultReader.BlockingResultReader( spool_dir, clear_jobs=self.clear_jobs, clear_results=self.clear_results ) self.exp_writer = ExperimentWriter.ExperimentWriter( spool_dir, inform_last_job=self.res_reader ) else: self.back_driver = None if self.exp_script: self.exp_writer = ExperimentWriter.ExperimentWriter( spool_dir ) if self.res_script: self.res_reader = ResultReader.ResultReader( spool_dir, clear_jobs=self.clear_jobs, clear_results=self.clear_results ) self.data = DataPool( ) def runScripts( self ): try: # get script engines self.exp_handling = self.res_handling = None if self.exp_script and self.exp_writer: self.exp_handling = ExperimentHandling.ExperimentHandling( self.exp_script, self.exp_writer, self.data ) if self.res_script and self.res_reader: self.res_handling = ResultHandling.ResultHandling( self.res_script, self.res_reader, self.data ) # start them if self.back_driver is not None: self.back_driver.start( ) while (not self.back_driver.quit_flag.isSet( ) and \ (self.back_driver.core_pid is None or self.back_driver.core_pid <= 0)): while gtk.events_pending(): gtk.main_iteration_do(False) self.back_driver.quit_flag.wait( 0.1 ) if self.exp_handling: self.exp_handling.start( ) if self.res_handling: self.res_handling.start( ) finally: self.exp_writer = self.res_reader = None def __del__( self ): self.exp_writer = None self.res_reader = None self.back_driver = None self.data = None self.exp_handling = None self.res_handling = None