def process_incoming(self):
        """
        process incoming messages

        data is a dict with several arrays
        """

        model = self.model
        sockets = self.sockets

        # Check for new messages
        if not sockets:
            return
        # unpack sockets
        poller = sockets['poller']
        rep = sockets['rep']
        # pull = sockets['pull']
        pub = sockets['pub']
        items = poller.poll(10)
        for sock, n in items:
            for i in range(n):

                A, metadata = recv_array(sock)
                var = None

                # bmi actions
                if "update" in metadata:
                    dt = float(metadata["update"])
                    model.update(dt)
                    metadata["dt"] = dt
                elif "get_var" in metadata:
                    name = metadata["get_var"]
                    # temporary implementation
                    if metadata.get("copy", False):
                        var = model.get_var(name).copy()
                    else:
                        var = model.get_var(name)
                    if var is None:
                        logger.warning("Get_var returns None for %s" % name)
                    metadata['name'] = name
                    # assert socket is req socket
                elif "get_var_count" in metadata:
                    # temporary implementation
                    n = model.get_var_count()
                    metadata['get_var_count'] = n
                    # assert socket is req socket
                elif "get_var_rank" in metadata:
                    # temporary implementation
                    var_name = metadata['get_var_rank']
                    n = model.get_var_rank(var_name)
                    metadata['get_var_rank'] = n
                    # assert socket is req socket
                elif "get_var_shape" in metadata:
                    # temporary implementation
                    var_name = metadata['get_var_shape']
                    n = model.get_var_shape(var_name)
                    metadata['get_var_shape'] = tuple([int(item) for item in n])
                    # assert socket is req socket
                elif "get_var_type" in metadata:
                    # temporary implementation
                    var_name = metadata['get_var_type']
                    n = model.get_var_type(var_name)
                    metadata['get_var_type'] = n
                    # assert socket is req socket
                elif "get_var_name" in metadata:
                    i = int(metadata["get_var_name"])
                    name = model.get_var_name(i)
                    metadata['get_var_name'] = name
                    # assert socket is req socket
                elif "set_current_time" in metadata:
                    t = float(metadata["set_current_time"])
                    model.set_current_time(t)
                elif "set_var" in metadata:
                    name = metadata["set_var"]
                    # logger.debug("setting variable %s", name)
                    model.set_var(name, A)
                    metadata["name"] = name  # !?
                elif "set_var_slice" in metadata:
                    name = metadata["set_var_slice"]
                    logger.debug("setting variable %s", name)
                    start = metadata["start"]
                    count = metadata["count"]
                    model.set_var_slice(name, start, count, A)
                    metadata["name"] = name  # !?
                elif "set_var_index" in metadata:
                    name = metadata["set_var_index"]
                    logger.debug("setting variable %s using index index", name)
                    index = metadata["index"]
                    # TODO: test if this is fast enough.
                    # Otherwise move to BMI ++ but that is
                    # a bit of a burden on implementers
                    var = model.get_var(name).copy()
                    var.flat[index] = A
                    model.set_var(name, var)
                    metadata["name"] = name  # !?
                elif "get_current_time" in metadata:
                    metadata["get_current_time"]
                    t = model.get_current_time()
                    metadata['get_current_time'] = t
                elif "get_time_step" in metadata:
                    metadata["get_time_step"]
                    dt = model.get_time_step()
                    metadata['get_time_step'] = dt
                elif "get_end_time" in metadata:
                    metadata["get_end_time"]
                    t = model.get_end_time()
                    metadata['get_end_time'] = t
                elif "get_start_time" in metadata:
                    metadata["get_start_time"]
                    t = model.get_start_time()
                    metadata['get_start_time'] = t
                    # assert socket is req socket
                # custom actions
                elif "remote" in metadata:
                    assert metadata["remote"] in {
                        "play", "stop", "pause", "rewind", "quit"}
                    model.state = metadata["remote"]
                elif "operator" in metadata:
                    # TODO: support same operators as MPI_ops here....,
                    # TODO: reduce before apply
                    # TODO: assert pull socket
                    pass
                    # S = tuple(slice(*x) for x in action['slice'])
                    # print(repr(arr[S]))
                    # if action['operator'] == 'setitem':
                    #     arr[S] = data
                    # elif action['operator'] == 'add':
                    #     arr[S] += data
                elif "initialize" in metadata:
                    config_file = metadata["initialize"]
                    model.initialize(str(config_file))
                elif "finalize" in metadata:
                    model.finalize()
                else:
                    logger.warn("got unknown message {} from socket {}".format(str(metadata), sock))
                if sock.socket_type == zmq.REP:
                    # reply
                    send_array(rep, var, metadata=metadata)
                # any getter requested through the pull socket?
                elif any(x.startswith("get_") for x in metadata) and sock.socket_type == zmq.PULL:
                    # return through the pub socket
                    send_array(pub, var, metadata=metadata)
