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

Source Code for Module VisionEgg.PyroApps.EPhysServer

  1  #!/usr/bin/env python 
  2  # 
  3  # The Vision Egg: EPhysServer 
  4  # 
  5  # Copyright (C) 2001-2004 Andrew Straw, 
  6  # Copyright (C) 2004 Imran S. Ali, Lachlan Dowd 
  7  # Copyright (C) 2004 California Institute of Technology 
  8  # 
  9  # Author: Andrew Straw <astraw@users.sourceforge.net> 
 10  # URL: <http://www.visionegg.org/> 
 11  # 
 12  # Distributed under the terms of the GNU Lesser General Public License 
 13  # (LGPL). See LICENSE.TXT that came with this file. 
 14  # 
 15  # $Id$ 
 16   
 17  import VisionEgg 
 18  __version__ = VisionEgg.release_name 
 19  __cvs__ = '$Revision$'.split()[1] 
 20  __date__ = ' '.join('$Date$'.split()[1:3]) 
 21  __author__ = 'Andrew Straw <astraw@users.sourceforge.net>' 
 22   
 23  import sys, os, math 
 24  import parser, symbol, token, compiler 
 25  import tkMessageBox 
 26  import pygame.display 
 27  import VisionEgg.Core 
 28  import VisionEgg.FlowControl 
 29  import VisionEgg.SphereMap 
 30  import VisionEgg.Text 
 31  import VisionEgg.Textures 
 32  import VisionEgg.PyroHelpers 
 33  import Pyro, Pyro.core 
 34   
 35  # AST extensions: 
 36  import VisionEgg.PyroApps.AST_ext as AST_ext 
 37   
 38  # Add your stimulus modules here 
 39  import VisionEgg.PyroApps.TargetServer 
 40  import VisionEgg.PyroApps.MouseTargetServer 
 41  import VisionEgg.PyroApps.FlatGratingServer 
 42  import VisionEgg.PyroApps.SphereGratingServer 
 43  import VisionEgg.PyroApps.SpinningDrumServer 
 44  import VisionEgg.PyroApps.GridServer 
 45  import VisionEgg.PyroApps.ColorCalServer 
 46  import VisionEgg.PyroApps.DropinServer 
 47   
 48  server_modules = [ VisionEgg.PyroApps.TargetServer, 
 49                     VisionEgg.PyroApps.MouseTargetServer, 
 50                     VisionEgg.PyroApps.FlatGratingServer, 
 51                     VisionEgg.PyroApps.SphereGratingServer, 
 52                     VisionEgg.PyroApps.SpinningDrumServer, 
 53                     VisionEgg.PyroApps.GridServer, 
 54                     VisionEgg.PyroApps.ColorCalServer, 
 55                     VisionEgg.PyroApps.DropinServer 
 56                     ] 
 57   
 58  # 3D screen positioning parameters 
 59  from VisionEgg.PyroApps.ScreenPositionServer import ScreenPositionMetaController 
 60  from VisionEgg.PyroApps.ScreenPositionGUI import ScreenPositionParameters 
 61   
