Tutorial 2 - A simple pipeline with signal postprocessing.¶

Here we demonstrate how to build a MathAgent as an intermediate to process the SineGeneratorAgent’s output before plotting. Subsequently, a MultiMathAgent is built to show the ability to send a dictionary of multiple fields to the recipient. We provide a custom on_received_message() function, which is called every time a message is received from input agents.

The received message is a dictionary of the form:

{
'from':agent_name,
'data': data,
'senderType': agent_class_name,
'channel':'channel_name'
}

By default, 'channel' is set to "default", however a custom channel can be set when needed, which is demonstrated in the next tutorial.

[4]:

# %load tutorial_2_math_agent.py
from agentMET4FOF.agents import AgentMET4FOF, AgentNetwork, MonitorAgent
from agentMET4FOF.streams import SineGenerator


class MathAgent(AgentMET4FOF):
    def on_received_message(self, message):
        data = self.divide_by_two(message["data"])
        self.send_output(data)

    # Define simple math functions.
    @staticmethod
    def divide_by_two(numerator: float) -> float:
        return numerator / 2


class MultiMathAgent(AgentMET4FOF):

    _minus_param: float
    _plus_param: float

    def init_parameters(self, minus_param=0.5, plus_param=0.5):
        self._minus_param = minus_param
        self._plus_param = plus_param

    def on_received_message(self, message):
        minus_data = self.minus(message["data"], self._minus_param)
        plus_data = self.plus(message["data"], self._plus_param)

        self.send_output({"minus": minus_data, "plus": plus_data})

    @staticmethod
    def minus(minuend: float, subtrahend: float) -> float:
        return minuend - subtrahend

    @staticmethod
    def plus(summand_1: float, summand_2: float) -> float:
        return summand_1 + summand_2


class SineGeneratorAgent(AgentMET4FOF):

    _stream: SineGenerator

    def init_parameters(self):
        self._stream = SineGenerator()

    def agent_loop(self):
        if self.current_state == "Running":
            sine_data = self._stream.next_sample()  # dictionary
            self.send_output(sine_data["x"])


def main():
    # start agent network server
    agentNetwork = AgentNetwork()
    # init agents
    gen_agent = agentNetwork.add_agent(agentType=SineGeneratorAgent)
    math_agent = agentNetwork.add_agent(agentType=MathAgent)
    multi_math_agent = agentNetwork.add_agent(agentType=MultiMathAgent)
    monitor_agent = agentNetwork.add_agent(agentType=MonitorAgent)
    # connect agents : We can connect multiple agents to any particular agent
    agentNetwork.bind_agents(gen_agent, math_agent)
    agentNetwork.bind_agents(gen_agent, multi_math_agent)
    # connect
    agentNetwork.bind_agents(gen_agent, monitor_agent)
    agentNetwork.bind_agents(math_agent, monitor_agent)
    agentNetwork.bind_agents(multi_math_agent, monitor_agent)
    # set all agents states to "Running"
    agentNetwork.set_running_state()

    # allow for shutting down the network after execution
    return agentNetwork


if __name__ == "__main__":
    main()

Starting NameServer...
Broadcast server running on 0.0.0.0:9091
NS running on 127.0.0.1:3333 (127.0.0.1)
URI = PYRO:Pyro.NameServer@127.0.0.1:3333
INFO [2020-07-08 20:12:37.967786] (AgentController): INITIALIZED
Dash is running on http://127.0.0.1:8050/

 Warning: This is a development server. Do not use app.run_server
INFO [2020-07-08 20:12:38.108067] (SineGeneratorAgent_1): INITIALIZED
 in production, use a production WSGI server like gunicorn instead.

 * Serving Flask app "agentMET4FOF.dashboard.Dashboard" (lazy loading)
 * Environment: production
   WARNING: This is a development server. Do not use it in a production deployment.
   Use a production WSGI server instead.
 * Debug mode: off
