How To Create Your Own Components

This guide lays out how you can implement your own components using KOSMOS. Most parts of KOSMOS are implemented as small, swappable classes (often abstract base classes) that you can subclass.

The general idea

• Pick the right base class and subclass it. This also inherits the constructor of the base class (usually no need to override it).

• Implement the required methods by overriding the base class methods

• In some cases, you may need to implement additional custom subclasses to achieve your goal

• Looking at existing implementations of similar components for guidance, as well as the documentation, may help

Example: A custom routing algorithm

In this example, we want to implement our own routing algorithm as a custom protocol in KOSMOS. For that, we need to know how routing protocols are defined in KOSMOS. The base class for routing protocols is RoutingProtocol (which extends the general protocol base class Protocol).

When approaching the implementation of a custom routing protocol, looking at existing implementations may help. In our case, DijkstraRoutingProtocol is an analogous implementation of a routing protocol we can refer to.

We implement our custom routing algorithm by subclassing the RoutingProtocol base class. There is one required method we need to implement: execute(self).

To understand which attributes and methods are available in our custom protocol class, we can refer to the base class. In this case, by looking at the RoutingProtocol and Protocol base classes, we find useful attributes such as self.network, self.source_node, self.target_node, self.status, and self.config that we can use in our implementation. By going through this, we will find that self.config holds the configuration for the routing protocol as a RoutingProtocolConfig, which specifies parameters such as allowed link types and the cost function (we ignore costs in our simple example)

As execute(self) should return a RoutingProtocolResult, we also need to construct and return this object containing our results.

In the following, we implement our custom protocol that only succeeds if the source and target nodes are direct neighbors (i.e., connected by a single link of an allowed type).

from kosmos.protocols.protocol_result import RoutingProtocolResult
from kosmos.protocols.routing.path import Path
from kosmos.protocols.routing.routing import RoutingProtocol
from kosmos.protocols.status import ProtocolStatus


class DirectNeighborRoutingProtocol(RoutingProtocol):
    """Succeeds only if source and target share a direct allowed link."""

    def execute(self) -> RoutingProtocolResult:
        """Execute the direct neighbor routing protocol."""
        self.status = ProtocolStatus.RUNNING

        # Iterate over the outgoing links of the source node to find the target node
        for link in self.network.outgoing_links(self.source_node):
            # Check if the type of the link is allowed
            if link.type not in self.config.allowed_link_types:
                continue

            # Determine the node on the other end of this link (relative to the source)
            neighbor = link.dst if link.src == self.source_node else link.src

            # Only accept the link if it leads directly to the target
            if neighbor != self.target_node:
                continue

            # Create the path and calculate the total cost
            path = Path(nodes=[self.source_node, self.target_node], links=[link])
            total_cost = self._link_cost(link)

            # Set the status to SUCCESS
            self.status = ProtocolStatus.SUCCESS

            # Build the result object for this protocol execution
            # execute() must return a RoutingProtocolResult
            return RoutingProtocolResult(
                status=self.status,
                path=path,
                total_cost=total_cost,
                total_distance=None,
            )

        self.status = ProtocolStatus.FAILED
        return RoutingProtocolResult(
            status=self.status,
            path=None,
            total_cost=None,
            total_distance=None,
        )

An extended example that also creates a simple network and runs the custom routing protocol can be found in the examples under the Custom Routing Protocol section.