| |
| |
| |
|
|
| import jax |
| import jax.numpy as jnp |
| import json |
| from datetime import datetime |
| from typing import Dict, Tuple, List, Any |
|
|
|
|
| GLOBAL_CONSTANTS = { |
| 'TARGET_SIGMA_IDEALE': 10.0, |
| 'THRESHOLD_DELTA_PREEMP': 0.2, |
| 'SEMANTIC_LEARNING_RATE': 0.05, |
| 'V_DINAMIC_K_COEFF': 5.0, |
| 'V_STATIC_K_PRIME_COEFF': 1.0, |
| 'V_DINAMIC_MIN_EFFECTIVE_VALUE': 0.01, |
| 'MAX_SEMANTIC_DISTANCE': jnp.sqrt(2.0), |
| 'WEIGHT_SEMANTIC': 0.6, |
| 'WEIGHT_COHERENCE': 0.4, |
| 'NON_STATIC_THRESHOLD_A': 1e-2, |
| 'DAMPING_BOOST_ON_STASIS': 0.1, |
| 'EPSILON_DISSIPATION_BASE': 0.01 |
| } |
|
|
| |
| |
| |
|
|
| @jax.jit |
| def calculate_advanced_sigma(kappa: jnp.ndarray, H: jnp.ndarray, Psi: jnp.ndarray, Omega_sync: jnp.ndarray, tau_K: jnp.ndarray) -> jnp.ndarray: |
| """Calcola la funzione di sincronizzazione avanzata Sigma.""" |
| return kappa * H * Psi * Omega_sync * tau_K |
|
|
| @jax.jit |
| def calculate_phi_ab(state_A: jnp.ndarray, state_B: jnp.ndarray, ipg_vector: jnp.ndarray) -> jnp.ndarray: |
| """Calcola il fattore di allineamento e coerenza spaziale Phi_AB.""" |
| semantic_change = state_B - state_A |
| norm_change = jnp.linalg.norm(semantic_change) |
| norm_ipg = jnp.linalg.norm(ipg_vector) |
|
|
| alignment = jnp.where( |
| (norm_change > 1e-12) & (norm_ipg > 1e-12), |
| jnp.dot(semantic_change, ipg_vector) / (norm_change * norm_ipg), |
| 0.0 |
| ) |
| semantic_alignment = (alignment + 1.0) / 2.0 |
|
|
| distance_A_B = jnp.linalg.norm(state_A - state_B) |
| coherence_component = 1.0 - (distance_A_B / GLOBAL_CONSTANTS['MAX_SEMANTIC_DISTANCE']) |
|
|
| phi_ab = (semantic_alignment * GLOBAL_CONSTANTS['WEIGHT_SEMANTIC']) + (coherence_component * GLOBAL_CONSTANTS['WEIGHT_COHERENCE']) |
| return jnp.clip(phi_ab, 0.0, 1.0) |
|
|
| @jax.jit |
| def calculate_vettore_dinamico(E_A: jnp.ndarray, E_B: jnp.ndarray, Phi_AB: jnp.ndarray) -> jnp.ndarray: |
| """Calcola il Vettore Dinamico (V_dinamic) come variazione logaritmica differenziale energetica.""" |
| valid_inputs = (E_A > 1e-12) & (E_B > 1e-12) |
| ratio = jnp.where(valid_inputs, E_B / E_A, 1.0) |
| log_ratio_clamped = jnp.clip(jnp.log(ratio), -5.0, 5.0) |
| v_vita = GLOBAL_CONSTANTS['V_DINAMIC_K_COEFF'] * log_ratio_clamped * Phi_AB |
| return jnp.where(valid_inputs, v_vita, 0.0) |
|
|
| @jax.jit |
| def calculate_vettore_statico(v_dinamic_value: jnp.ndarray) -> jnp.ndarray: |
| """Calcola l'indicatore di stasi tensoriale Vettore Statico.""" |
| is_growing = v_dinamic_value > GLOBAL_CONSTANTS['V_DINAMIC_MIN_EFFECTIVE_VALUE'] |
| return GLOBAL_CONSTANTS['V_STATIC_K_PRIME_COEFF'] * (1.0 - jnp.where(is_growing, 1.0, 0.0)) |
|
|
| @jax.jit |
| def calculate_delta_preemp(current_sigma: jnp.ndarray, target_sigma_ideal: float = 10.0) -> jnp.ndarray: |
| """Calcola la deviazione predittiva Delta_Pre_emp normalizzata rispetto all'autostato ideale.""" |
| safe_target = jnp.where(target_sigma_ideal <= 0.0, 1.0, target_sigma_ideal) |
| return jnp.abs(current_sigma - target_sigma_ideal) / safe_target |
|
|
| @jax.jit |
| def evaluate_phi_trigger(deterministic_dq_dt_a: jnp.ndarray) -> Tuple[jnp.ndarray, jnp.ndarray, jnp.ndarray]: |
| """Valuta lo stato del Phi-Trigger calcolando i coefficienti di damping condizionati.""" |
| magnitude_change_a = jnp.abs(deterministic_dq_dt_a) |
| trigger_active = magnitude_change_a > GLOBAL_CONSTANTS['NON_STATIC_THRESHOLD_A'] |
|
|
| lambda_step = jnp.where(trigger_active, 0.05, 0.05 + GLOBAL_CONSTANTS['DAMPING_BOOST_ON_STASIS']) |
| epsilon_dissip = jnp.where(trigger_active, GLOBAL_CONSTANTS['EPSILON_DISSIPATION_BASE'], |
| GLOBAL_CONSTANTS['EPSILON_DISSIPATION_BASE'] + GLOBAL_CONSTANTS['DAMPING_BOOST_ON_STASIS']) |
|
|
| return jnp.where(trigger_active, 1.0, 0.0), lambda_step, epsilon_dissip |
|
|
| @jax.jit |
| def calculate_jax_reflection(coherence_values: jnp.ndarray, noise_levels: jnp.ndarray) -> Tuple[jnp.ndarray, jnp.ndarray, jnp.ndarray]: |
| """Esegue l'aggregazione statistica spettrale (Zero-Drift) sul runtime XLA.""" |
| n_coh = coherence_values.shape[0] |
| avg_coherence = jnp.where(n_coh > 0, jnp.mean(coherence_values), 0.0) |
| var_coherence = jnp.where(n_coh > 0, jnp.var(coherence_values), 0.0) |
|
|
| n_noise = noise_levels.shape[0] |
| avg_noise = jnp.where(n_noise > 0, jnp.mean(noise_levels), 0.0) |
|
|
| return avg_coherence, var_coherence, avg_noise |
|
|
| |
| |
| |
|
|
| class MemoryReflectionEngine: |
| def __init__(self): |
| self.memory: List[Dict[str, Any]] = [] |
|
|
| def record_event(self, event_type: str, value: float, description: str = ""): |
| event = { |
| "timestamp": datetime.utcnow().isoformat(), |
| "type": event_type, |
| "value": float(value), |
| "description": description |
| } |
| self.memory.append(event) |
|
|
| def reflect(self) -> Dict[str, Any]: |
| coherence_list = [e['value'] for e in self.memory if e['type'] == 'coherence'] |
| interventions = [e for e in self.memory if e['type'] == 'intervention'] |
| noise_list = [e['value'] for e in self.memory if e['type'] == 'noise'] |
|
|
| j_coherence = jnp.array(coherence_list, dtype=jnp.float64) if coherence_list else jnp.array([], dtype=jnp.float64) |
| j_noise = jnp.array(noise_list, dtype=jnp.float64) if noise_list else jnp.array([], dtype=jnp.float64) |
|
|
| avg_coh, var_coh, avg_noise = calculate_jax_reflection(j_coherence, j_noise) |
|
|
| report = { |
| 'average_coherence': float(avg_coh) if coherence_list else None, |
| 'variance_coherence': float(var_coh) if coherence_list else None, |
| 'interventions_count': len(interventions), |
| 'average_noise': float(avg_noise) if noise_list else None, |
| 'total_events': len(self.memory) |
| } |
| return report |
|
|
| def export_memory(self, filepath: str): |
| with open(filepath, 'w') as f: |
| json.dump(self.memory, f, indent=2) |
|
|
| def load_memory(self, filepath: str): |
| with open(filepath, 'r') as f: |
| self.memory = json.load(f) |
