resonance-engine/scripts/direct_zmq_sweep.py

#!/usr/bin/env python3
"""
EM Frequency Sweep - Direct ZMQ Version (v2)
Persistent PUB socket with ACK verification.
Bypasses observer /ask endpoint, sends commands directly to daemon.
"""

import zmq
import json
import time
import csv
import requests
from datetime import datetime
import sys
import atexit

OBSERVER_URL = "http://127.0.0.1:28820"
COMMAND_PORT = 5557
ACK_PORT = 5559

# Sweep ranges — omega capped at 1.99 (daemon rejects > 1.99)
OMEGA_VALUES = [round(0.5 + 0.1*i, 1) for i in range(15)]   # 0.5 to 1.9
KHRA_VALUES = [round(0.01 + 0.01*i, 2) for i in range(5)]    # 0.01 to 0.05
GIXX_VALUES = [round(0.004 + 0.002*i, 3) for i in range(5)]  # 0.004 to 0.012

STABILIZE_TIME = 3  # seconds
ACK_TIMEOUT_MS = 5000  # 5s per ACK

# Safety limits
MAX_TEMP = 64
MAX_POWER = 320

# --- Persistent ZMQ sockets (module-level, created once) ---
_ctx = zmq.Context()

_cmd_pub = _ctx.socket(zmq.PUB)
_cmd_pub.setsockopt(zmq.LINGER, 1000)
_cmd_pub.connect(f"tcp://127.0.0.1:{COMMAND_PORT}")

_ack_sub = _ctx.socket(zmq.SUB)
_ack_sub.connect(f"tcp://127.0.0.1:{ACK_PORT}")
_ack_sub.setsockopt_string(zmq.SUBSCRIBE, "")
_ack_sub.setsockopt(zmq.RCVTIMEO, ACK_TIMEOUT_MS)


def _cleanup():
    _cmd_pub.close()
    _ack_sub.close()
    _ctx.term()

atexit.register(_cleanup)


def drain_acks():
    """Drain any stale ACKs from the SUB socket."""
    count = 0
    while True:
        try:
            _ack_sub.recv_string(zmq.NOBLOCK)
            count += 1
        except zmq.Again:
            break
    return count


def send_zmq_command(cmd, value=None):
    """Send command via persistent PUB socket, verify ACK."""
    if value is not None:
        msg = json.dumps({"cmd": cmd, "value": float(value)})
    else:
        msg = json.dumps({"cmd": cmd})

    _cmd_pub.send_string(msg)

    # Wait for ACK
    try:
        ack_raw = _ack_sub.recv_string()
        ack = json.loads(ack_raw)
        if ack.get("status") == "ok":
            return True
        else:
            print(f"  [ACK] {cmd}: {ack.get('status', 'unknown')}")
            return False
    except zmq.Again:
        # Retry once
        print(f"  [RETRY] No ACK for {cmd}, resending...")
        _cmd_pub.send_string(msg)
        try:
            ack_raw = _ack_sub.recv_string()
            ack = json.loads(ack_raw)
            if ack.get("status") == "ok":
                return True
            print(f"  [ACK] {cmd} retry: {ack.get('status', 'unknown')}")
            return False
        except zmq.Again:
            print(f"  [FAIL] No ACK for {cmd} after retry")
            return False

def get_telemetry():
    """Get current telemetry from observer."""
    try:
        response = requests.get(f"{OBSERVER_URL}/telemetry", timeout=5)
        return response.json()
    except Exception as e:
        print(f"  Telemetry Error: {e}")
        return None

