Package VisionEgg :: Package PyroApps :: Module EPhysGUIUtils |
[frames] | no frames]

# Source Code for Module VisionEgg.PyroApps.EPhysGUIUtils

      1  #!/usr/bin/env python
      2  #
      3  # The Vision Egg: EPhysGUIUtils
      4  #
      5  # Copyright (C) 2001-2003 Andrew Straw.
      6  # Author: Andrew Straw <astraw@users.sourceforge.net>
      7  # URL: <http://www.visionegg.org/>
      8  #
      9  # Distributed under the terms of the GNU Lesser General Public License
     10  # (LGPL). See LICENSE.TXT that came with this file.
     11  #
     12  # $Id$
     13
     14  import sys, os, time, types, socket
     15  import Tkinter
     16  import Pyro.core
     17
     18  import VisionEgg.PyroClient
     19
     20  __version__ = VisionEgg.release_name
     21  __cvs__ = '$Revision$'.split()[1]
     22  __date__ = ' '.join('$Date$'.split()[1:3])
     23  __author__ = 'Andrew Straw <astraw@users.sourceforge.net>'
     24
     25 -class StimulusControlFrame(Tkinter.Frame):
     26 -    def __init__(self,
     27                   master=None,
     28                   suppress_go_buttons=0,
     29                   title="Stimulus Control",
     30                   meta_params_class=None,
     31                   **kw):
     32          Tkinter.Frame.__init__(self,master,**kw)
     33          self.pyro_client = None
     34          self.entry_width = 10
     35          self.connected = 0
     36          if meta_params_class is not None:
     37              self.meta_params = meta_params_class()
     38          self.loopable_variables = {}
     39
     40          Tkinter.Label(self,
     41                        text=title,
     42                        font=("Helvetica",12,"bold")).pack(expand=Tkinter.YES,
     43                                                           fill=Tkinter.X)
     44
     45          if not suppress_go_buttons:
     46              connected_frame = Tkinter.Frame(self)
     47              connected_frame.pack(expand=Tkinter.YES,
     48                                   fill=Tkinter.X)
     49
     50              self.connected_text = Tkinter.StringVar()
     51              self.connected_text.set("Server status: Not connected")
     52              self.server_hostname = Tkinter.StringVar()
     53              self.server_hostname.set( socket.getfqdn('') )
     54              self.server_port = Tkinter.IntVar()
     55              self.server_port.set( 7766 )
     56
     57              Tkinter.Label(connected_frame,
     58                            text="Server hostname").grid(row=0,
     59                                                         column=0)
     60              Tkinter.Entry(connected_frame,
     61                            textvariable=self.server_hostname).grid(row=0,
     62                                                                    column=1,
     63                                                                    columnspan=2)
     64              Tkinter.Label(connected_frame,
     65                            textvariable=self.connected_text).grid(row=1,
     66                                                                   column=0)
     67              Tkinter.Button(connected_frame,
     68                             text="Connect",
     69                             command=self.standalone_connect).grid(row=1,
     70                                                                   column=1)
     71              Tkinter.Button(connected_frame,
     72                             text="Quit server",
     73                             command=self.quit_server).grid(row=1,
     74                                                            column=2)
     75
     76          self.param_frame = Tkinter.Frame(self)
     77          self.param_frame.pack(expand=Tkinter.YES,fill=Tkinter.BOTH)
     78
     79          if not suppress_go_buttons:
     80              Tkinter.Button(self,text="Begin Trial",command=self.go).pack()#expand=Tkinter.YES,fill=Tkinter.BOTH)
     81
     82 -    def make_callback_entry(self, master=None, **kw):
     83          if 'width' not in kw.keys():
     84              kw['width'] = self.entry_width
     85          if master==None:
     86              master=self.param_frame
     87          e = Tkinter.Entry(master,**kw)
     88          e.bind('<Return>',self.send_values)
     89          e.bind('<Tab>',self.send_values)
     90          return e
     91
     92 -    def get_shortname(self):
     93          """Used as basename for saving parameter files and other ID purposes"""
     94          raise NotImplementedError("Must be overriden by derived class")
     95
     96 -    def set_param_dict(self,new_param_dict):
     97          orig_params = dir(self.meta_params)
     98          for new_param_name in new_param_dict.keys():
     99              if new_param_name[:2] != '__' and new_param_name[-2:] != '__':
    100                  if new_param_name not in orig_params:
    101                      raise ValueError('Gave parameter "%s", which I do not know about.'%(new_param_name,))
    102                  setattr(self.meta_params,new_param_name,new_param_dict[new_param_name])
    103          self.update_tk_vars()
    104          self.update() # update screen with new tk_var value
    105
    106 -    def update_tk_vars(self):
    107          """Update Tkinter variables with (new) values from meta_params"""
    108          raise NotImplementedError("Must be overriden by derived class")
    109
    110 -    def get_parameters_dict(self):
    111          result = {}
    112          for param_name in dir(self.meta_params):
    113              if param_name[:2] != '__' and param_name[-2:] != '__':
    114                  result[param_name] = getattr(self.meta_params,param_name)
    115          return result
    116
    117 -    def get_parameters_as_py_strings(self):
    118          """Return parameter values as Python-executable strings"""
    119          result = []
    120          for param_name in dir(self.meta_params):
    121              if param_name[:2] != '__' and param_name[-2:] != '__':
    122                  value = getattr(self.meta_params,param_name)
    123                  value_string = repr(value)
    124                  result.append((param_name,value_string))
    125          return result
    126
    127 -    def get_parameters_as_m_strings(self):
    128          """Return parameter values as Matlab-executable strings"""
    129          result = []
    130          for param_name in dir(self.meta_params):
    131              if param_name[:2] != '__' and param_name[-2:] != '__':
    132                  value = getattr(self.meta_params,param_name)
    133                  value_string = self.get_matlab_string(value)
    134                  result.append((param_name,value_string))
    135          return result
    136
    137 -    def get_matlab_string(self, value):
    138          # I'm no Matlab whiz, so you may have to modify this!!
    139          if type(value) in [types.IntType, types.FloatType]:
    140              return str(value)
    141          elif type(value) in [types.ListType, types.TupleType]:
    142              s = "[ "
    143              for v in value:
    144                  s += str(v) + " "
    145              s += "]"
    146              return s
    147          elif type(value) == types.StringType:
    148              s = "'%s'"%value
    149              return s
    150          else:
    151              raise NotImplementedError("No support for converting %s to Matlab format."%str(type(value)))
    152
    153 -    def set_parameters_dict(self, dict):
    154          for key in dict.keys():
    155              if not key in dir(self.meta_params):
    156                  raise RuntimeError("Parameter %s not in %s"%(key, str(self.meta_params)))
    157              setattr(self.meta_params,key,dict[key])
    158
    159 -    def get_loopable_variable_names(self):
    160          return self.loopable_variables.keys()
    161
    162 -    def set_loopable_variable(self,easy_name,value):
    163          meta_param_var_name,tk_var = self.loopable_variables[easy_name]
    164          setattr(self.meta_params,meta_param_var_name,value)
    165          tk_var.set(value)
    166          self.update() # update screen with new tk_var value
    167
    168 -    def send_values(self,dummy_arg=None):
    169          """Update meta_params variables with values from Tkinter fields"""
    170          raise NotImplementedError("Must be overriden by derived class")
    171
    172 -    def get_duration_sec(self):
    173          """Calculate total duration in go loop"""
    174          raise NotImplementedError("Must be overriden by derived class")
    175
    176 -    def go(self,dummy_arg=None):
    177          self.send_values()
    178          if not self.connected:
    179              raise RuntimeError("must be connected to run trial")
    180
    181          root = self.winfo_toplevel()
    182          old_cursor = root["cursor"]
    183
    184          root["cursor"] = "watch"
    185          root.update()
    186          self.meta_controller.go()
    187          root["cursor"] = old_cursor
    188          root.update()
    189
    190 -    def standalone_connect(self):
    191          self.connect(self.server_hostname.get(),self.server_port.get())
    192
    193 -    def connect(self,server_hostname,server_port):
    194          self.pyro_client = VisionEgg.PyroClient.PyroClient(server_hostname,server_port)
    195
    196          shortname = self.get_shortname()
    197          meta_controller_name = shortname + "_server"
    198          timeout_seconds = 60.0
    199          retry_interval_seconds = 0.1
    200          start_time = time.time()
    201          if hasattr(self,"meta_controller"):
    202              del self.meta_controller # get rid of old meta_controller
    203
    204          # get new meta_controller
    205          while not hasattr(self,"meta_controller"):
    206              try:
    207                  self.meta_controller = self.pyro_client.get(meta_controller_name)
    208              except Pyro.errors.NamingError, x:
    209                  if str(x) == "name not found":
    210                      if (time.time()-start_time) >= timeout_seconds:
    211                          raise # Couldn't find experiment controller on Pyro network
    212                      time.sleep(retry_interval_seconds)
    213                  else:
    214                      raise # unknown error
    215
    216          # attribute error: check: stimkey == short_name + "_server"
    217          self.meta_params = self.meta_controller.get_parameters()
    218
    219          self.connected = 1
    220          if hasattr(self,'connected_text'): # EPhysGUI client suppresses this
    221              self.connected_text.set("Server status: Connected")
    222
    223 -    def quit_server(self,dummy=None):
    224          self.meta_controller.quit_server()
    225          self.connected = 0
    226          if hasattr(self,'connected_text'): # EPhysGUI client suppresses this label
    227              self.connected_text.set("Server status: Not connected")
    228
    229  if __name__=='__main__':
    230      frame = StimulusControlFrame()
    231      frame.pack(expand=Tkinter.YES,fill=Tkinter.BOTH)
    232      frame.winfo_toplevel().title("%s"%(os.path.basename(os.path.splitext(sys.argv[0])[0]),))
    233      frame.mainloop()
    234