#!/usr/bin/env python3 """ Landour Memory Inducement - Telepathic Intelligence Transfer Q-CCT Water Intelligence / Skiss-Mathematics CONTINUOUS LOAD VERSION: - Spawns one worker thread per CPU core. - Each worker runs a tight loop, but inserts small sleeps proportional to (1 - intensity). - Intensity is either L(t) (structured) or random uniform (control). - This creates sustained, measurable CPU load and temperature rise. """ import numpy as np import time import argparse import sys import threading from dataclasses import dataclass from typing import Optional, List # For ODE solving (simulation mode) from scipy.integrate import solve_ivp # For plotting import matplotlib.pyplot as plt # For system monitoring try: import psutil PSUTIL_AVAILABLE = True except ImportError: PSUTIL_AVAILABLE = False print("Warning: psutil not installed. Install with: pip install psutil") # ---------------------------------------------------------------------- # Temperature reading # ---------------------------------------------------------------------- def get_cpu_temperature() -> Optional[float]: """Return CPU temperature in Celsius or None.""" if not PSUTIL_AVAILABLE: return None try: temps = psutil.sensors_temperatures() if not temps: return None for name, entries in temps.items(): if entries and entries[0].current is not None: return entries[0].current return None except Exception: return None # ---------------------------------------------------------------------- # ODE System (Landour Telepathy) - for simulation mode # ---------------------------------------------------------------------- @dataclass class TelepathyParams: kappa: float = 1.2 rho: float = 0.8 deltaQ: float = 0.25 psi: float = 0.5 phi: float = 0.4 omega_tau: float = 0.3 eta_pulse: float = 0.7 L0: float = 0.4 AL: float = 0.3 fL: float = 0.6 tau_gate: float = 0.8 def lag_function(t: float, params: TelepathyParams) -> float: L = params.L0 + params.AL * np.sin(2 * np.pi * params.fL * t) return np.clip(L, 0.0, 1.0) def telepathy_ode(t, y, params, landour_pulse=0.0): MA, MB, Q, tau = y MA = np.clip(MA, 0.0, 1.0) MB = np.clip(MB, 0.0, 1.0) Q = np.clip(Q, 0.0, 1.0) tau = np.clip(tau, 0.0, 1.0) L = lag_function(t, params) gate = max(0.0, 1.0 - tau * params.tau_gate) landour_A = params.eta_pulse * landour_pulse * (1 - MA) * gate landour_B = params.eta_pulse * landour_pulse * (1 - MB) * gate dMA = params.kappa * (MB - MA) * Q * L + landour_A dMB = params.kappa * (MA - MB) * Q * L + landour_B avg_mem = (MA + MB) / 2.0 dQ = params.rho * L * avg_mem - params.deltaQ * Q - params.phi * tau * Q dTau = params.psi * (1 - L) * (abs(dMA) + abs(dMB)) - params.omega_tau * tau return np.array([dMA, dMB, dQ, dTau]) def run_simulation(duration=35.0, dt_eval=0.05, pulse_time=1.0, pulse_start=5.0, params=None, initial_state=(0.9, 0.1, 0.3, 0.0)): if params is None: params = TelepathyParams() def pulse_func(t): return 1.0 if (pulse_start <= t < pulse_start + pulse_time) else 0.0 t_span = (0.0, duration) t_eval = np.arange(0.0, duration, dt_eval) def ode_with_pulse(t, y): return telepathy_ode(t, y, params, pulse_func(t)) sol = solve_ivp(ode_with_pulse, t_span, initial_state, t_eval=t_eval, method='RK45') return sol.t, sol.y # ---------------------------------------------------------------------- # Continuous Load Experiment # ---------------------------------------------------------------------- class ContinuousLoadExperiment: """ Runs a real experiment with continuous multi‑core load. Intensity (0..1) is either L(t) (structured) or random uniform (control). Temperature is logged every 0.5 seconds. """ def __init__(self, params: TelepathyParams, duration_sec: float = 120.0, pattern_type: str = 'structured', log_interval: float = 0.5): self.params = params self.duration = duration_sec self.pattern_type = pattern_type self.log_interval = log_interval self.running = False self.stop_event = threading.Event() self.temp_history: List[float] = [] self.time_history: List[float] = [] self.intensity_history: List[float] = [] self.threads = [] def _get_intensity(self, t: float) -> float: """Return target intensity (0..1) at time t.""" if self.pattern_type == 'structured': return lag_function(t, self.params) else: # random # Use a slowly varying random walk to avoid abrupt jumps if not hasattr(self, '_last_intensity'): self._last_intensity = 0.5 self._last_time = t dt = t - self._last_time if dt > 0.1: # Change slowly change = np.random.uniform(-0.1, 0.1) new_int = np.clip(self._last_intensity + change, 0.2, 0.9) self._last_intensity = new_int self._last_time = t return self._last_intensity def _worker(self, intensity_getter): """Worker thread: runs continuous loop with sleep proportional to (1 - intensity).""" # Use a small fixed step to keep CPU load responsive step = 0.02 # 20ms while not self.stop_event.is_set(): intensity = intensity_getter() # Busy period: fraction = intensity busy_start = time.perf_counter() busy_end = busy_start + step * intensity # Spin (busy loop) until busy_end while time.perf_counter() < busy_end: _ = sum(i*i for i in range(2000)) # moderate work # Idle period: fraction = 1 - intensity idle_time = step * (1.0 - intensity) if idle_time > 0: time.sleep(idle_time) def _temperature_logger(self): """Background thread to log temperature at fixed intervals.""" start_time = time.time() while not self.stop_event.is_set(): t = time.time() - start_time if t > self.duration: break temp = get_cpu_temperature() if temp is not None: self.time_history.append(t) self.temp_history.append(temp) # Also record current intensity intensity = self._get_intensity(t) if hasattr(self, '_get_intensity') else 0.5 self.intensity_history.append(intensity) time.sleep(self.log_interval) def run(self): print(f"Starting {self.pattern_type} continuous load experiment for {self.duration}s...") if not PSUTIL_AVAILABLE: print("psutil not installed. Cannot monitor temperature.") return # Check initial temperature temp0 = get_cpu_temperature() print(f"Initial CPU temperature: {temp0:.1f}°C" if temp0 else "Initial temperature N/A") self.running = True self.stop_event.clear() # Create intensity getter closure that tracks time start_time = time.time() def intensity_getter(): t = time.time() - start_time return self._get_intensity(t) # Spawn worker threads (one per CPU core) num_cores = psutil.cpu_count(logical=True) print(f"Spawning {num_cores} worker threads...") self.threads = [] for _ in range(num_cores): t = threading.Thread(target=self._worker, args=(intensity_getter,)) t.daemon = True t.start() self.threads.append(t) # Start temperature logger logger_thread = threading.Thread(target=self._temperature_logger) logger_thread.daemon = True logger_thread.start() # Wait for duration time.sleep(self.duration) # Stop all threads self.stop_event.set() for t in self.threads: t.join(timeout=1.0) logger_thread.join(timeout=1.0) print("Experiment finished.") final_temp = get_cpu_temperature() print(f"Final CPU temperature: {final_temp:.1f}°C" if final_temp else "Final temperature N/A") def plot_results(self): if not self.time_history: print("No data to plot.") return fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10, 8), sharex=True) ax1.plot(self.time_history, self.temp_history, 'r-', linewidth=1.5, label='CPU Temperature') ax1.set_ylabel('Temperature (°C)') ax1.set_title(f'{self.pattern_type.capitalize()} Load Pattern (Continuous)') ax1.grid(True) ax1.legend() ax2.plot(self.time_history, self.intensity_history, 'b-', alpha=0.7, label='Target Intensity') ax2.set_xlabel('Time (s)') ax2.set_ylabel('Intensity (0..1)') ax2.set_ylim(0, 1) ax2.grid(True) ax2.legend() plt.tight_layout() plt.show() def save_data(self, filename: str = "landour_experiment.npz"): """Save time, temperature, intensity to .npz file.""" np.savez(filename, time=np.array(self.time_history), temperature=np.array(self.temp_history), intensity=np.array(self.intensity_history)) print(f"Data saved to {filename}") # ---------------------------------------------------------------------- # Main CLI # ---------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser(description='Landour Telepathy - Continuous Load Experiment') parser.add_argument('--mode', choices=['simulate', 'experiment'], default='simulate', help='Run ODE simulation or real CPU load experiment') parser.add_argument('--duration', type=float, default=120.0, help='Duration in seconds (for experiment)') parser.add_argument('--pattern', choices=['structured', 'random'], default='structured', help='For experiment: type of CPU load pattern') parser.add_argument('--plot', action='store_true', default=True, help='Show plots') parser.add_argument('--save', type=str, default=None, help='Save experiment data to .npz file (e.g., data.npz)') args = parser.parse_args() if args.mode == 'simulate': print("Running Landour Telepathy ODE Simulation...") t, y = run_simulation(duration=35.0) MA, MB, Q, tau = y if args.plot: plt.figure(figsize=(12,8)) plt.subplot(2,2,1) plt.plot(t, MA, label='Memory A (Sender)', color='blue') plt.plot(t, MB, label='Memory B (Receiver)', color='orange') plt.xlabel('Time (s)'); plt.ylabel('Memory'); plt.legend(); plt.grid(True) plt.subplot(2,2,2) plt.plot(t, Q, label='Water Coherence Q', color='green') plt.xlabel('Time (s)'); plt.ylabel('Coherence'); plt.grid(True) plt.subplot(2,2,3) plt.plot(t, tau, label='Toxicity τ', color='red') plt.xlabel('Time (s)'); plt.ylabel('Toxicity'); plt.grid(True) plt.subplot(2,2,4) L_vals = [lag_function(ti, TelepathyParams()) for ti in t] plt.plot(t, L_vals, label='Lag L(t)', linestyle='--', color='purple') plt.xlabel('Time (s)'); plt.ylabel('Lag Feature'); plt.grid(True) plt.suptitle('Landour Memory Inducement - Telepathic Transfer Simulation') plt.tight_layout() plt.show() else: print(f"Final: MA={MA[-1]:.3f}, MB={MB[-1]:.3f}, Q={Q[-1]:.3f}, tau={tau[-1]:.3f}") elif args.mode == 'experiment': if not PSUTIL_AVAILABLE: print("psutil not installed. Please install: pip install psutil") sys.exit(1) temp = get_cpu_temperature() if temp is None: print("Could not read CPU temperature. Ensure sensors work.") print("On Linux: install lm-sensors and run 'sensors'") sys.exit(1) params = TelepathyParams() exp = ContinuousLoadExperiment(params, duration_sec=args.duration, pattern_type=args.pattern) exp.run() if args.plot: exp.plot_results() if args.save: exp.save_data(args.save) # Optional: run companion ODE simulation print("\nRunning companion ODE simulation for same parameters...") t, y = run_simulation(duration=min(args.duration, 60.0)) MA, MB, Q, tau = y print(f"Simulation final: MA={MA[-1]:.3f}, MB={MB[-1]:.3f}, Q={Q[-1]:.3f}, tau={tau[-1]:.3f}") if __name__ == '__main__': main()