mirror of
https://github.com/blackboxprogramming/lucidia.git
synced 2026-03-17 03:57:13 -05:00
126 lines
4.3 KiB
Python
126 lines
4.3 KiB
Python
"""
|
|
Graph/Network Mirror Module
|
|
|
|
This module implements the mirror operator Psi' and breath operator B for directed graphs
|
|
represented by adjacency matrices. The mirror split decomposes a square adjacency
|
|
matrix into its symmetric (undirected) part and antisymmetric (orientation) part.
|
|
The breath update combines previous and current adjacency matrices to evolve the network
|
|
while preserving the original out-degree distribution. A delta_kick randomly toggles edges.
|
|
|
|
Functions:
|
|
- mirror_split_network(A): return symmetric and antisymmetric parts of adjacency matrix A.
|
|
- degree_distribution(A): return row-sum of adjacency matrix (out-degree).
|
|
- breath_update(A, target_deg=None): evolve A by squaring and normalizing rows to match target_deg.
|
|
- delta_kick(A, strength=1): randomly toggles directed edges.
|
|
- run_network_demo(...): demonstration of mirror and breath on a random graph; saves results to out_network/.
|
|
|
|
Usage:
|
|
python graph_network_mirror.py
|
|
"""
|
|
|
|
import os
|
|
import numpy as np
|
|
import json
|
|
import csv
|
|
|
|
|
|
def mirror_split_network(A: np.ndarray):
|
|
"""Return symmetric and antisymmetric parts of adjacency matrix A."""
|
|
A = A.astype(float)
|
|
sym = (A + A.T) / 2.0
|
|
anti = (A - A.T) / 2.0
|
|
return sym, anti
|
|
|
|
|
|
def degree_distribution(A: np.ndarray) -> np.ndarray:
|
|
"""Return out-degree distribution (row sums) of adjacency matrix A."""
|
|
return np.sum(A, axis=1)
|
|
|
|
|
|
def breath_update(A: np.ndarray, target_deg: np.ndarray = None) -> np.ndarray:
|
|
"""
|
|
Update adjacency matrix by a single 'breath' step.
|
|
We square A (compute two-step connectivity) and normalize row sums to match target_deg.
|
|
"""
|
|
if target_deg is None:
|
|
target_deg = degree_distribution(A)
|
|
# Multiply adjacency by itself (two steps)
|
|
B = A.dot(A)
|
|
# Compute new row sums
|
|
row_sums = degree_distribution(B)
|
|
# Initialize next matrix as copy of B
|
|
A_next = B.copy()
|
|
for i, (deg0, deg_new) in enumerate(zip(target_deg, row_sums)):
|
|
if deg_new > 0:
|
|
A_next[i, :] = B[i, :] * (deg0 / deg_new)
|
|
else:
|
|
A_next[i, :] = B[i, :]
|
|
return A_next
|
|
|
|
|
|
def delta_kick(A: np.ndarray, strength: int = 1) -> np.ndarray:
|
|
"""
|
|
Apply a delta-kick to adjacency matrix A by toggling 'strength' random edges.
|
|
Each toggle flips the presence/absence of a directed edge (except self-loops).
|
|
"""
|
|
n = A.shape[0]
|
|
A = A.copy()
|
|
for _ in range(strength):
|
|
i = np.random.randint(n)
|
|
j = np.random.randint(n)
|
|
if i == j:
|
|
continue
|
|
A[i, j] = 1.0 - A[i, j]
|
|
return A
|
|
|
|
|
|
def run_network_demo(
|
|
n_nodes: int = 5,
|
|
n_steps: int = 12,
|
|
kick_step: int = 6,
|
|
kick_strength: int = 2,
|
|
seed: int = 0,
|
|
) -> dict:
|
|
"""
|
|
Demonstrate the network mirror and breath operators on a random directed graph.
|
|
Generates a random adjacency matrix, computes symmetric/antisymmetric parts,
|
|
applies breath updates, introduces a delta-kick, and records degree variance.
|
|
Results are saved to out_network/ as CSV and JSON.
|
|
"""
|
|
np.random.seed(seed)
|
|
# Generate random adjacency matrix with approx 30% connectivity
|
|
A = (np.random.rand(n_nodes, n_nodes) < 0.3).astype(float)
|
|
# Remove self-loops
|
|
np.fill_diagonal(A, 0)
|
|
# Compute target degree distribution for invariance
|
|
target_deg = degree_distribution(A)
|
|
history = {"step": [], "degree_var": []}
|
|
for t in range(n_steps):
|
|
if t == kick_step:
|
|
A = delta_kick(A, strength=kick_strength)
|
|
# Breath update: square and renormalize to target degrees
|
|
A = breath_update(A, target_deg)
|
|
current_deg = degree_distribution(A)
|
|
diff = current_deg - target_deg
|
|
history["step"].append(t)
|
|
history["degree_var"].append(float(np.var(diff)))
|
|
# Prepare output directory
|
|
out_dir = "out_network"
|
|
os.makedirs(out_dir, exist_ok=True)
|
|
# Save history to CSV
|
|
csv_path = os.path.join(out_dir, "degree_variance.csv")
|
|
with open(csv_path, "w", newline="") as csvfile:
|
|
writer = csv.writer(csvfile)
|
|
writer.writerow(["step", "degree_variance"])
|
|
for s, var in zip(history["step"], history["degree_var"]):
|
|
writer.writerow([s, var])
|
|
# Save history to JSON
|
|
json_path = os.path.join(out_dir, "degree_variance.json")
|
|
with open(json_path, "w") as f:
|
|
json.dump(history, f, indent=2)
|
|
return history
|
|
|
|
|
|
if __name__ == "__main__":
|
|
run_network_demo()
|