Utilities

Graph & Network

cognac.utils.graph_utils.generate_adjacency_matrix(n: int, gain: float = 0.3, p=0.2, symmetry: bool = False) → ndarray

Generate a random weighted adjacency matrix.

Parameters

nint: Number of nodes in the graph.
gainfloat, optional: Maximum value of weights (default is 0.3).
pfloat, optional: Probability of an edge between two nodes (default is 0.2).
symmetrybool, optional: If True, the matrix will be symmetric (undirected graph).

Returns

np.ndarray: A square matrix of shape (n, n) representing weighted edges.

cognac.utils.graph_utils.generate_band_adjacency_matrix(n: int, neighborhood_size: int, p_min: float = 0.2, p_max: float = 1.0) → ndarray

Generate a banded adjacency matrix where each node connects to neighbors within a band.

Parameters

nint: Number of nodes.
neighborhood_sizeint: Range of neighbors each node connects to on either side.
p_minfloat, optional: Minimum weight of an edge (default is 0.2).
p_maxfloat, optional: Maximum weight of an edge (default is 1.0).

Returns

np.ndarray: A banded weighted adjacency matrix of shape (n, n).

cognac.utils.graph_utils.generate_coordination_graph(matrix: ndarray) → DiGraph

Generate a directed NetworkX graph from an influence matrix.

Parameters

matrixnp.ndarray: A 2D square numpy array where entry (i, j) indicates the influence of node j on node i.

Returns

networkx.DiGraph: A directed graph representation of the influence matrix.

cognac.utils.graph_utils.generate_dag_influence_matrix(n: int, p: float = 1.0) → ndarray

Generate a random Directed Acyclic Graph (DAG) influence matrix.

Parameters

nint: Number of nodes.
pfloat, optional: Probability of an edge existing between two nodes (default is 1).

Returns

np.ndarray: An upper-triangular binary matrix representing a DAG.

cognac.utils.graph_utils.generate_deterministic_adjacency_matrix(n: int, p: float = 0.2, symmetry: bool = False) → ndarray

Generate a deterministic binary adjacency matrix with fixed edge probability.

Parameters

nint: Number of nodes.
pfloat, optional: Probability of an edge existing (default is 0.2).
symmetrybool, optional: Whether the matrix should be symmetric.

Returns

np.ndarray: A binary adjacency matrix of shape (n, n).

cognac.utils.graph_utils.plot_influence_graph(matrix: ndarray)

Plot a directed graph representing influences between nodes.

Each node represents an entity, and directed edges (with weights) indicate influence from one node to another.

Parameters

matrixnp.ndarray: A square 2D numpy array where entry (i, j) represents the influence of node j on node i.

Returns

None: Displays the plotted influence graph using matplotlib.

cognac.utils.graph_utils.to_stochastic_matrix(prob_matrix: ndarray) → ndarray

Convert a matrix into a row-stochastic matrix.

Each row is normalized so it sums to 1. If a row has all zeros, it is replaced with a uniform distribution over columns.

Parameters

prob_matrixnp.ndarray: A 2D numpy array representing transition probabilities or weights.

Returns

np.ndarray: A row-normalized stochastic matrix.

Rendering & Visualization

class cognac.utils.visualization.GifRenderingWrapper(env: ParallelEnv, duration: float = 0.3)

Bases: object

A wrapper for PettingZoo environments that captures and saves the environment’s visual trajectory as an animated GIF.

Warning

The gif rendering wrapper is still a work in progress and rely on proper r endering capability of environment. We are working on it to provide working solution to generate gif from trajectories.

This is useful for visualizing multi-agent interactions over time, particularly in notebook-based workflows or for producing documentation and demos.

Parameters

envpettingzoo.ParallelEnv: The environment to wrap. It must implement a render(fig, ax) method that draws the environment state using Matplotlib.
durationfloat, optional: Duration (in seconds) per frame in the output GIF. Default is 0.3.

Attributes

envpettingzoo.ParallelEnv: The original environment instance being wrapped.
frameslist: A list of image frames (as numpy arrays) captured during the environment rollout.
durationfloat: Time interval between frames in the generated GIF.

capture_frame(): Capture the current environment state as an image.

generate_gif(output_filename: str = 'trajectory.gif')

Generate and save a GIF from the captured frames.

Parameters

output_filenamestr, optional: Name of the output GIF file (default is “trajectory.gif”).

Returns

None

reset()

Reset the environment and clear any previously stored frames.

Returns

Any: The output of the wrapped environment’s reset method.

step(action_dict)

Step the environment forward using the provided actions and capture the frame.

Parameters

action_dictdict: A dictionary mapping agent names to their actions.

Returns

tuple: A tuple (obs, reward, done, trunc, info) as returned by the wrapped environment.