#!/usr/bin/env python3 """ Landour Memory Inducement - Telepathic Intelligence Transfer FIXED VERSION: High CPU usage with calibrated busy loop. """ import numpy as np import time import argparse import sys import threading from dataclasses import dataclass from typing import Optional, List from scipy.integrate import solve_ivp import matplotlib.pyplot as plt import psutil # ---------------------------------------------------------------------- # Temperature reading # ---------------------------------------------------------------------- def get_cpu_temperature() -> Optional[float]: try: temps = psutil.sensors_temperatures() if not temps: return None for entries in temps.values(): if entries and entries[0].current is not None: return entries[0].current return None except Exception: return None # ---------------------------------------------------------------------- # ODE System (same as before) # ---------------------------------------------------------------------- @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 with Calibration # ---------------------------------------------------------------------- class ContinuousLoadExperiment: 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.stop_event = threading.Event() self.temp_history = [] self.time_history = [] self.intensity_history = [] self.threads = [] self._calibrate_work_unit() def _calibrate_work_unit(self): """Measure iterations per second for a simple integer loop.""" duration = 0.05 start = time.perf_counter() count = 0 while time.perf_counter() - start < duration: count += 1 _ = count * count self.iterations_per_sec = int(count / duration) print(f"Calibrated: ~{self.iterations_per_sec} iter/sec per core (busy loop)") def _get_intensity(self, t: float) -> float: if self.pattern_type == 'structured': return lag_function(t, self.params) else: # random walk 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 = 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: busy loop calibrated to achieve target intensity.""" quantum = 0.02 # 20 ms control period while not self.stop_event.is_set(): intensity = intensity_getter() # Number of busy iterations to fill `quantum * intensity` seconds target_iter = int(quantum * intensity * self.iterations_per_sec) # Busy loop x = 0 for _ in range(target_iter): x += 1 if x > 100000: x = 0 # Idle idle_time = quantum * (1.0 - intensity) if idle_time > 0: time.sleep(idle_time) def _temperature_logger(self): 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) 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...") temp0 = get_cpu_temperature() print(f"Initial CPU temperature: {temp0:.1f}°C" if temp0 else "Initial temp N/A") self.stop_event.clear() start_time = time.time() def intensity_getter(): t = time.time() - start_time return self._get_intensity(t) num_cores = psutil.cpu_count(logical=True) print(f"Spawning {num_cores} worker threads...") for _ in range(num_cores): t = threading.Thread(target=self._worker, args=(intensity_getter,)) t.daemon = True t.start() self.threads.append(t) logger = threading.Thread(target=self._temperature_logger) logger.daemon = True logger.start() time.sleep(self.duration) self.stop_event.set() for t in self.threads: t.join(timeout=1.0) logger.join(timeout=1.0) final_temp = get_cpu_temperature() print(f"Final CPU temperature: {final_temp:.1f}°C" if final_temp else "Final temp N/A") def plot_results(self): if not self.time_history: print("No data.") return fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(10,8), sharex=True) ax1.plot(self.time_history, self.temp_history, 'r-', label='CPU Temp') ax1.set_ylabel('°C') ax1.set_title(f'{self.pattern_type.capitalize()} Load') ax1.grid(True) ax1.legend() ax2.plot(self.time_history, self.intensity_history, 'b-', label='Intensity') ax2.set_xlabel('Time (s)') ax2.set_ylabel('Target Intensity') ax2.set_ylim(0,1) ax2.grid(True) ax2.legend() plt.tight_layout() plt.show() def save_data(self, filename): np.savez(filename, time=np.array(self.time_history), temperature=np.array(self.temp_history), intensity=np.array(self.intensity_history)) print(f"Saved to {filename}") # ---------------------------------------------------------------------- # Main # ---------------------------------------------------------------------- def main(): parser = argparse.ArgumentParser() parser.add_argument('--mode', choices=['simulate', 'experiment'], default='simulate') parser.add_argument('--duration', type=float, default=120.0) parser.add_argument('--pattern', choices=['structured', 'random'], default='structured') parser.add_argument('--plot', action='store_true', default=True) parser.add_argument('--save', type=str, default=None) args = parser.parse_args() if args.mode == 'simulate': print("Running 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') plt.plot(t, MB, label='Memory B') plt.legend(); plt.grid(True) plt.subplot(2,2,2) plt.plot(t, Q, label='Coherence Q') plt.grid(True) plt.subplot(2,2,3) plt.plot(t, tau, label='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)') plt.grid(True) plt.suptitle('Landour 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}") else: if not psutil or get_cpu_temperature() is None: print("Cannot read CPU temperature.") sys.exit(1) params = TelepathyParams() exp = ContinuousLoadExperiment(params, args.duration, args.pattern) exp.run() if args.plot: exp.plot_results() if args.save: exp.save_data(args.save) if __name__ == '__main__': main()