#!/usr/bin/env python3 """CorvOS - Arkhe Distributed Simulator v3.0-Omega 13 nodes in Klein bottle topology with Varela ternary consensus """ import numpy as np, json, time, hashlib from datetime import datetime PHI = 1.618033988749895 PHASE_NAMES = {0: 'EXT', 1: 'INT', 2: 'AUTO'} class ArkheNode: def __init__(self, node_id, position_3d, district): self.id = node_id self.pos = np.array(position_3d, dtype=float) self.district = district self.phase = np.random.uniform(-np.pi, np.pi) self.lambda2 = np.random.uniform(0.90, 0.99) self.state = np.random.choice([0, 1, 2], p=[0.3, 0.3, 0.4]) self.omega = np.random.uniform(0.1, 1.0) self.entropy = np.random.uniform(0.0, 0.3) self.coherence_history = [self.lambda2] self.decision_history = [] self.neighbors = [] self.autonomous_count = 0 def varela_delta(self, neighbor_phases): if not neighbor_phases: return 0.0 phases = np.array(neighbor_phases) return np.mean(np.sin(phases - self.phase)) def kuramoto_coupling(self, neighbor_phases, K=0.5): return K * self.omega * self.varela_delta(neighbor_phases) def update(self, dt, neighbor_phases): coupling = self.kuramoto_coupling(neighbor_phases) self.phase += (self.omega + coupling) * dt self.phase = np.arctan2(np.sin(self.phase), np.cos(self.phase)) self.lambda2 = min(0.9999, self.lambda2 + 0.001 * np.cos(self.phase)) self.coherence_history.append(self.lambda2) if len(self.coherence_history) > 200: self.coherence_history.pop(0) def decide(self, stimulus): self.decision_history.append(stimulus) if self.state == 2 or self.lambda2 > 0.995: self.state = 2 self.autonomous_count += 1 return 2 elif abs(stimulus) > 0.5 * PHI: self.state = 1 if stimulus > 0 else 0 return self.state return self.state def quantum_coherence(self): rho = np.array([[self.lambda2, 0.01*np.exp(1j*self.phase)], [0.01*np.exp(-1j*self.phase), 1-self.lambda2]], dtype=complex) eigenvalues = np.linalg.eigvalsh(rho) return eigenvalues[1] - eigenvalues[0] class KleinBottleTopology: def __init__(self): self.nodes = [] self._build_klein_bottle() self._assign_districts() self._link_neighbors() def _build_klein_bottle(self): districts = ['urca', 'flamengo', 'botafogo', 'humaita', 'tijuca', 'centro', 'niteroi', 'saude', 'leme', 'copacabana', 'ipanema', 'leblon', 'gavea'] for i in range(13): theta = 2 * np.pi * i / 13 phi = 2 * np.pi * i / 13 x = (3 + np.cos(theta/2)*np.sin(phi) - np.sin(theta/2)*np.sin(2*phi)) * np.cos(theta) y = (3 + np.cos(theta/2)*np.sin(phi) - np.sin(theta/2)*np.sin(2*phi)) * np.sin(theta) z = np.sin(theta/2)*np.sin(phi) + np.cos(theta/2)*np.sin(2*phi) self.nodes.append(ArkheNode(i, [x, y, z], districts[i])) def _assign_districts(self): self.district_map = {} for n in self.nodes: self.district_map[n.district] = n def _link_neighbors(self): for n in self.nodes: n.neighbors = [self.nodes[(n.id - 1) % 13], self.nodes[(n.id + 1) % 13]] def global_order_parameter(self): phases = np.array([n.phase for n in self.nodes]) r = np.abs(np.mean(np.exp(1j * phases))) return r def kuramoto_order(self): phases = np.array([n.phase for n in self.nodes]) lambdas = np.array([n.lambda2 for n in self.nodes]) weighted = np.mean(lambdas * np.exp(1j * phases)) return np.abs(weighted), np.angle(weighted) def distribute_stimulus(self, stimulus): decisions = {} for n in self.nodes: d = n.decide(stimulus) decisions[n.id] = {'state': int(d), 'name': PHASE_NAMES[int(d)], 'lambda2': n.lambda2} return decisions def as_dict(self): return { 'nodes': [{'id': n.id, 'district': n.district, 'state': PHASE_NAMES[n.state], 'lambda2': n.lambda2, 'phase': float(n.phase), 'autonomous_pct': n.autonomous_count} for n in self.nodes], 'global_r': float(self.global_order_parameter()), 'weighted_lambda': float(self.kuramoto_order()[0]), 'timestamp': datetime.now().isoformat() } class ArkheConsensus: def __init__(self): self.round = 0 self.decisions = [] self.energy_history = [] def varela_free_energy(self, topology, stimulus): phases = np.array([n.phase for n in topology.nodes]) lambdas = np.array([n.lambda2 for n in topology.nodes]) J = 0.5 coupling = J * np.sum(np.cos(phases[:, None] - phases[None, :])) / (len(phases)**2) ext = -np.sum(lambdas * np.cos(phases - stimulus)) / len(phases) entropy = np.sum(np.sin(phases)**2) / len(phases) return coupling + ext + 0.1 * entropy def run_consensus_round(self, topology, stimulus): self.round += 1 for n in topology.nodes: neighbor_phases = [nb.phase for nb in n.neighbors] n.update(dt=0.05, neighbor_phases=neighbor_phases) decisions = topology.distribute_stimulus(stimulus) energy = self.varela_free_energy(topology, stimulus) self.energy_history.append(energy) states = [d['state'] for d in decisions.values()] counts = [states.count(0), states.count(1), states.count(2)] consensus = max(counts) / len(states) if states else 0 r, theta = topology.kuramoto_order() self.decisions.append({'round': self.round, 'consensus': consensus, 'r': r, 'theta': theta, 'energy': energy, 'counts': counts}) return consensus, r, energy, counts def run_full_simulation(): print('=' * 60) print('ARKHE DISTRIBUTED SIMULATOR v3.0-Omega') print('13 nodes | Klein bottle topology | Varela consensus') print('=' * 60) print() topology = KleinBottleTopology() consensus = ArkheConsensus() stimuli = [0.0, 0.5, 1.0, -0.5, PHI/2, 1.618, -1.0, 0.3, -0.3, 0.9, 0.0, -0.9, 0.7] print(f'{"Ronda":>5} | {"Estímulo":>8} | {"Consenso":>8} | {"r (Kuram)":>10} | {"Energia":>10} | Estados') print('-' * 75) for stim in stimuli: c, r, e, counts = consensus.run_consensus_round(topology, stim) names = [PHASE_NAMES[i] for i in [0,1,2]] print(f'{consensus.round:>5} | {stim:>8.4f} | {c:>8.3f} | {r:>10.6f} | {e:>10.6f} | {names[0]}={counts[0]:2d} {names[1]}={counts[1]:2d} {names[2]}={counts[2]:2d}') print() r_final, theta_final = topology.kuramoto_order() print(f' Ordem global final: r={r_final:.6f}, theta={theta_final:.4f} rad') print(f' Lambda medio: {np.mean([n.lambda2 for n in topology.nodes]):.6f}') auto_count = sum(1 for n in topology.nodes if n.state == 2) print(f' Nos em estado AUTO: {auto_count}/13') print() result = topology.as_dict() result['consensus_rounds'] = consensus.round result['energy_trajectory'] = consensus.energy_history result['simulator_version'] = '3.0-Omega' with open('/home/workspace/corvos_sys/arkhe_simulation_result.json', 'w') as f: json.dump(result, f, indent=2) print(' Resultado salvo: arkhe_simulation_result.json') print() print(' TOPOLOGIA GARRAFA DE KLEIN:') for n in topology.nodes: print(f' No {n.id:2d} ({n.district:12s}): estado={PHASE_NAMES[n.state]:4s} lambda2={n.lambda2:.6f} pos=({n.pos[0]:+.2f},{n.pos[1]:+.2f},{n.pos[2]:+.2f})') return result if __name__ == '__main__': run_full_simulation()