panda3d/direct/src/cluster/ClusterServer.py

from pandac.PandaModules import *
from ClusterMsgs import *
from direct.distributed.MsgTypes import *
from direct.directnotify import DirectNotifyGlobal
from direct.showbase import DirectObject
from direct.task import Task

# NOTE: This assumes the following variables are set via bootstrap command line
# arguments on server startup:
#     clusterServerPort
#     clusterSyncFlag
#     clusterDaemonClient
#     clusterDaemonPort
# Also, I'm not sure multiple camera-group configurations are working for the
# cluster system.

class ClusterServer(DirectObject.DirectObject):
    notify = DirectNotifyGlobal.directNotify.newCategory("ClusterServer")
    MSG_NUM = 2000000

    def __init__(self, cameraJig, camera):
        global clusterServerPort, clusterSyncFlag
        global clusterDaemonClient, clusterDaemonPort
        # Store information about the cluster's camera
        self.cameraJig = cameraJig
        self.camera = camera
        self.lens = camera.node().getLens()
        self.lastConnection = None
        self.fPosReceived = 0
        # Create network layer objects
        self.qcm = QueuedConnectionManager()
        self.qcl = QueuedConnectionListener(self.qcm, 0)
        self.qcr = QueuedConnectionReader(self.qcm, 0)
        self.cw = ConnectionWriter(self.qcm, 0)
        try:
            port = clusterServerPort
        except NameError:
            port = CLUSTER_SERVER_PORT
        self.tcpRendezvous = self.qcm.openTCPServerRendezvous(port, 1)
        self.qcl.addConnection(self.tcpRendezvous)
        self.msgHandler = ClusterMsgHandler(ClusterServer.MSG_NUM, self.notify)
        # Start cluster tasks
        self.startListenerPollTask()
        self.startReaderPollTask()
        # If synchronized server, start swap coordinator too
        try:
            clusterSyncFlag
        except NameError:
            clusterSyncFlag = 0
        if clusterSyncFlag:
            self.startSwapCoordinator()
            base.graphicsEngine.setAutoFlip(0)
        # Set global clock mode to slave mode
        globalClock.setMode(ClockObject.MSlave)
        # Send verification of startup to client
        self.daemon = DirectD()

        self.objectMappings  = {}
        self.controlMappings = {}
        self.controlOffsets  = {}


        # These must be passed in as bootstrap arguments and stored in
        # the __builtins__ namespace
        try:
            clusterDaemonClient
        except NameError:
            clusterDaemonClient = 'localhost'
        try:
            clusterDaemonPort
        except NameError:
            clusterDaemonPort = CLUSTER_DAEMON_PORT
        self.daemon.serverReady(clusterDaemonClient, clusterDaemonPort)



    def startListenerPollTask(self):
        # Run this task near the start of frame, sometime after the dataLoop
        taskMgr.add(self.listenerPollTask, "serverListenerPollTask", -40)

    def listenerPollTask(self, task):
        """ Task to listen for a new connection from the client """
        # Run this task after the dataLoop
        if self.qcl.newConnectionAvailable():
            self.notify.info("New connection is available")
            rendezvous = PointerToConnection()
            netAddress = NetAddress()
            newConnection = PointerToConnection()
            if self.qcl.getNewConnection(rendezvous, netAddress, newConnection):
                # Crazy dereferencing
                newConnection=newConnection.p()
                self.qcr.addConnection(newConnection)
                self.lastConnection = newConnection
                self.notify.info("Got a connection!")
            else:
                self.notify.warning("getNewConnection returned false")
        return Task.cont


    def addNamedObjectMapping(self,object,name):
        if (not self.objectMappings.has_key(name)):
            self.objectMappings[name] = object
        else:
            self.notify.debug('attempt to add duplicate named object: '+name)

    def removeObjectMapping(self,name):
        if (self.objectMappings.has_key(name)):
            self.objectMappings.pop(name)


    def addControlMapping(self,objectName,controlledName, offset = None):
        if (not self.controlMappings.has_key(objectName)):
            self.controlMappings[objectName] = controlledName
            if (offset == None):
                offset = Vec3(0,0,0)
            self.controlOffsets[objectName]  = offset
        else:
            self.notify.debug('attempt to add duplicate controlled object: '+name)

    def setControlMappingOffset(self,objectName,offset):
        if (self.controlMappings.has_key(objectName)):
            self.controlOffsets[objectName] = offset
    

    def removeControlMapping(self,name):
        if (self.controlMappings.has_key(name)):
            self.controlMappings.pop(name)

            
    def startControlObjectTask(self):
        self.notify.debug("moving control objects")
        taskMgr.add(self.controlObjectTask,"controlObjectTask")

    def controlObjectTask(self, task):
        for object in self.controlMappings:
            name       = self.controlMappings[object]
            if (self.objectMappings.has_key(object)):
                self.moveObject(self.objectMappings[object],name,self.controlOffsets[object])

        return Task.cont

    def moveObject(self, nodePath, object, offset):
        self.notify.debug('moving object '+object)
        xyz = nodePath.getPos(render) + offset
        hpr = nodePath.getHpr(render)
        scale = nodePath.getScale(render)
        hidden = nodePath.isHidden()
        datagram = self.msgHandler.makeNamedObjectMovementDatagram(xyz,hpr,scale,hidden,object)
        self.cw.send(datagram, self.lastConnection)

    def startReaderPollTask(self):
        """ Task to handle datagrams from client """
        # Run this task just after the listener poll task
        if clusterSyncFlag:
            # Sync version
            taskMgr.add(self._syncReaderPollTask, "serverReaderPollTask", -39)
        else:
            # Asynchronous version
            taskMgr.add(self._readerPollTask, "serverReaderPollTask", -39)

    def _readerPollTask(self, state):
        """ Non blocking task to read all available datagrams """
        while 1:
            (datagram, dgi, type) = self.msgHandler.nonBlockingRead(self.qcr)
            # Queue is empty, done for now
            if type is CLUSTER_NONE:
                break
            else:
                # Got a datagram, handle it
                self.handleDatagram(dgi, type)
        return Task.cont

    def _syncReaderPollTask(self, task):
        if self.lastConnection is None:
            pass
        elif self.qcr.isConnectionOk(self.lastConnection):
            # Process datagrams till you get a postion update
            type = CLUSTER_NONE
            while type != CLUSTER_CAM_MOVEMENT:
                # Block until you get a new datagram
                (datagram, dgi, type) = self.msgHandler.blockingRead(self.qcr)
                # Process datagram
                self.handleDatagram(dgi, type)
        return Task.cont

    def startSwapCoordinator(self):
        taskMgr.add(self.swapCoordinatorTask, "serverSwapCoordinator", 51)

    def swapCoordinatorTask(self, task):
        if self.fPosReceived:
            self.fPosReceived = 0
            # Alert client that this server is ready to swap
            self.sendSwapReady()
            # Wait for swap command (processing any intermediate datagrams)
            while 1:
                (datagram, dgi, type) = self.msgHandler.blockingRead(self.qcr)
                self.handleDatagram(dgi, type)
                if type == CLUSTER_SWAP_NOW:
                    break
        return Task.cont

    def sendSwapReady(self):
        self.notify.debug(
            'send swap ready packet %d' % self.msgHandler.packetNumber)
        datagram = self.msgHandler.makeSwapReadyDatagram()
        self.cw.send(datagram, self.lastConnection)

    def handleDatagram(self, dgi, type):
        """ Process a datagram depending upon type flag """
        if (type == CLUSTER_NONE):
            pass
        elif (type == CLUSTER_EXIT):
            print 'GOT EXIT'
            import sys
            sys.exit()
        elif (type == CLUSTER_CAM_OFFSET):
            self.handleCamOffset(dgi)
        elif (type == CLUSTER_CAM_FRUSTUM):
            self.handleCamFrustum(dgi)
        elif (type == CLUSTER_CAM_MOVEMENT):
            self.handleCamMovement(dgi)
        elif (type == CLUSTER_SELECTED_MOVEMENT):
            self.handleSelectedMovement(dgi)
        elif (type == CLUSTER_COMMAND_STRING):
            self.handleCommandString(dgi)
        elif (type == CLUSTER_SWAP_READY):
            pass
        elif (type == CLUSTER_SWAP_NOW):
            self.notify.debug('swapping')
            base.graphicsEngine.flipFrame()
        elif (type == CLUSTER_TIME_DATA):
            self.notify.debug('time data')
            self.handleTimeData(dgi)
        elif (type == CLUSTER_NAMED_OBJECT_MOVEMENT):
            self.handleNamedMovement(dgi)
        else:
            self.notify.warning("Received unknown packet type:" % type)
        return type

    # Server specific tasks
    def handleCamOffset(self, dgi):
        """ Set offset of camera from cameraJig """
        (x, y, z, h, p, r) = self.msgHandler.parseCamOffsetDatagram(dgi)
        self.camera.setPos(x,y,z)
        self.lens.setViewHpr(h, p, r)

    def handleCamFrustum(self, dgi):
        """ Adjust camera frustum based on parameters sent by client """
        (fl, fs, fo) = self.msgHandler.parseCamFrustumDatagram(dgi)
        self.lens.setFocalLength(fl)
        self.lens.setFilmSize(fs[0], fs[1])
        self.lens.setFilmOffset(fo[0], fo[1])

    def handleNamedMovement(self, dgi):
        """ Update cameraJig position to reflect latest position """
        (name,x, y, z, h, p, r,sx,sy,sz, hidden) = self.msgHandler.parseNamedMovementDatagram(
            dgi)
        if (self.objectMappings.has_key(name)):
            self.objectMappings[name].setPosHpr(render, x, y, z, h, p, r)
            self.objectMappings[name].setScale(render,sx,sy,sz)
            if (hidden):
                self.objectMappings[name].hide()
            else:
                self.objectMappings[name].show()
        else:
            self.notify.debug("recieved unknown named object command: "+name)

    def handleCamMovement(self, dgi):
        """ Update cameraJig position to reflect latest position """
        (x, y, z, h, p, r) = self.msgHandler.parseCamMovementDatagram(dgi)
        self.cameraJig.setPosHpr(render, x, y, z, h, p, r)
        self.fPosReceived = 1        

    def handleSelectedMovement(self, dgi):
        """ Update cameraJig position to reflect latest position """
        (x, y, z, h, p, r, sx, sy, sz) = self.msgHandler.parseSelectedMovementDatagram(
            dgi)
        if last:
            last.setPosHprScale(x, y, z, h, p, r, sx, sy, sz)

    def handleTimeData(self, dgi):
        """ Update cameraJig position to reflect latest position """
        (frameCount, frameTime, dt) = self.msgHandler.parseTimeDataDatagram(dgi)
        # Use frame time from client for both real and frame time
        globalClock.setFrameCount(frameCount)
        globalClock.setFrameTime(frameTime)
        globalClock.setDt(dt)

    def handleCommandString(self, dgi):
        """ Handle arbitrary command string from client """
        command = self.msgHandler.parseCommandStringDatagram(dgi)
        try:
            exec(command, __builtins__)
        except:
            pass