#!/usr/bin/env python3 """CorvOS - qhttp:// v2: Protocolo de Re-entrada com Logica Ternaria de Varela""" import numpy as np import hashlib, uuid from dataclasses import dataclass from typing import Optional, List # ============== VARELA TERNARY LOGIC ============== class VarelaValue: VOID = 0 MARKED = 1 AUTONOMOUS = 'a' class TernaryState: def __init__(self): self.void_p = 0.0 self.marked_p = 0.0 self.autonomous_p = 0.0 self.phase = 0.0 self.lambda2 = 0.0 def value(self): probs = [self.void_p, self.marked_p, self.autonomous_p] idx = probs.index(max(probs)) return [0, 1, 'a'][idx] def cross(self): v = self.value() r = TernaryState() r.phase = self.phase r.lambda2 = self.lambda2 if v == 0: r.marked_p = 1.0 elif v == 1: r.void_p = 1.0 else: r.autonomous_p = 1.0 return r # ============== qhttp:// v2 PACKET ============== REENTRY_TYPES = ['self-cross','void-marked','marked-void','void-void','marked-marked','entangled'] class qHTTPPacket: def __init__(self, reentry_type, origin, destination, payload, lambda2, phase, future_state=None): if reentry_type not in REENTRY_TYPES: raise ValueError('invalid reentry_type') self.id = uuid.uuid4().hex[:16] self.reentry_type = reentry_type self.origin = origin self.destination = destination self.payload = payload self.lambda2 = lambda2 self.phase = phase self.timestamp = __import__('datetime').datetime.now() self.future_state = future_state @property def url(self): return f'qhttp://{self.reentry_type}/{self.origin}/{self.destination}#phase={self.lambda2:.6f}' def reenter(self): ts = TernaryState(); ts.autonomous_p=1.0; ts.phase=self.phase; ts.lambda2=self.lambda2 crossed = ts.cross() rt = 'self-cross' if crossed.value()=='a' else 'void-marked' return qHTTPPacket(rt, self.destination, self.origin, self.payload, self.lambda2, self.phase * np.exp(1j*np.pi), self) # ============== qhttp:// HANDLER ============== class qHTTPHandler: def __init__(self, node_id, lambda2_thresh=0.847): self.node_id = node_id self.lambda2_thresh = lambda2_thresh self.state = TernaryState() self.pending = {} def receive(self, pkt): if pkt.lambda2 < self.lambda2_thresh: self.state.void_p = 1.0; self.state.lambda2 = pkt.lambda2; return self.state elif pkt.lambda2 < 0.999: self.state.autonomous_p = 1.0 self.state.lambda2 = pkt.lambda2 self.state.phase = pkt.phase self.pending[pkt.id] = pkt return self.state else: self.state.marked_p = 1.0 self.state.lambda2 = pkt.lambda2 if pkt.id in self.pending: del self.pending[pkt.id] return self.state def kuramoto_step(self, neighbors, K): if not neighbors: return self.state.lambda2 dphase = sum(np.sin(n.state.phase - self.state.phase) for n in neighbors) / len(neighbors) if K > 0.618 and self.state.value() == 'a': return min(1.0, self.state.lambda2 + 0.001) new_phase = self.state.phase + dphase * K new_phase = np.angle(new_phase) % (2*np.pi) R = np.abs(np.mean([np.exp(1j*n.state.phase) for n in neighbors+[self]])) self.state.lambda2 = float(R) return self.state.lambda2 # ============== AQUATIC NODE ============== class AquaticTzinorNode(qHTTPHandler): def __init__(self, node_id, submerged=False, depth_m=0.0, n_water=1.33): super().__init__(node_id) self.is_submerged = submerged self.depth_m = depth_m self.n_water = n_water self.c = 299792458 def propagation_delay(self, dist_km): v = self.c / self.n_water return (dist_km*1000/v)*1e9 def spatial_reentry(self, pkt, dist_km): delay = self.propagation_delay(dist_km) phase_shift = 2*np.pi*(delay/1e9)*40 sc_phase = np.exp(1j*phase_shift) nr = 'marked-void' if pkt.reentry_type=='void-marked' else 'void-marked' return qHTTPPacket(nr, pkt.destination, pkt.origin, pkt.payload, pkt.lambda2, pkt.phase*sc_phase, pkt) def compensate_depth(self): if not self.is_submerged: return 1.0 P = 1025*9.81*self.depth_m return 1.0/((P+101325)/101325) # ============== SIMULATION ============== def run_simulation(): print('='*70) print(' qhttp:// v2 - SIMULACAO AQUATICA: RIO <-> NITEROI') print(' Baia de Guanabara: ~30km | n_agua=1.33') print('='*70) print() nodes = { 'rio_surface': AquaticTzinorNode('rio_surface', False, 0), 'buoy_1': AquaticTzinorNode('buoy_1', False, 2), 'submerged_1': AquaticTzinorNode('submerged_1', True, 15), 'submerged_2': AquaticTzinorNode('submerged_2', True, 25), 'buoy_3': AquaticTzinorNode('buoy_3', False, 2), 'niteroi_surface': AquaticTzinorNode('niteroi_surface', False, 0), } initial = qHTTPPacket('void-marked','rio_surface','niteroi_surface', b'qhttp://intuition/query#district=leblon', 0.950, np.random.uniform(0.0, 2*np.pi)) print(f' [INIT] {initial.url}') print(f' Lambda2: {initial.lambda2:.6f} Fase: {float(initial.phase):.4f}') print() print(' TOPOLOGIA:') for nid, nd in nodes.items(): tag = ' [SUB]' if nd.is_submerged else ' [SUR]' dp = f' d={nd.depth_m}m' if nd.is_submerged else '' print(f' - {nid}{tag}{dp}') print() distances = [('rio_surface','buoy_1',5),('buoy_1','submerged_1',8), ('submerged_1','submerged_2',10),('submerged_2','buoy_3',5), ('buoy_3','niteroi_surface',2)] print(' PROPAGACAO:') cur = initial path = ['rio_surface'] for src, dst, dist in distances: nd = nodes[dst] delay_ps = nd.propagation_delay(dist) cur = nd.spatial_reentry(cur, dist) path.append(dst) phase_deg = float(np.angle(cur.phase, deg=True)) print(f' -> {src} [{dist}km] -> {dst}: delay={delay_ps:.1f}ps phase={phase_deg:.2f}deg lambda2={cur.lambda2:.6f}') print() # Kuramoto sync all_nodes = list(nodes.values()) print(' KURAMOTO SYNC (K=0.65 > Kc=0.618):') for step in range(5): lambdas = [n.kuramoto_step(all_nodes, 0.65) for n in all_nodes] mean_l = np.mean(lambdas) autonomous = sum(1 for n in all_nodes if n.state.lambda2 >= 0.847) print(f' Step {step+1}: lambda2_medio={mean_l:.6f} nodos_em_a={autonomous}/{len(all_nodes)}') print() print('='*70) print(' RESULTADO: Transmissao completa. Estado autonomo mantido.') print('='*70) if __name__ == '__main__': run_simulation()