Scruff AI
65 lines
2.0 KiB
Python
65 lines
2.0 KiB
Python
#!/usr/bin/env python3
|
|
"""Round-trip test: set_param lam_gixx 16.0, verify ack and telemetry change, restore."""
|
|
import zmq, json, time, subprocess
|
|
|
|
ctx = zmq.Context()
|
|
|
|
# Subscribe to acks on 5559
|
|
ack_sub = ctx.socket(zmq.SUB)
|
|
ack_sub.connect('tcp://127.0.0.1:5559')
|
|
ack_sub.setsockopt_string(zmq.SUBSCRIBE, '')
|
|
|
|
# Send command to 5557
|
|
cmd_pub = ctx.socket(zmq.PUB)
|
|
cmd_pub.connect('tcp://127.0.0.1:5557')
|
|
time.sleep(0.3) # allow connect
|
|
|
|
print('Sending: set_param lam_gixx=16.0')
|
|
cmd_pub.send_string(json.dumps({'cmd': 'set_param', 'param': 'lam_gixx', 'value': 16.0}))
|
|
time.sleep(4)
|
|
|
|
# Read acks
|
|
while True:
|
|
try:
|
|
msg = ack_sub.recv_string(flags=zmq.NOBLOCK)
|
|
print('ACK:', msg)
|
|
except zmq.ZMQError:
|
|
break
|
|
|
|
# Check telemetry
|
|
result = subprocess.run(['tail', '-1', '/mnt/d/resonance-engine-active/telemetry.jsonl'],
|
|
capture_output=True, text=True)
|
|
d = json.loads(result.stdout.strip())
|
|
print(f'lam_gixx={d.get("lam_gixx")} alpha={d.get("alpha")} cycle={d.get("cycle")}')
|
|
|
|
# Set back to 8.0
|
|
print('Restoring: set_param lam_gixx=8.0')
|
|
cmd_pub.send_string(json.dumps({'cmd': 'set_param', 'param': 'lam_gixx', 'value': 8.0}))
|
|
time.sleep(4)
|
|
while True:
|
|
try:
|
|
msg = ack_sub.recv_string(flags=zmq.NOBLOCK)
|
|
print('ACK (restore):', msg)
|
|
except zmq.ZMQError:
|
|
break
|
|
|
|
result = subprocess.run(['tail', '-1', '/mnt/d/resonance-engine-active/telemetry.jsonl'],
|
|
capture_output=True, text=True)
|
|
d = json.loads(result.stdout.strip())
|
|
print(f'After restore: lam_gixx={d.get("lam_gixx")} alpha={d.get("alpha")} cycle={d.get("cycle")}')
|
|
|
|
# Test rejection: out-of-range value
|
|
print('Testing rejection: lam_gixx=200.0 (out of range)')
|
|
cmd_pub.send_string(json.dumps({'cmd': 'set_param', 'param': 'lam_gixx', 'value': 200.0}))
|
|
time.sleep(2)
|
|
while True:
|
|
try:
|
|
msg = ack_sub.recv_string(flags=zmq.NOBLOCK)
|
|
print('ACK (reject):', msg)
|
|
except zmq.ZMQError:
|
|
break
|
|
|
|
cmd_pub.close()
|
|
ack_sub.close()
|
|
ctx.term()
|
|
print('Done.') |