Spaces:
Running
Running
File size: 7,023 Bytes
3bb804c |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 |
"""
Rhythm Gated Perturbation Node
--------------------------------
Models a "temporal gating" mechanism. It takes a stable latent
vector ("Soma" thought) and a "Rhythm" signal ("Dendritic" clock).
If the rhythm becomes incoherent (unstable), it "breaks the gate"
and "leaks" a high-frequency "Phase Field" (a perturbation)
into the latent vector, simulating a "fractal leak" hallucination
.
"""
import numpy as np
from PyQt6 import QtGui
import cv2
from collections import deque
# --- Magic import block ---
import __main__
BaseNode = __main__.BaseNode
QtGui = __main__.QtGui
PA_INSTANCE = getattr(__main__, "PA_INSTANCE", None)
# --------------------------
class RhythmGatedPerturbationNode(BaseNode):
NODE_CATEGORY = "Cognitive"
NODE_COLOR = QtGui.QColor(140, 70, 180) # Deep Purple
def __init__(self, history_length=50, coherence_threshold=0.8, perturb_strength=1.0):
super().__init__()
self.node_title = "Rhythm Gated Perturbation"
self.inputs = {
'latent_in': 'spectrum', # The stable "Soma" state
'rhythm_in': 'signal', # The "Dendritic" timing signal
'fractal_field_in': 'spectrum' # Optional: The "raw" field to leak
}
self.outputs = {
'latent_out': 'spectrum', # The final (potentially corrupted) state
'leakage_amount': 'signal' # 0=Stable, 1=Full Leak
}
# Configurable
self.history_length = int(history_length)
self.coherence_threshold = float(coherence_threshold)
self.perturb_strength = float(perturb_strength)
# Internal state
self.rhythm_history = deque(maxlen=self.history_length)
self.current_coherence = 1.0 # Start in a stable state
self.leakage_amount_out = 0.0
self.latent_out = None
# Ensure deque is initialized
for _ in range(self.history_length):
self.rhythm_history.append(0.0)
def step(self):
# 1. Get Inputs
latent_in = self.get_blended_input('latent_in', 'first')
rhythm_in = self.get_blended_input('rhythm_in', 'sum')
fractal_field_in = self.get_blended_input('fractal_field_in', 'first')
# Update rhythm history, even if it's None (to detect drops)
self.rhythm_history.append(rhythm_in if rhythm_in is not None else 0.0)
# 2. Calculate Rhythm Coherence
# Coherence = inverse of standard deviation (variance)
rhythm_std = np.std(self.rhythm_history)
# This maps std=0 to coherence=1. Higher std -> lower coherence.
self.current_coherence = 1.0 / (1.0 + rhythm_std * 10.0)
self.current_coherence = np.clip(self.current_coherence, 0.0, 1.0)
# 3. Calculate "Fractal Leakage"
if self.current_coherence < self.coherence_threshold:
# The gate is "broken"
self.leakage_amount_out = (self.coherence_threshold - self.current_coherence) / self.coherence_threshold
else:
# The gate is "stable"
self.leakage_amount_out = 0.0
self.leakage_amount_out = np.clip(self.leakage_amount_out, 0.0, 1.0)
# 4. Apply the Leak
if latent_in is None:
self.latent_out = None
return
if self.leakage_amount_out > 0.01:
# --- THE FRACTAL LEAK IS HAPPENING ---
# Get the perturbation vector
if fractal_field_in is not None and len(fractal_field_in) == len(latent_in):
perturb_vector = fractal_field_in
else:
# If no field is provided, create high-frequency noise
perturb_vector = np.random.randn(len(latent_in)).astype(np.float32)
# Scale perturbation
perturb_vector = perturb_vector * self.perturb_strength
# Blend: (Stable Thought * Coherence) + (Raw Field * Leakage)
self.latent_out = (latent_in * (1.0 - self.leakage_amount_out)) + \
(perturb_vector * self.leakage_amount_out)
else:
# --- STABLE OPERATION ---
self.latent_out = latent_in
def get_output(self, port_name):
if port_name == 'latent_out':
return self.latent_out
elif port_name == 'leakage_amount':
return self.leakage_amount_out
return None
def get_display_image(self):
w, h = 256, 128
img = np.zeros((h, w, 3), dtype=np.uint8)
# Draw Coherence / Leakage
coherence_w = int(self.current_coherence * w)
leakage_w = int(self.leakage_amount_out * w)
# Coherence Bar (Green)
cv2.rectangle(img, (0, 0), (coherence_w, h // 3), (0, 150, 0), -1)
# Leakage Bar (Red)
cv2.rectangle(img, (0, h // 3), (leakage_w, 2 * h // 3), (150, 0, 0), -1)
cv2.putText(img, f"Coherence: {self.current_coherence:.2f}", (10, 20),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (255, 255, 255), 1)
cv2.putText(img, f"Leakage: {self.leakage_amount_out:.2f}", (10, 60),
cv2.FONT_HERSHEY_SIMPLEX, 0.4, (255, 255, 255), 1)
# Draw a line for the coherence threshold
thresh_x = int(self.coherence_threshold * w)
cv2.line(img, (thresh_x, 0), (thresh_x, h // 3), (0, 255, 0), 2)
# Display the output latent vector
if self.latent_out is not None:
latent_dim = len(self.latent_out)
bar_width = max(1, w // latent_dim)
val_max = np.abs(self.latent_out).max()
if val_max < 1e-6: val_max = 1.0
for i, val in enumerate(self.latent_out):
x = i * bar_width
norm_val = val / val_max
bar_h = int(np.clip(abs(norm_val) * (h/3 - 5), 0, h/3 - 5))
y_base = h - (h // 6) # Center of bottom 3rd
color = (200, 200, 200) # Default
if self.leakage_amount_out > 0.01:
color = (255, 255, 0) # Tint yellow during leak
if val >= 0:
cv2.rectangle(img, (x, y_base-bar_h), (x+bar_width-1, y_base), color, -1)
else:
cv2.rectangle(img, (x, y_base), (x+bar_width-1, y_base+bar_h), color, -1)
img = np.ascontiguousarray(img)
return QtGui.QImage(img.data, w, h, 3*w, QtGui.QImage.Format.Format_BGR888)
def get_config_options(self):
return [
("History Length", "history_length", self.history_length, None),
("Coherence Threshold", "coherence_threshold", self.coherence_threshold, None),
("Perturbation Strength", "perturb_strength", self.perturb_strength, None)
] |