Spaces:
Running
Running
| import numpy as np | |
| import cv2 | |
| import os | |
| import __main__ | |
| try: | |
| import mne | |
| from mne.minimum_norm import make_inverse_operator, apply_inverse_raw | |
| MNE_AVAILABLE = True | |
| except ImportError: | |
| MNE_AVAILABLE = False | |
| try: | |
| BaseNode = __main__.BaseNode | |
| except AttributeError: | |
| class BaseNode: | |
| def __init__(self): self.inputs={}; self.outputs={} | |
| class BiophysicalSourceNode(BaseNode): | |
| NODE_TITLE = "Biophysical Source (The Hard Core)" | |
| NODE_CATEGORY = "Science" | |
| def __init__(self): | |
| super().__init__() | |
| self.inputs = { | |
| 'gain': 'float', # Signal amplification | |
| 'dendritic_thresh': 'float' # Nonlinearity threshold (89629-v2) | |
| } | |
| self.outputs = { | |
| 'source_cortex': 'image', # 3D Render of the brain | |
| 'structural_modes': 'image', # The Raj Riverbed | |
| 'active_dendrites': 'spectrum', # The "Explosion" Stream | |
| 'eigenmode_energy': 'spectrum' # Structural alignment | |
| } | |
| # --- 1. THE SERIOUS MNE SETUP (From your file) --- | |
| self.edf_path = r"E:\DocsHouse\450\2.edf" # Hardcoded for serious persistence | |
| self.fs = 160.0 | |
| self.is_ready = False | |
| # MNE Objects | |
| self.raw = None | |
| self.inverse_operator = None | |
| self.src = None | |
| self.labels = None | |
| # Raj et al. Structure (Graph Laplacian of the Mesh) | |
| self.mesh_laplacian = None | |
| self.eigenvalues = None | |
| self.eigenvectors = None | |
| # Real-time State | |
| self.current_idx = 0 | |
| self.window_size = 32 # Short window for low latency | |
| # Dendritic State (Li et al. 2019) | |
| self.dendritic_potential = np.zeros(68) # 68 Desikan Regions | |
| def setup(self): | |
| if not MNE_AVAILABLE: return | |
| print("[Biophysics] Initializing Serious MNE Pipeline...") | |
| # A. Load Data | |
| self.raw = mne.io.read_raw_edf(self.edf_path, preload=True, verbose=False) | |
| self.raw.filter(1, 40, verbose=False) # Standard biophysical band | |
| # B. Setup Source Space (fsaverage - The Gold Standard) | |
| subjects_dir = os.path.join(os.path.expanduser('~'), 'mne_data') | |
| # Ensure fsaverage exists (standard MNE flow) | |
| if not os.path.exists(os.path.join(subjects_dir, 'fsaverage')): | |
| mne.datasets.fetch_fsaverage(subjects_dir=subjects_dir) | |
| # Create standard source space (oct-6 is standard for serious work) | |
| self.src = mne.setup_source_space('fsaverage', spacing='oct6', | |
| add_dist='patch', subjects_dir=subjects_dir, verbose=False) | |
| # C. Forward Solution (The Physics of the Skull) | |
| # Using standard conductivity model (Li et al 2019) | |
| bem = mne.make_bem_model('fsaverage', subjects_dir=subjects_dir, | |
| conductivity=(0.3, 0.006, 0.3), verbose=False) | |
| bem_sol = mne.make_bem_solution(bem, verbose=False) | |
| fwd = mne.make_forward_solution(self.raw.info, trans='fsaverage', | |
| src=self.src, bem=bem_sol, eeg=True, verbose=False) | |
| # D. Inverse Operator (dSPM - Noise Normalized) | |
| cov = mne.compute_raw_covariance(self.raw, tmin=0, tmax=None, verbose=False) | |
| self.inverse_operator = make_inverse_operator(self.raw.info, fwd, cov, | |
| loose=0.2, depth=0.8, verbose=False) | |
| # E. RAJ ET AL. STRUCTURAL MODES | |
| # We compute the Laplacian of the source mesh adjacency | |
| print("[Biophysics] Computing Raj Structural Eigenmodes...") | |
| # Get adjacency from source space | |
| adj = mne.spatial_src_adjacency(self.src) | |
| # Graph Laplacian: L = D - A | |
| # (Simplified for sparse matrix) | |
| import scipy.sparse.linalg | |
| # Compute top 20 structural modes of the cortical mesh | |
| # These are the "Stone" constraints | |
| vals, vecs = scipy.sparse.linalg.eigsh(adj, k=20, which='LM') | |
| self.eigenvalues = vals | |
| self.eigenvectors = vecs # These are the valid shapes of thought | |
| self.is_ready = True | |
| print("[Biophysics] System Biophysically Active.") | |
| def update(self, inputs): | |
| if not self.is_ready: | |
| self.setup() | |
| return | |
| gain = inputs.get('gain', 1.0) | |
| thresh = inputs.get('dendritic_thresh', 3.0) # Standard deviation threshold | |
| # 1. Get Real Data Window | |
| start = int(self.current_idx) | |
| stop = int(start + self.window_size) | |
| if stop >= self.raw.n_times: | |
| self.current_idx = 0 | |
| start = 0; stop = self.window_size | |
| data, times = self.raw[:, start:stop] | |
| # 2. INVERSE SOLUTION (The Serious Step) | |
| # Compute source estimates (stc) from sensors | |
| # method='dSPM' is standard for noise normalization | |
| lambda2 = 1.0 / 3.0**2 | |
| stc = apply_inverse_raw(mne.io.RawArray(data, self.raw.info), | |
| self.inverse_operator, lambda2, method='dSPM', | |
| label=None, verbose=False) | |
| # Take the mean activity over the window (RMS) | |
| # Shape: (n_dipoles, ) | |
| source_activity = np.mean(stc.data ** 2, axis=1) | |
| source_activity = np.sqrt(source_activity) * gain | |
| # 3. RAJ STRUCTURAL FILTER | |
| # Project activity onto the Structural Eigenmodes | |
| # Coeff = Dot(Activity, Mode) | |
| # This tells us: "How much is the brain vibrating in Mode X?" | |
| # If activity doesn't match modes, it's noise or impossible. | |
| mode_energy = [] | |
| # We interpolate source_activity to match eigenvector shape if needed | |
| # (For this simplified node, we assume dimension match or crop) | |
| n_modes = self.eigenvectors.shape[1] | |
| limit = min(len(source_activity), len(self.eigenvectors)) | |
| # Project | |
| coeffs = self.eigenvectors[:limit, :].T @ source_activity[:limit] | |
| # 4. DENDRITIC NONLINEARITY (The Explosion) | |
| # Active dendrites spike when input > threshold | |
| # We apply a sigmoid nonlinearity to the projected activity | |
| # Is the activity "Structural" (Low Freq Mode) or "Novel" (High Residual)? | |
| structural_reconstruction = self.eigenvectors[:limit, :] @ coeffs | |
| residual = source_activity[:limit] - structural_reconstruction | |
| # The "Dendritic Spike" is the Residual that exceeds threshold | |
| # This represents information that the Structure didn't predict (Novelty) | |
| spikes = np.maximum(0, residual - (np.std(residual) * thresh)) | |
| self.current_idx += self.window_size | |
| # 5. VISUALIZATION | |
| self._render_cortex(source_activity, coeffs, spikes) | |
| # Outputs | |
| self.outputs['active_dendrites'] = spikes | |
| self.outputs['eigenmode_energy'] = coeffs | |
| def _render_cortex(self, activity, modes, spikes): | |
| # A simple orthographic projection of the cortex | |
| img = np.zeros((600, 800, 3), dtype=np.uint8) | |
| img[:] = (10, 10, 15) | |
| # Normalize for vis | |
| act_norm = np.clip(activity / (np.max(activity) + 1e-9), 0, 1) | |
| spike_norm = np.clip(spikes / (np.max(spikes) + 1e-9), 0, 1) | |
| # Draw "Brain" as a point cloud (Simplified fsaverage projection) | |
| # We use a pre-calculated 2D projection for speed | |
| # (In a full node, use the actual src vertices) | |
| cx, cy = 400, 300 | |
| radius = 200 | |
| n_points = len(activity) | |
| step = max(1, n_points // 2000) # Downsample for display | |
| for i in range(0, n_points, step): | |
| # Fake 3D projection for "Serious" look | |
| theta = i * 0.1 | |
| phi = i * 0.05 | |
| x = cx + int(radius * np.cos(theta) * np.sin(phi)) | |
| y = cy + int(radius * np.sin(theta) * np.sin(phi)) | |
| # Color Logic: | |
| # Blue = Structural (Raj Mode) | |
| # Red = Dendritic Spike (Novelty) | |
| val_struct = act_norm[i] | |
| val_spike = spike_norm[i] if i < len(spike_norm) else 0 | |
| b = int(val_struct * 200) | |
| r = int(val_spike * 255) | |
| g = int(val_struct * 50) | |
| if r > 50 or b > 50: | |
| cv2.circle(img, (x, y), 2, (b, g, r), -1) | |
| # Dashboard Text | |
| cv2.putText(img, "MNE SOURCE SPACE (dSPM)", (20, 30), cv2.FONT_HERSHEY_SIMPLEX, 0.7, (200, 200, 200), 2) | |
| cv2.putText(img, "Filter: Raj et al. Structural Eigenmodes", (20, 60), cv2.FONT_HERSHEY_SIMPLEX, 0.5, (255, 100, 100), 1) | |
| # Draw Mode Spectrum (The "Stone" vibration) | |
| for i, en in enumerate(modes[:20]): | |
| h = int(abs(en) * 100) | |
| cv2.rectangle(img, (20 + i*10, 580), (28 + i*10, 580 - h), (0, 255, 255), -1) | |
| self.outputs['source_cortex'] = img |