Random

abses.utils.random.ListRandom

ListRandom(model, actors)

Bases: Random

Create a random generator from an ActorsList.

Inherits from Python's Random class to provide Mesa-compatible shuffle() method. Extends Random with ABSESpy-specific methods for working with actors.

Source code in abses/utils/random.py

def __init__(self, model: MainModelProtocol, actors: Iterable[Any]) -> None:
    # Get seed from model and ensure it's a valid type
    if hasattr(model, "_seed"):
        seed = model._seed
        # Ensure seed is a valid type for Random
        if seed is not None and not isinstance(
            seed, (int, float, str, bytes, bytearray)
        ):
            seed = None
    else:
        seed = None

    # Initialize parent Random with seed
    super().__init__(seed)

    self.model = model
    self.actors = self._to_actors_list(actors)
    self.rng = model.rng if model.rng else np.random.default_rng()
    self.seed = seed

clean_p

clean_p(prob)

Clean the probabilities. Any negative values, NaN values, or zeros will be recognized as in-valid probabilities. For all valid probabilities, normalize them into a prob-array (the sum is equal to 1.0).

Parameters:

Name	Type	Description	Default
prob	Union[ndarray, str]	An array-like numbers of probabilities.	required

Returns:

Type	Description
ndarray	The probabilities after cleaned.

Example:

>>> clean_p([0, 0])
>>> [0.5, 0.5]

>>> clean_p([-1, np.nan])
>>> [0.5, 0.5]

>>> clean_p([3, 2])
>>> [0.6, 0.4]

Source code in abses/utils/random.py

def clean_p(self, prob: Union[np.ndarray, str]) -> np.ndarray:
    """Clean the probabilities.
    Any negative values, NaN values, or zeros will be recognized as in-valid probabilities.
    For all valid probabilities, normalize them into a prob-array (the sum is equal to 1.0).

    Parameters:
        prob:
            An array-like numbers of probabilities.

    Returns:
        The probabilities after cleaned.

    Example:
    ```
    >>> clean_p([0, 0])
    >>> [0.5, 0.5]

    >>> clean_p([-1, np.nan])
    >>> [0.5, 0.5]

    >>> clean_p([3, 2])
    >>> [0.6, 0.4]
    ```
    """
    if isinstance(prob, str):
        prob = self.actors.array(attr=prob)
    else:
        prob = np.array(make_list(prob))
    length = len(prob)
    prob = np.nan_to_num(prob)
    prob[prob < 0] = 0.0
    total = prob.sum()
    prob = prob / total if total else np.repeat(1 / length, length)
    return prob

choice

choice(
    size: int = 1,
    prob: ndarray | None = None,
    replace: bool = False,
    as_list: bool = True,
    when_empty: WHEN_EMPTY = "raise exception",
) -> ActorsList[ActorProtocol]

choice(
    size: int = 1,
    prob: ndarray | None = None,
    replace: bool = False,
    as_list: bool = False,
    when_empty: WHEN_EMPTY = "raise exception",
) -> ActorProtocol | ActorsList[ActorProtocol]

choice(
    size=1,
    prob=None,
    replace=False,
    as_list=False,
    when_empty="raise exception",
    double_check=False,
)

Randomly choose one or more actors from the current self object.

Source code in abses/utils/random.py

def choice(
    self,
    size: int = 1,
    prob: np.ndarray | None | str = None,
    replace: bool = False,
    as_list: bool = False,
    when_empty: WHEN_EMPTY = "raise exception",
    double_check: bool = False,
) -> Optional[ActorProtocol | ActorsList[ActorProtocol] | list]:
    """Randomly choose one or more actors from the current self object."""
    instances_num = len(self.actors)
    if instances_num == 0:
        self._when_empty(when_empty=when_empty)
        return None
    if not isinstance(size, int):
        raise ValueError(f"{size} isn't an integer size.")
    if instances_num < size and not replace:
        raise ABSESpyError(f"Trying to choose {size} actors from {self.actors}.")
    # 有概率的时候，先清理概率
    if prob is not None:
        prob = self.clean_p(prob=prob)
        valid_prob = prob.astype(bool)
        # 特别处理有概率的主体数量不足预期的情况
        if valid_prob.sum() < size and not replace:
            return self._when_p_not_enough(double_check, prob, size, as_list)
        # 如果只有一个有效概率且需要重复选择，直接返回对应的 actor
        if valid_prob.sum() == 1 and replace and size > 1:
            idx = np.where(valid_prob)[0][0]
            chosen = [self.actors[idx]] * size
            return chosen if as_list else self._to_actors_list(chosen)
    # 其他情况就正常随机选择
    indices = np.arange(len(self.actors))
    chosen_indices = self.rng.choice(indices, size=size, replace=replace, p=prob)
    # 如果不允许重复，按索引排序
    if not replace:
        chosen_indices.sort()
    chosen = [self.actors[i] for i in chosen_indices]
    return (
        chosen[0]
        if size == 1 and not as_list
        else (chosen if as_list else self._to_actors_list(chosen))
    )