def main():
    timestamp_tag = datetime.now().strftime("%Y%m%d_%H%M%S")
    output_file = f"/mnt/d/Resonance_Engine/sweep_results/em_direct_sweep_{timestamp_tag}.csv"

    total_points = len(OMEGA_VALUES) * len(KHRA_VALUES) * len(GIXX_VALUES)

    print("=" * 60)
    print("EM Frequency Sweep - Direct ZMQ v2 (persistent socket + ACK)")
    print("=" * 60)
    print(f"Omega range: {OMEGA_VALUES[0]} - {OMEGA_VALUES[-1]} ({len(OMEGA_VALUES)} steps)")
    print(f"Khra range:  {KHRA_VALUES[0]} - {KHRA_VALUES[-1]} ({len(KHRA_VALUES)} steps)")
    print(f"Gixx range:  {GIXX_VALUES[0]} - {GIXX_VALUES[-1]} ({len(GIXX_VALUES)} steps)")
    print(f"Total points: {total_points}")
    print(f"Output: {output_file}")
    print()

    # Wait for ZMQ subscription propagation (matching Observer pattern)
    print("Waiting 2s for ZMQ subscription propagation...")
    time.sleep(2.0)
    drained = drain_acks()
    if drained:
        print(f"  Drained {drained} stale ACK(s)")

    # Capture initial state for restore
    initial = get_telemetry()
    if not initial:
        print("ERROR: Cannot read telemetry — aborting")
        sys.exit(1)
    restore_omega = initial.get('omega', 1.5)
    restore_khra = initial.get('khra_amp', 0.02)
    restore_gixx = initial.get('gixx_amp', 0.008)
    print(f"Initial state: Ω={restore_omega} K={restore_khra} G={restore_gixx}")
    print()

    # Create CSV
    with open(output_file, 'w', newline='') as f:
        writer = csv.writer(f)
        writer.writerow([
            'timestamp', 'omega', 'khra_amp', 'gixx_amp',
            'coherence', 'asymmetry', 'vorticity_mean',
            'gpu_temp_c', 'gpu_power_w', 'cycle'
        ])

        point_count = 0
        fail_count = 0

        for omega in OMEGA_VALUES:
            for khra in KHRA_VALUES:
                for gixx in GIXX_VALUES:
                    point_count += 1
                    print(f"\n[{point_count}/{total_points}] Ω={omega} K={khra} G={gixx}")

                    # Safety check
                    telem_pre = get_telemetry()
                    if telem_pre:
                        temp = telem_pre.get('gpu_temp_c', 0)
                        power = telem_pre.get('gpu_power_w', 0)
                        if temp > MAX_TEMP:
                            print(f"  THERMAL PAUSE: {temp}C > {MAX_TEMP}C, cooling...")
                            while True:
                                time.sleep(5)
                                t = get_telemetry()
                                if t and t.get('gpu_temp_c', 99) < MAX_TEMP - 5:
                                    break
                        if power > MAX_POWER:
                            print(f"  POWER WARN: {power}W > {MAX_POWER}W")

                    # Send commands via persistent ZMQ
                    ok1 = send_zmq_command("set_omega", omega)
                    ok2 = send_zmq_command("set_khra_amp", khra)
                    ok3 = send_zmq_command("set_gixx_amp", gixx)

                    if not (ok1 and ok2 and ok3):
                        fail_count += 1
                        print(f"  Command delivery failed ({fail_count} total failures)")
                        if fail_count > 10:
                            print("ERROR: Too many failures, aborting sweep")
                            break

                    # Wait for stabilization
                    time.sleep(STABILIZE_TIME)

                    # Get telemetry
                    telem = get_telemetry()
                    if telem:
                        writer.writerow([
                            datetime.now().isoformat(), omega, khra, gixx,
                            telem.get('coherence', 0),
                            telem.get('asymmetry', 0),
                            telem.get('vorticity_mean', 0),
                            telem.get('gpu_temp_c', 0),
                            telem.get('gpu_power_w', 0),
                            telem.get('cycle', 0)
                        ])
                        f.flush()
                        print(f"  OK Coh={telem.get('coherence', 0):.4f} "
                              f"T={telem.get('gpu_temp_c', 0)}C "
                              f"P={telem.get('gpu_power_w', 0)}W")
                    else:
                        print(f"  SKIP: no telemetry")

                    if point_count % 25 == 0:
                        print(f"\n*** Progress: {point_count}/{total_points} ***\n")
                else:
                    continue
                break
            else:
                continue
            break

    print("\n" + "=" * 60)
    print(f"Sweep complete! {point_count} points measured ({fail_count} failures)")
    print(f"Output: {output_file}")
    print("=" * 60)

    # Restore initial parameters
    print(f"\nRestoring: Ω={restore_omega} K={restore_khra} G={restore_gixx}")
    send_zmq_command("set_omega", restore_omega)
    send_zmq_command("set_khra_amp", restore_khra)
    send_zmq_command("set_gixx_amp", restore_gixx)
    print("Restored.")

if __name__ == "__main__":
    main()