62 -class EPhysServer( Pyro.core.ObjBase ):
63 - def __init__(self, presentation, server_modules ):
64 Pyro.core.ObjBase.__init__(self) 65 self.stimdict = {} 66 self.stimkey = server_modules[0].get_meta_controller_stimkey() # first stimulus will be this 67 self.quit_status = False 68 self.exec_demoscript_flag = False 69 self.presentation = presentation 70 # target for stimulus onset calibration 71 self.onset_cal_bg = VisionEgg.MoreStimuli.Target2D(color=(0.0,0.0,0.0,1.0), 72 position=(30.0,30.0), 73 anchor='center', 74 size=(50.0,50.0)) 75 self.onset_cal_fg = VisionEgg.MoreStimuli.Target2D(on=0, 76 color=(1.0,1.0,1.0,1.0), 77 position=(30.0,30.0), 78 anchor='center', 79 size=(50.0,50.0)) 80 self.presentation.add_controller(self.onset_cal_fg,'on',VisionEgg.FlowControl.ConstantController(during_go_value=1, 81 between_go_value=0)) 82 # get screen (hack) 83 self.onset_cal_screen = self.presentation.parameters.viewports[0].parameters.screen 84 self.onset_cal_viewport = VisionEgg.Core.Viewport(screen=self.onset_cal_screen, 85 stimuli=[self.onset_cal_bg, 86 self.onset_cal_fg] 87 ) 88 89 for server_module in server_modules: 90 stimkey = server_module.get_meta_controller_stimkey() 91 klass = server_module.get_meta_controller_class() 92 make_stimuli = server_module.make_stimuli 93 self.stimdict[stimkey] = (klass, make_stimuli)
94
95 - def __del__(self):
96 self.presentation.remove_controller(self.onset_cal_fg,'on') 97 Pyro.core.ObjBase.__del__(self)
98
99 - def get_quit_status(self):
100 return self.quit_status
101
102 - def set_quit_status(self,quit_status):
103 self.quit_status = quit_status 104 self.presentation.parameters.quit = quit_status
105
106 - def first_connection(self):
107 # break out of initial run_forever loop 108 self.presentation.parameters.quit = True
109
110 - def set_stim_onset_cal(self, on):
111 if on: 112 if self.onset_cal_viewport not in self.presentation.parameters.viewports: 113 self.presentation.parameters.viewports.append(self.onset_cal_viewport) 114 else: 115 if self.onset_cal_viewport in self.presentation.parameters.viewports: 116 self.presentation.parameters.viewports.remove(self.onset_cal_viewport)
117
118 - def set_stim_onset_cal_location(self, center, size):
119 self.onset_cal_bg.parameters.position = center 120 self.onset_cal_fg.parameters.position = center 121 self.onset_cal_bg.parameters.size = size 122 self.onset_cal_fg.parameters.size = size[0]-2,size[1]-2
123
125 x,y = self.onset_cal_bg.parameters.position 126 width,height = self.onset_cal_bg.parameters.size 127 return x,y,width,height
128
129 - def set_gamma_ramp(self, red, blue, green):
130 return pygame.display.set_gamma_ramp(red,green,blue)
131
132 - def is_in_go_loop(self):
133 return self.presentation.is_in_go_loop()
134
136 return self.presentation.were_frames_dropped_in_last_go_loop()
137
139 return self.presentation.get_last_go_loop_start_time_absolute_sec()
140
141 - def set_override_t_abs_sec(self, value_sec_string):
142 value_sec = float(value_sec_string) # Pyro loses precision 143 self.presentation.parameters.override_t_abs_sec = value_sec
144
146 if self.stimkey: 147 klass, make_stimuli = self.stimdict[self.stimkey] 148 stimuli = make_stimuli() 149 return self.stimkey, klass, stimuli 150 else: 151 raise RuntimeError("No stimkey")
152
153 - def get_stimkey(self):
154 return self.stimkey
155
156 - def set_next_stimkey(self,stimkey):
157 self.stimkey = stimkey
158
159 - def get_cwd(self):
160 return os.path.abspath(os.curdir)
161
162 - def save_image_sequence(self,fps=12.0,filename_base="im",filename_suffix=".tif",save_dir="."):
163 try: 164 self.presentation.export_movie_go(frames_per_sec=fps, 165 filename_base=filename_base, 166 filename_suffix=filename_suffix, 167 path=save_dir 168 ) 169 except Exception,x: 170 # do this because Pyro doesn't (by default) print a traceback 171 import traceback 172 traceback.print_exc() 173 raise
174
175 - def build_AST(self, source):
176 AST = parser.suite(source) 177 self.AST = AST
178
179 - def exec_AST(self, screen, dropin_meta_params):
180 181 if dropin_meta_params.vars_list is not None: 182 for var in dropin_meta_params.vars_list: 183 self.AST = AST_ext.modify_AST(self.AST, var[0], var[1]) 184 185 code_module = self.AST.compile() 186 exec code_module in locals() 187 self.script_dropped_frames = p.were_frames_dropped_in_last_go_loop() 188 self.presentation.last_go_loop_start_time_absolute_sec = p.last_go_loop_start_time_absolute_sec # evil hack... 189 self.exec_demoscript_flag = False
190
191 - def run_demoscript(self):
192 self.exec_demoscript_flag = True
193
194 -def start_server( server_modules, server_class=EPhysServer ):
195 196 loadNewExpr = True 197 pyro_server = VisionEgg.PyroHelpers.PyroServer() 198 199 # get Vision Egg stimulus ready to go 200 screen = VisionEgg.Core.Screen.create_default() 201 202 temp = ScreenPositionParameters() 203 204 projection = VisionEgg.Core.PerspectiveProjection(temp.left, 205 temp.right, 206 temp.bottom, 207 temp.top, 208 temp.near, 209 temp.far) 210 perspective_viewport = VisionEgg.Core.Viewport(screen=screen, 211 projection=projection) 212 overlay2D_viewport = VisionEgg.Core.Viewport(screen=screen) 213 p = VisionEgg.FlowControl.Presentation(viewports=[perspective_viewport, overlay2D_viewport]) # 2D overlay on top 214 print 'main Presentation',p 215 216 wait_text = VisionEgg.Text.Text( 217 text = "Starting up...", 218 position = (screen.size[0]/2.0,5), 219 anchor='bottom', 220 color = (1.0,0.0,0.0,0.0)) 221 222 overlay2D_viewport.parameters.stimuli = [wait_text] 223 p.between_presentations() # draw wait_text 224 225 # now hand over control of projection to ScreenPositionMetaController 226 projection_controller = ScreenPositionMetaController(p,projection) 227 pyro_server.connect(projection_controller,"projection_controller") 228 229 ephys_server = server_class(p, server_modules) 230 pyro_server.connect(ephys_server,"ephys_server") 231 hostname,port = pyro_server.get_hostname_and_port() 232 233 wait_text.parameters.text = "Waiting for connection at %s port %d"%(hostname,port) 234 235 # get listener controller and register it 236 p.add_controller(None,None, pyro_server.create_listener_controller()) 237 238 p.run_forever() # run until we get first connnection, which breaks out immmediately 239 240 wait_text.parameters.text = "Loading new experiment, please wait." 241 242 while not ephys_server.get_quit_status(): 243 # this flow control configuration SEEMS to be stable for 244 # contiguously loaded scripts more rigorous testing would be 245 # appreciated 246 if loadNewExpr: 247 wait_text.parameters.text = "Loading new experiment, please wait." 248 perspective_viewport.parameters.stimuli = [] 249 overlay2D_viewport.parameters.stimuli = [wait_text] 250 p.between_presentations() # draw wait_text 251 pyro_name, meta_controller_class, stimulus_list = ephys_server.get_next_stimulus_meta_controller() 252 stimulus_meta_controller = meta_controller_class(screen, p, stimulus_list) # instantiate meta_controller 253 pyro_server.connect(stimulus_meta_controller, pyro_name) 254 255 if ephys_server.get_stimkey() == "dropin_server": 256 wait_text.parameters.text = "Vision Egg script mode" 257 258 p.parameters.enter_go_loop = False 259 p.parameters.quit = False 260 p.run_forever() 261 262 # At this point quit signal was sent by client to either: 263 264 # 1) Execute the script (ie. "exec_demoscript_flag" has 265 # been set) 266 267 # 2) Load a DIFFERENT script ("loadNewExpr" should be set 268 # to False in this event) 269 270 # 3) Load a BUILT IN experiment ("loadNewExpr" should be 271 # set to True in this event) 272 273 if ephys_server.exec_demoscript_flag: 274 dropin_meta_params = stimulus_meta_controller.get_parameters() 275 ephys_server.exec_AST(screen, dropin_meta_params) 276 277 if ephys_server.get_stimkey() == "dropin_server": 278 # Either: 279 # 1) Same script (just finished executing) 280 # 2) Loading a new script 281 loadNewExpr = False 282 else: 283 # 3) load a BUILT IN experiment 284 pyro_server.disconnect(stimulus_meta_controller) 285 del stimulus_meta_controller # we have to do this explicitly because Pyro keeps a copy of the reference 286 loadNewExpr = True 287 else: 288 overlay2D_viewport.parameters.stimuli = [] # clear wait_text 289 for stim in stimulus_list: 290 if stim[0] == '3d_perspective': 291 perspective_viewport.parameters.stimuli.append(stim[1]) 292 elif stim[0] == '3d_perspective_with_set_viewport_callback': 293 key, stimulus, callback_function = stim 294 callback_function(perspective_viewport) 295 perspective_viewport.parameters.stimuli.append(stimulus) 296 elif stim[0] == '2d_overlay': 297 overlay2D_viewport.parameters.stimuli.append(stim[1]) 298 else: 299 raise RuntimeError("Unknown viewport id %s"%stim[0]) 300 301 # enter loop 302 p.parameters.enter_go_loop = False 303 p.parameters.quit = False 304 p.run_forever() 305 306 # At this point quit signal was sent by client to either: 307 308 # 1) Load a script ("loadNewExpr" should be set to 1 in 309 # this event) 310 311 # 2) Load a BUILT IN experiment ("loadNewExpr" should be 312 # set to 1 in this event) 313 314 pyro_server.disconnect(stimulus_meta_controller) 315 del stimulus_meta_controller # we have to do this explicitly because Pyro keeps a copy of the reference
316 317 if __name__ == '__main__': 318 start_server( server_modules ) 319