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
    124 -    def get_stim_onset_cal_location(self):
    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
    135 -    def were_frames_dropped_in_last_go_loop(self):
    136          return self.presentation.were_frames_dropped_in_last_go_loop()
    137
    138 -    def get_last_go_loop_start_time_absolute_sec(self):
    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
    145 -    def get_next_stimulus_meta_controller(self):
    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