1
2
3
4
5
6
7
8
9
10
11
12
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
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()
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
93 """Used as basename for saving parameter files and other ID purposes"""
94 raise NotImplementedError("Must be overriden by derived class")
95
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()
105
107 """Update Tkinter variables with (new) values from meta_params"""
108 raise NotImplementedError("Must be overriden by derived class")
109
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
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
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
138
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
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
160 return self.loopable_variables.keys()
161
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()
167
169 """Update meta_params variables with values from Tkinter fields"""
170 raise NotImplementedError("Must be overriden by derived class")
171
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
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
203
204
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
212 time.sleep(retry_interval_seconds)
213 else:
214 raise
215
216
217 self.meta_params = self.meta_controller.get_parameters()
218
219 self.connected = 1
220 if hasattr(self,'connected_text'):
221 self.connected_text.set("Server status: Connected")
222
224 self.meta_controller.quit_server()
225 self.connected = 0
226 if hasattr(self,'connected_text'):
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