def process_incoming ( self ) : model = self . model sockets = self . sockets if not sockets : return poller = sockets [ 'poller' ] rep = sockets [ 'rep' ] pub = sockets [ 'pub' ] items = poller . poll ( 10 ) for sock , n in items : for i in range ( n ) : A , metadata = recv_array ( sock ) var = None if "update" in metadata : dt = float ( metadata [ "update" ] ) model . update ( dt ) metadata [ "dt" ] = dt elif "get_var" in metadata : name = metadata [ "get_var" ] if metadata . get ( "copy" , False ) : var = model . get_var ( name ) . copy ( ) else : var = model . get_var ( name ) if var is None : logger . warning ( "Get_var returns None for %s" % name ) metadata [ 'name' ] = name elif "get_var_count" in metadata : n = model . get_var_count ( ) metadata [ 'get_var_count' ] = n elif "get_var_rank" in metadata : var_name = metadata [ 'get_var_rank' ] n = model . get_var_rank ( var_name ) metadata [ 'get_var_rank' ] = n elif "get_var_shape" in metadata : var_name = metadata [ 'get_var_shape' ] n = model . get_var_shape ( var_name ) metadata [ 'get_var_shape' ] = tuple ( [ int ( item ) for item in n ] ) elif "get_var_type" in metadata : var_name = metadata [ 'get_var_type' ] n = model . get_var_type ( var_name ) metadata [ 'get_var_type' ] = n elif "get_var_name" in metadata : i = int ( metadata [ "get_var_name" ] ) name = model . get_var_name ( i ) metadata [ 'get_var_name' ] = name elif "set_current_time" in metadata : t = float ( metadata [ "set_current_time" ] ) model . set_current_time ( t ) elif "set_var" in metadata : name = metadata [ "set_var" ] model . set_var ( name , A ) metadata [ "name" ] = name elif "set_var_slice" in metadata : name = metadata [ "set_var_slice" ] logger . debug ( "setting variable %s" , name ) start = metadata [ "start" ] count = metadata [ "count" ] model . set_var_slice ( name , start , count , A ) metadata [ "name" ] = name elif "set_var_index" in metadata : name = metadata [ "set_var_index" ] logger . debug ( "setting variable %s using index index" , name ) index = metadata [ "index" ] var = model . get_var ( name ) . copy ( ) var . flat [ index ] = A model . set_var ( name , var ) metadata [ "name" ] = name elif "get_current_time" in metadata : metadata [ "get_current_time" ] t = model . get_current_time ( ) metadata [ 'get_current_time' ] = t elif "get_time_step" in metadata : metadata [ "get_time_step" ] dt = model . get_time_step ( ) metadata [ 'get_time_step' ] = dt elif "get_end_time" in metadata : metadata [ "get_end_time" ] t = model . get_end_time ( ) metadata [ 'get_end_time' ] = t elif "get_start_time" in metadata : metadata [ "get_start_time" ] t = model . get_start_time ( ) metadata [ 'get_start_time' ] = t elif "remote" in metadata : assert metadata [ "remote" ] in { "play" , "stop" , "pause" , "rewind" , "quit" } model . state = metadata [ "remote" ] elif "operator" in metadata : pass elif "initialize" in metadata : config_file = metadata [ "initialize" ] model . initialize ( str ( config_file ) ) elif "finalize" in metadata : model . finalize ( ) else : logger . warn ( "got unknown message {} from socket {}" . format ( str ( metadata ) , sock ) ) if sock . socket_type == zmq . REP : send_array ( rep , var , metadata = metadata ) elif any ( x . startswith ( "get_" ) for x in metadata ) and sock . socket_type == zmq . PULL : send_array ( pub , var , metadata = metadata )
process incoming messages

        data is a dict with several arrays