Merge pull request #117 from OFFIS-DAI/feature-topologies

Added topology feature + Switched to modern build tool.

.github/workflows/publish-mango.yml

@@ -18,8 +18,7 @@ jobs:
           pip install virtualenv
           virtualenv venv
           source venv/bin/activate
-          pip3 install -r requirements.txt
-          pip3 install -e .
+          pip3 install -e .[test]
           sudo apt update
           sudo apt install --assume-yes mosquitto
           sudo service mosquitto start
@@ -33,7 +32,8 @@ jobs:
       - name: Build package
         run: |
           source venv/bin/activate
-          python setup.py sdist bdist_wheel
+          python -m pip install build
+          python -m build
       - name: Publish package
         uses: pypa/gh-action-pypi-publish@release/v1
         with:

.github/workflows/test-mango.yml

@@ -34,8 +34,7 @@ jobs:
         source venv/bin/activate
         pip3 install -U sphinx
         pip3 install -r docs/requirements.txt
-        pip3 install -r requirements.txt
-        pip3 install -e .
+        pip3 install -e .[test]
         brew install mosquitto
         brew services start mosquitto
         pip3 install pytest coverage ruff
@@ -76,8 +75,7 @@ jobs:
         source venv/bin/activate
         pip3 install -U sphinx
         pip3 install -r docs/requirements.txt
-        pip3 install -r requirements.txt
-        pip3 install -e .
+        pip3 install -e .[test]
         sudo apt update
         sudo apt install --assume-yes mosquitto
         sudo service mosquitto start
@@ -95,5 +93,8 @@ jobs:
     - name: Test+Coverage
       run: |
         source venv/bin/activate
-        coverage run -m pytest
-        coverage report
+        pytest --cov --cov-report=xml
+    - uses: codecov/codecov-action@v4
+      with:
+        token: ${{ secrets.CODECOV_TOKEN  }}
+        fail_ci_if_error: false

docs/source/conf.py

@@ -11,7 +11,7 @@
 author = "mango team"
 
 # The full version, including alpha/beta/rc tags
-version = release = "2.0.0"
+version = release = "2.1.0"
 
 
 # -- General configuration ---------------------------------------------------

docs/source/index.rst

@@ -30,6 +30,7 @@ Features
    message exchange
    role-api
    scheduling
+   topology
    codecs
    api_ref/index
    migration

docs/source/topology.rst

@@ -0,0 +1,81 @@
+================
+Topologies
+================
+
+The framework provides out of the box support for creating and distributing topologies to the agents.
+This can be done using either :meth:`mango.create_topology` or :meth:`mango.per_node`. With the first method
+a topology has to be created from scratch, and for every node you need to assign the agents.
+
+.. testcode::
+
+    import asyncio
+    from typing import Any
+
+    from mango import Agent, run_with_tcp, create_topology
+
+    class TopAgent(Agent):
+        counter: int = 0
+
+        def handle_message(self, content, meta: dict[str, Any]):
+            self.counter += 1
+
+    async def start_example():
+        agents = [TopAgent(), TopAgent(), TopAgent()]
+        with create_topology() as topology:
+            id_1 = topology.add_node(agents[0])
+            id_2 = topology.add_node(agents[1])
+            id_3 = topology.add_node(agents[2])
+            topology.add_edge(id_1, id_2)
+            topology.add_edge(id_1, id_3)
+
+        async with run_with_tcp(1, *agents):
+            for neighbor in agents[0].neighbors():
+                await agents[0].send_message("hello neighbors", neighbor)
+            await asyncio.sleep(0.1)
+
+        print(agents[1].counter)
+        print(agents[2].counter)
+
+    asyncio.run(start_example())
+
+.. testoutput::
+
+    1
+    1
+
+The other method would be to use :meth:`mango.per_node`. Here you want to create a topology beforehand, this can be done using :meth:`mango.custom_topology` by providing
+a networkx Graph.
+
+.. testcode::
+
+    import asyncio
+    from typing import Any
+
+    from mango import Agent, run_with_tcp, per_node, complete_topology
+
+    class TopAgent(Agent):
+        counter: int = 0
+
+        def handle_message(self, content, meta: dict[str, Any]):
+            self.counter += 1
+
+    async def start_example():
+        topology = complete_topology(3)
+        for node in per_node(topology):
+            agent = TopAgent()
+            node.add(agent)
+
+        async with run_with_tcp(1, *topology.agents):
+            for neighbor in topology.agents[0].neighbors():
+                await topology.agents[0].send_message("hello neighbors", neighbor)
+            await asyncio.sleep(0.1)
+
+        print(topology.agents[1].counter)
+        print(topology.agents[2].counter)
+
+    asyncio.run(start_example())
+
+.. testoutput::
+
+    1
+    1

mango/__init__.py

@@ -23,3 +23,10 @@
     PROTOBUF,
     SerializationError,
 )
+from .express.topology import (
+    Topology,
+    create_topology,
+    complete_topology,
+    per_node,
+    custom_topology,
+)

mango/agent/core.py

@@ -9,6 +9,7 @@
 import logging
 from abc import ABC
 from dataclasses import dataclass
+from enum import Enum
 from typing import Any
 
 from ..util.clock import Clock
@@ -23,6 +24,21 @@ class AgentAddress:
     aid: str
 
 
