Source code for eclypse.placement.strategies.random

"""Module for the Random placement strategy.

It overrides the `place` method of the
PlacementStrategy class to place services of an application on infrastructure nodes
randomly.
"""

from __future__ import annotations

import os
import random as rnd
from typing import (
    TYPE_CHECKING,
)

from eclypse.utils.constants import RND_SEED

from .strategy import PlacementStrategy

if TYPE_CHECKING:
    from eclypse.graph import (
        Application,
        Infrastructure,
    )
    from eclypse.placement import (
        Placement,
        PlacementView,
    )




class RandomStrategy(PlacementStrategy):
    """A placement strategy that places services randomly onto nodes."""



    def __init__(self, spread: bool = False, seed: int | None = None):
        """Initializes the Random placement strategy.

        Args:
            spread (bool, optional):
                Whether to spread the services across different nodes.
                Defaults to False.
            seed (int | None, optional):
                The seed for the random number generator. Defaults to
                None.
        """
        self._rnd = rnd.Random(seed if seed is not None else os.environ[RND_SEED])
        self.spread = spread
        super().__init__()




    def place(
        self,
        _: Infrastructure,
        application: Application,
        __: dict[str, Placement],
        placement_view: PlacementView,
    ) -> dict[str, str]:
        """Places the services of an application on the infrastructure nodes, randomly.

        Args:
            _ (Infrastructure): The infrastructure to place the application on.
            application (Application): The application to place on the infrastructure.
            __ (dict[str, Placement]):
                The placement of all the applications in the simulations.
            placement_view (PlacementView):
                The snapshot of the current state of the infrastructure.

        Returns:
            dict[str, str]: A mapping of services to infrastructure nodes.
        """
        infrastructure_nodes = list(placement_view.residual.nodes())
        if not infrastructure_nodes:
            return {}

        self._rnd.shuffle(infrastructure_nodes)

        if self.spread:
            return {
                service: infrastructure_nodes[i % len(infrastructure_nodes)]
                for i, service in enumerate(application.nodes)
            }

        return {
            service: self._rnd.choice(infrastructure_nodes)
            for service in application.nodes
        }