import random from typing import Any, List, Mapping, Optional, Sequence, Tuple from core import clamp, parse_mu_disp_mode, require, require_unit_interval_closed_open, require_unit_interval_open_closed from graph import SimpleGraph, add_edge, remove_edge from observables import EdgeCounts, mu_hat_by_comm_and_dispersion_from_counts, mu_risk_by_comm_and_dispersion_from_counts from schema_types import TopologyActuatorEvent def _choose_edge_to_remove( g: SimpleGraph, *, remove_strategy: str, low_nodes: Sequence[int], high_nodes: Sequence[int], rng: random.Random, max_tries: int, ) -> Optional[Tuple[int, int]]: if not g.edges: return None strat = str(remove_strategy).lower() if strat == "random_edge": return tuple(rng.choice(g.edges)) if strat == "within_low": s = {int(x) for x in low_nodes} elif strat == "within_high": s = {int(x) for x in high_nodes} else: raise ValueError(f"Unsupported topology_actuator.remove_strategy: {remove_strategy}") tries = max(1, int(max_tries)) for _ in range(tries): u, v = rng.choice(g.edges) if int(u) in s and int(v) in s: return (int(u), int(v)) return tuple(rng.choice(g.edges)) def _choose_bridge_edge( g: SimpleGraph, *, low_nodes: Sequence[int], high_nodes: Sequence[int], rng: random.Random, max_tries: int, ) -> Optional[Tuple[int, int]]: if not low_nodes or not high_nodes: return None tries = max(1, int(max_tries)) for _ in range(tries): u = int(rng.choice(low_nodes)) v = int(rng.choice(high_nodes)) if u == v: continue a, b = (u, v) if u < v else (v, u) if int(b) in g.adj[int(a)]: continue return (int(u), int(v)) return None def apply_topology_actuator( g: SimpleGraph, *, t: int, u: float, ethics_cfg: Mapping[str, Any], meso_cfg: Mapping[str, Any], comms: Sequence[Sequence[int]], comm_obs: Optional[Sequence[EdgeCounts]], meso_alpha0: Optional[Sequence[float]], cfg: Mapping[str, Any], rng: random.Random, risk_rng: Optional[random.Random] = None, ) -> List[TopologyActuatorEvent]: if not cfg: return [] if not bool(cfg.get("enabled", True)): return [] typ = str(require(cfg, "type", "topology_actuator")).lower() if typ not in ("bridge_low_high", "mu_disp_bridge", "bridge_quantiles"): raise ValueError(f"Unsupported topology_actuator.type: {typ}") every_steps = int(require(cfg, "every_steps", "topology_actuator")) if every_steps <= 0: raise ValueError("topology_actuator.every_steps must be > 0") if (int(t) % int(every_steps)) != 0: return [] mu_disp_max = ethics_cfg.get("mu_disp_max") if mu_disp_max is None: return [] if comm_obs is None: return [] if meso_alpha0 is None: raise ValueError("topology_actuator: meso_alpha0 is required for posterior meso actuation") min_total_edges = float(meso_cfg.get("min_total_edges", 1.0)) mu_disp_mode = parse_mu_disp_mode(ethics_cfg) if mu_disp_mode == "risk": mu_min = float(require(ethics_cfg, "mu_min", "ethics")) if "risk_k_mc" in cfg: risk_k_mc = int(cfg["risk_k_mc"]) else: risk_k_mc = int(ethics_cfg.get("risk_k_mc", 64)) if risk_k_mc <= 0: raise ValueError("risk_k_mc must be > 0 (set topology_actuator.risk_k_mc or ethics.risk_k_mc)") risk_rng_eff = risk_rng if risk_rng is not None else rng mu_pairs, mu_disp_hat = mu_risk_by_comm_and_dispersion_from_counts( comm_obs, meso_alpha0=meso_alpha0, mu_min=float(mu_min), min_total_edges=min_total_edges, k_mc=int(risk_k_mc), rng=risk_rng_eff, ) else: mu_pairs, mu_disp_hat = mu_hat_by_comm_and_dispersion_from_counts( comm_obs, meso_alpha0=meso_alpha0, min_total_edges=min_total_edges, ) if len(mu_pairs) < 2: return [] margin = float(mu_disp_hat) - float(mu_disp_max) if margin <= 0.0: return [] budget_max = int(require(cfg, "budget_max", "topology_actuator")) if budget_max <= 0: return [] u_min = require_unit_interval_closed_open(require(cfg, "u_min", "topology_actuator"), "topology_actuator.u_min") u_power = float(require(cfg, "u_power", "topology_actuator")) disp_power = float(require(cfg, "disp_power", "topology_actuator")) remove_strategy = str(require(cfg, "remove_strategy", "topology_actuator")).lower() max_tries = int(require(cfg, "max_tries", "topology_actuator")) if max_tries <= 0: raise ValueError("topology_actuator.max_tries must be > 0") uu = clamp(float(u), 0.0, 1.0) u_gate = clamp((uu - float(u_min)) / (1.0 - float(u_min)), 0.0, 1.0) disp_gate = 0.0 if float(mu_disp_max) > 0.0: disp_gate = clamp(float(margin) / float(mu_disp_max), 0.0, 1.0) elif float(margin) > 0.0: disp_gate = 1.0 drive = (u_gate**float(u_power)) * (disp_gate**float(disp_power)) drive = clamp(float(drive), 0.0, 1.0) budget = int(round(float(drive) * float(budget_max))) if budget <= 0: return [] mu_pairs_sorted = sorted(mu_pairs, key=lambda x: float(x[1])) if typ in ("bridge_low_high", "mu_disp_bridge"): if mu_disp_mode == "risk": low_comm = int(mu_pairs_sorted[-1][0]) high_comm = int(mu_pairs_sorted[0][0]) mu_low = float(mu_pairs_sorted[-1][1]) mu_high = float(mu_pairs_sorted[0][1]) else: low_comm = int(mu_pairs_sorted[0][0]) high_comm = int(mu_pairs_sorted[-1][0]) mu_low = float(mu_pairs_sorted[0][1]) mu_high = float(mu_pairs_sorted[-1][1]) elif typ == "bridge_quantiles": low_frac = require_unit_interval_open_closed(require(cfg, "low_frac", "topology_actuator"), "topology_actuator.low_frac") high_frac = require_unit_interval_open_closed(require(cfg, "high_frac", "topology_actuator"), "topology_actuator.high_frac") n_mu = int(len(mu_pairs_sorted)) low_n = max(1, int(float(low_frac) * float(n_mu))) high_n = max(1, int(float(high_frac) * float(n_mu))) if mu_disp_mode == "risk": low_cands = mu_pairs_sorted[-int(low_n) :] high_cands = mu_pairs_sorted[: int(high_n)] else: low_cands = mu_pairs_sorted[: int(low_n)] high_cands = mu_pairs_sorted[-int(high_n) :] if not low_cands or not high_cands: return [] chosen = None tries = max(1, int(max_tries)) for _ in range(tries): lo = rng.choice(low_cands) hi = rng.choice(high_cands) if int(lo[0]) != int(hi[0]): chosen = (lo, hi) break if chosen is None: return [] low_comm = int(chosen[0][0]) mu_low = float(chosen[0][1]) high_comm = int(chosen[1][0]) mu_high = float(chosen[1][1]) else: raise ValueError(f"Unsupported topology_actuator.type: {typ}") low_nodes = list(comms[int(low_comm)]) if int(low_comm) < len(comms) else [] high_nodes = list(comms[int(high_comm)]) if int(high_comm) < len(comms) else [] if not low_nodes or not high_nodes: return [] rewired = 0 for _ in range(int(budget)): e_rem = _choose_edge_to_remove( g, remove_strategy=remove_strategy, low_nodes=low_nodes, high_nodes=high_nodes, rng=rng, max_tries=max_tries, ) if e_rem is None: break e_add = _choose_bridge_edge(g, low_nodes=low_nodes, high_nodes=high_nodes, rng=rng, max_tries=max_tries) if e_add is None: break remove_edge(g, int(e_rem[0]), int(e_rem[1])) add_edge(g, int(e_add[0]), int(e_add[1])) rewired += 1 if rewired <= 0: return [] return [ { "type": "topology_actuator", "t": int(t), "mode": str(typ), "rewired": int(rewired), "budget": int(budget), "u": float(uu), "drive": float(drive), "mu_disp_hat": float(mu_disp_hat), "mu_disp_max": float(mu_disp_max), "low_comm": int(low_comm), "high_comm": int(high_comm), "mu_low": float(mu_low), "mu_high": float(mu_high), "remove_strategy": str(remove_strategy), } ]