"""Star infrastructure generator.
This module provides a generator for building star-shaped network
topologies, where a central node is connected directly to multiple
peripheral client nodes. The centre can be configured separately from
the outer nodes in terms of resource assets and roles.
This topology is useful for modelling centralised communication systems
such as access points, base stations, master-slave control, or
cloud-edge scenarios with a hub-spoke structure.
"""
from __future__ import annotations
from typing import (
TYPE_CHECKING,
Any,
)
from eclypse.graph import Infrastructure
if TYPE_CHECKING:
from collections.abc import Callable
import networkx as nx
from eclypse.graph.assets import Asset
from eclypse.utils.types import (
InitPolicy,
UpdatePolicies,
)
[docs]
def get_star(
n_clients: int,
infrastructure_id: str = "star",
symmetric: bool = False,
update_policies: UpdatePolicies = None,
node_assets: dict[str, Asset] | None = None,
link_assets: dict[str, Asset] | None = None,
center_assets_values: dict[str, Any] | None = None,
outer_assets_values: dict[str, Any] | None = None,
include_default_assets: bool = False,
strict: bool = False,
resource_init: InitPolicy = "min",
path_algorithm: Callable[[nx.Graph, str, str], list[str]] | None = None,
seed: int | None = None,
):
"""Create a star infrastructure with `n_clients` clients around a central node.
The group of the clients can be specified.
Args:
n_clients (int): The number of clients in the infrastructure.
infrastructure_id (str): The ID of the infrastructure.
symmetric (bool): Whether the links are symmetric. Defaults to False.
update_policies (Callable | list[Callable] | None):
Graph update policies. Defaults to None.
node_assets (dict[str, Asset] | None):
The assets for the nodes. Defaults to None.
link_assets (dict[str, Asset] | None):
The assets for the links. Defaults to None.
center_assets_values (dict[str, Any] | None):
The assets for the center node. Defaults to None.
outer_assets_values (dict[str, Any] | None):
The assets for the outer nodes. Defaults to None.
include_default_assets (bool):
Whether to include the default assets. Defaults to False.
strict (bool): If True, raises an error if the asset values are not \
consistent with their spaces. Defaults to False.
resource_init (InitPolicy):
The initialization policy for the resources. Defaults to "min".
path_algorithm (Callable[[nx.Graph, str, str], list[str]] | None):
The algorithm to compute the paths between nodes. Defaults to
None.
seed (int | None): The seed for the random number generator. Defaults to None.
Returns:
Infrastructure: The star infrastructure.
"""
infrastructure = Infrastructure(
infrastructure_id=infrastructure_id,
update_policies=update_policies,
node_assets=node_assets,
edge_assets=link_assets,
include_default_assets=include_default_assets,
resource_init=resource_init,
path_algorithm=path_algorithm,
seed=seed,
)
_outer_assets_values = outer_assets_values or {}
_center_assets_values = center_assets_values or {}
for i in range(n_clients):
infrastructure.add_node(f"outer_{i}", strict=strict, **_outer_assets_values)
infrastructure.add_node("center", strict=strict, **_center_assets_values)
for i in range(n_clients):
infrastructure.add_edge(
f"outer_{i}", "center", symmetric=symmetric, strict=strict
)
return infrastructure