+class State(Enum):
+    NORMAL = 0  # normal neighbor
+    INACTIVE = (
+        1  # neighbor link exists but link is not active (could be activated/used)
+    )
+    BROKEN = 2  # neighbor link exists but link is not usable (can not be activated)
+
+
+class TopologyService:
+    state_to_neighbors: dict[State, list] = dict()
+
+    def neighbors(self, state: State = State.NORMAL):
+        return [f() for f in self.state_to_neighbors.get(state, [])]
+
+
 class AgentContext:
     def __init__(self, container) -> None:
         self._container = container
@@ -69,6 +85,7 @@ def __init__(self) -> None:
         self.context: AgentContext = None
         self.scheduler: Scheduler = None
         self._aid = None
+        self._services = {}
 
     def on_start(self):
         """Called when container started in which the agent is contained"""
@@ -344,6 +361,28 @@ async def tasks_complete(self, timeout=1):
         """
         await self.scheduler.tasks_complete(timeout=timeout)
 
+    def service_of_type(self, type: type, default: Any = None) -> Any:
+        """Return the service of the type ``type`` or set the default as service and return it.
+
+        :param type: the type of the service
+        :type type: type
+        :param default: the default if applicable
+        :type default: Any (optional)
+        :return: the service
+        :rtype: Any
+        """
+        if type not in self._services:
+            self._services[type] = default
+        return self._services[type]
+
+    def neighbors(self, state: State = State.NORMAL) -> list[AgentAddress]:
+        """Return the neighbors of the agent (controlled by the topology api).
+
+        :return: the list of agent addresses filtered by state
+        :rtype: list[AgentAddress]
+        """
+        return self.service_of_type(TopologyService).neighbors(state)
+
 
 class Agent(ABC, AgentDelegates):
     """Base class for all agents."""

mango/agent/role.py

@@ -34,7 +34,8 @@
 
 import asyncio
 from abc import ABC
-from typing import Any, Callable
+from collections.abc import Callable
+from typing import Any
 
 from mango.agent.core import Agent, AgentAddress, AgentDelegates
 
@@ -67,15 +68,14 @@ class Role:
 class RoleHandler:
     """Contains all roles and their models. Implements the communication between roles."""
 
-    def __init__(self, agent_context, scheduler):
+    def __init__(self, scheduler):
         self._role_models = {}
         self._roles = []
         self._role_to_active = {}
         self._role_model_type_to_subs = {}
         self._message_subs = []
         self._send_msg_subs = {}
         self._role_event_type_to_handler = {}
-        self._agent_context = agent_context
         self._scheduler = scheduler
         self._data = DataContainer()
 
@@ -252,7 +252,6 @@ def __init__(
         inbox,
     ):
         super().__init__()
-        self._agent_context = None
         self._role_handler = role_handler
         self._aid = aid
         self._inbox = inbox
@@ -266,17 +265,9 @@ def data(self):
         """
         return self._get_container()
 
-    @property
-    def context(self) -> float:
-        return self._agent_context
-
-    @context.setter
-    def context(self, value: bool):
-        pass
-
     @property
     def current_timestamp(self) -> float:
-        return self._agent_context.current_timestamp
+        return self.context.current_timestamp
 
     def _get_container(self):
         return self._role_handler._data
@@ -342,7 +333,7 @@ async def send_message(
         **kwargs,
     ):
         self._role_handler._notify_send_message_subs(content, receiver_addr, **kwargs)
-        return await self._agent_context.send_message(
+        return await self.context.send_message(
             content=content,
             receiver_addr=receiver_addr,
             sender_id=self.aid,
@@ -396,7 +387,7 @@ def __init__(self):
                               Using the generated aid-style ("agentX") is not allowed.
         """
         super().__init__()
-        self._role_handler = RoleHandler(None, None)
+        self._role_handler = RoleHandler(None)
         self._role_context = RoleContext(self._role_handler, self.aid, self.inbox)
 
     def on_start(self):
@@ -406,8 +397,7 @@ def on_ready(self):
         self._role_context.on_ready()
 
     def on_register(self):
-        self._role_context._agent_context = self.context
-        self._role_handler._agent_context = self.context
+        self._role_context.context = self.context
         self._role_context.scheduler = self.scheduler
         self._role_handler._scheduler = self.scheduler
         self._role_context._aid = self.aid

mango/container/tcp.py

@@ -220,7 +220,7 @@ async def send_message(
         protocol_addr = receiver_addr.protocol_addr
         if isinstance(protocol_addr, str) and ":" in protocol_addr:
             protocol_addr = protocol_addr.split(":")
-        elif isinstance(protocol_addr, (tuple, list)) and len(protocol_addr) == 2:
+        elif isinstance(protocol_addr, tuple | list) and len(protocol_addr) == 2:
             protocol_addr = tuple(protocol_addr)
         else:
             logger.warning("Address for sending message is not valid;%s", protocol_addr)
@@ -257,7 +257,7 @@ async def _send_external_message(self, addr, message, meta) -> bool:
         :param message: The message
         :return:
         """
-        if addr is None or not isinstance(addr, (tuple, list)) or len(addr) != 2:
+        if addr is None or not isinstance(addr, tuple | list) or len(addr) != 2:
             logger.warning(
                 "Sending external message not successful, invalid address; %s",
                 addr,