INFO [2020-07-08 20:12:38.185234] (MathAgent_1): INITIALIZED
INFO [2020-07-08 20:12:38.266143] (MultiMathAgent_1): INITIALIZED
INFO [2020-07-08 20:12:38.347055] (MonitorAgent_1): INITIALIZED
[2020-07-08 20:12:38.413035] (SineGeneratorAgent_1): Connected output module: MathAgent_1
[2020-07-08 20:12:38.418991] (SineGeneratorAgent_1): Connected output module: MultiMathAgent_1
[2020-07-08 20:12:38.425940] (SineGeneratorAgent_1): Connected output module: MonitorAgent_1
[2020-07-08 20:12:38.465907] (MathAgent_1): Connected output module: MonitorAgent_1
[2020-07-08 20:12:38.538248] (MultiMathAgent_1): Connected output module: MonitorAgent_1
SET STATE:   Running
[2020-07-08 20:12:39.125553] (SineGeneratorAgent_1): Pack time: 0.000893
[2020-07-08 20:12:39.128378] (SineGeneratorAgent_1): Sending: [0.]
[2020-07-08 20:12:39.134086] (MultiMathAgent_1): Received: {'from': 'SineGeneratorAgent_1', 'data': array([0.]), 'senderType': 'SineGeneratorAgent', 'channel': 'default'}
[2020-07-08 20:12:39.134202] (MonitorAgent_1): Received: {'from': 'SineGeneratorAgent_1', 'data': array([0.]), 'senderType': 'SineGeneratorAgent', 'channel': 'default'}
[2020-07-08 20:12:39.135154] (MathAgent_1): Received: {'from': 'SineGeneratorAgent_1', 'data': array([0.]), 'senderType': 'SineGeneratorAgent', 'channel': 'default'}
[2020-07-08 20:12:39.137182] (MathAgent_1): Pack time: 0.001182
[2020-07-08 20:12:39.142027] (MultiMathAgent_1): Pack time: 0.00693
[2020-07-08 20:12:39.135086] (MonitorAgent_1): Tproc: 0.000153
[2020-07-08 20:12:39.138563] (MathAgent_1): Sending: [0.]
[2020-07-08 20:12:39.146332] (MultiMathAgent_1): Sending: {'minus': array([-0.5]), 'plus': array([0.5])}
[2020-07-08 20:12:39.140792] (MonitorAgent_1): Received: {'from': 'MultiMathAgent_1', 'data': {'minus': array([-0.5]), 'plus': array([0.5])}, 'senderType': 'MultiMathAgent', 'channel': 'default'}
[2020-07-08 20:12:39.140299] (MathAgent_1): Tproc: 0.00405
[2020-07-08 20:12:39.147118] (MultiMathAgent_1): Tproc: 0.012217
[2020-07-08 20:12:39.141319] (MonitorAgent_1): Tproc: 8.7e-05
[2020-07-08 20:12:39.143377] (MonitorAgent_1): Received: {'from': 'MathAgent_1', 'data': array([0.]), 'senderType': 'MathAgent', 'channel': 'default'}
[2020-07-08 20:12:39.144143] (MonitorAgent_1): Tproc: 4.8e-05
[2020-07-08 20:12:40.127624] (MultiMathAgent_1): Received: {'from': 'SineGeneratorAgent_1', 'data': array([0.47942554]), 'senderType': 'SineGeneratorAgent', 'channel': 'default'}
[2020-07-08 20:12:40.129917] (MultiMathAgent_1): Pack time: 0.000519
[2020-07-08 20:12:40.128351] (MathAgent_1): Received: {'from': 'SineGeneratorAgent_1', 'data': array([0.47942554]), 'senderType': 'SineGeneratorAgent', 'channel': 'default'}
[2020-07-08 20:12:40.125310] (SineGeneratorAgent_1): Pack time: 0.000539
[2020-07-08 20:12:40.127820] (MonitorAgent_1): Received: {'from': 'SineGeneratorAgent_1', 'data': array([0.47942554]), 'senderType': 'SineGeneratorAgent', 'channel': 'default'}
[2020-07-08 20:12:40.131738] (MultiMathAgent_1): Sending: {'minus': array([-0.02057446]), 'plus': array([0.97942554])}
[2020-07-08 20:12:40.129241] (MathAgent_1): Pack time: 0.000428
[2020-07-08 20:12:40.127622] (SineGeneratorAgent_1): Sending: [0.47942554]
[2020-07-08 20:12:40.132038] (MonitorAgent_1): Memory: {'SineGeneratorAgent_1': array([0.        , 0.47942554]), 'MultiMathAgent_1': {'minus': array([-0.5]), 'plus': array([0.5])}, 'MathAgent_1': array([0.])}
[2020-07-08 20:12:40.132147] (MultiMathAgent_1): Tproc: 0.004128
[2020-07-08 20:12:40.130428] (MathAgent_1): Sending: [0.23971277]
[2020-07-08 20:12:40.132367] (MonitorAgent_1): Tproc: 0.004143
[2020-07-08 20:12:40.135466] (MonitorAgent_1): Received: {'from': 'MathAgent_1', 'data': array([0.23971277]), 'senderType': 'MathAgent', 'channel': 'default'}
[2020-07-08 20:12:40.130764] (MathAgent_1): Tproc: 0.002032
[2020-07-08 20:12:40.156668] (MonitorAgent_1): Memory: {'SineGeneratorAgent_1': array([0.        , 0.47942554]), 'MultiMathAgent_1': {'minus': array([-0.5]), 'plus': array([0.5])}, 'MathAgent_1': array([0.        , 0.23971277])}
[2020-07-08 20:12:40.161541] (MonitorAgent_1): Tproc: 0.024927
[2020-07-08 20:12:40.164795] (MonitorAgent_1): Received: {'from': 'MultiMathAgent_1', 'data': {'minus': array([-0.02057446]), 'plus': array([0.97942554])}, 'senderType': 'MultiMathAgent', 'channel': 'default'}
[2020-07-08 20:12:40.171365] (MonitorAgent_1): Memory: {'SineGeneratorAgent_1': array([0.        , 0.47942554]), 'MultiMathAgent_1': {'minus': array([-0.5       , -0.02057446]), 'plus': array([0.5       , 0.97942554])}, 'MathAgent_1': array([0.        , 0.23971277])}
[2020-07-08 20:12:40.171872] (MonitorAgent_1): Tproc: 0.006705

 * Running on http://127.0.0.1:8050/ (Press CTRL+C to quit)
127.0.0.1 - - [08/Jul/2020 22:12:40] "POST /_dash-update-component HTTP/1.1" 200 -
127.0.0.1 - - [08/Jul/2020 22:12:40] "POST /_dash-update-component HTTP/1.1" 200 -