new

new(actor_cls, actor_attrs=None, **kwargs)

Randomly creating new agents for a given actor type.

Source code in abses/utils/random.py

def new(
    self,
    actor_cls: Type[Actor],
    actor_attrs: Optional[Dict[str, Any]] = None,
    **kwargs,
) -> ActorsList[Actor]:
    """Randomly creating new agents for a given actor type."""
    if actor_attrs is None:
        actor_attrs = {}
    cells = self.choice(as_list=True, **kwargs)
    # Ensure we operate on an ActorsList for chaining operations like `.apply()`
    cells = self._to_actors_list(cells)
    objs = cells.apply(
        lambda c: c.agents.new(breed_cls=actor_cls, singleton=True, **actor_attrs)
    )
    return self._to_actors_list(objs)

link

link(link, p=1.0, mutual=True)

Random build links between actors.

Parameters:

Name	Type	Description	Default
link	str	Name of the link.	required
p	float	Probability to generate a link.	1.0

Returns:

Type	Description
List[Tuple[Actor, Actor]]	A list of tuple, in each tuple, there are two actors who got linked.

Example

# generate three actors
actors = model.agents.new(Actor, 3)
# with `probability=1`, all possible actor-actor links would be generated.
>>> actors.random.link('test', p=1)
>>> a1, a2, a3 = actors
>>> assert a1.link.get('test) == [a2, a3]
>>> assert a2.link.get('test) == [a1, a3]
>>> assert a3.link.get('test) == [a1, a2]

Source code in abses/utils/random.py

def link(
    self, link: str, p: float = 1.0, mutual: bool = True
) -> List[Tuple[Actor, Actor]]:
    """Random build links between actors.

    Parameters:
        link:
            Name of the link.
        p:
            Probability to generate a link.

    Returns:
        A list of tuple, in each tuple, there are two actors who got linked.

    Example:
        ```
        # generate three actors
        actors = model.agents.new(Actor, 3)
        # with `probability=1`, all possible actor-actor links would be generated.
        >>> actors.random.link('test', p=1)
        >>> a1, a2, a3 = actors
        >>> assert a1.link.get('test) == [a2, a3]
        >>> assert a2.link.get('test) == [a1, a3]
        >>> assert a3.link.get('test) == [a1, a2]
        ```
    """
    linked_combs = []
    for source, target in list(combinations(self.actors, 2)):
        if self.rng.random() < p:
            source.link.to(target, link_name=link, mutual=mutual)
            linked_combs.append((source, target))
    return linked_combs

assign

assign(value, attr, when_empty='raise exception')

Randomly assign a value to each actor.

Source code in abses/utils/random.py

def assign(
    self,
    value: float | int,
    attr: str,
    when_empty: WHEN_EMPTY = "raise exception",
) -> np.ndarray:
    """Randomly assign a value to each actor."""
    num = len(self.actors)
    if num == 0:
        self._when_empty(when_empty=when_empty, operation="assign")
        return np.array([])
    if num == 1:
        values = np.array([value])
    else:
        # 生成 n-1 个随机切割点
        cuts = np.sort(self.rng.uniform(0, value, num - 1))
        # 将 0 和总面积 X 添加到切割点数组中，方便计算每段区间长度
        full_range = np.append(np.append(0, cuts), value)
        # 计算每个区间的长度，即为每个对象的分配面积
        values = np.diff(full_range)
    # 将分配的值赋予每个对象
    self.actors.update(attr, values)
    return values