#!/usr/bin/env python3
"""
cog-ws.py -- WebSocket client for CogServer /scm endpoint.
Usage:
python3 cog-ws.py '(+ 1 2)' # One-shot command
python3 cog-ws.py -f run-corpus-test.scm # Load and run a .scm file
python3 cog-ws.py -i # Interactive REPL
Connects to ws://localhost:18080/scm (Guile REPL over WebSocket).
Handles long-running operations without dropping the connection.
"""
import asyncio
import sys
import signal
try:
import websockets
except ImportError:
print("pip3 install websockets", file=sys.stderr)
sys.exit(1)
WS_URL = "ws://localhost:18080/scm"
async def send_cmd(cmd, timeout=None):
"""Send a single Scheme command, print the reply."""
async with websockets.connect(WS_URL, ping_interval=20, ping_timeout=60,
max_size=10_000_000) as ws:
await ws.send(cmd)
try:
reply = await asyncio.wait_for(ws.recv(), timeout=timeout)
print(reply)
except asyncio.TimeoutError:
print(f"[timeout after {timeout}s — command may still be running in CogServer]")
async def send_file(path, timeout=None):
"""Send a file load command. The file runs inside the CogServer process."""
# Use absolute path so CogServer can find it regardless of cwd
import os
abspath = os.path.abspath(path)
cmd = f'(load "{abspath}")'
print(f">>> {cmd}")
await send_cmd(cmd, timeout=timeout)
async def interactive():
"""Interactive REPL — type Scheme, get replies."""
print(f"Connected to {WS_URL}")
print("Type Scheme expressions. Ctrl-C to quit.\n")
async with websockets.connect(WS_URL, ping_interval=20, ping_timeout=60,
max_size=10_000_000) as ws:
loop = asyncio.get_event_loop()
while True:
try:
line = await loop.run_in_executor(None, lambda: input("scm> "))
if not line.strip():
continue
await ws.send(line)
reply = await ws.recv()
print(reply)
except (EOFError, KeyboardInterrupt):
print("\nbye")
break
def main():
args = sys.argv[1:]
if not args:
print(__doc__)
sys.exit(0)
if args[0] == "-i":
asyncio.run(interactive())
elif args[0] == "-f":
if len(args) < 2:
print("Usage: cog-ws.py -f <file.scm>", file=sys.stderr)
sys.exit(1)
# Long-running file loads: no timeout (wait indefinitely)
asyncio.run(send_file(args[1], timeout=None))
else:
# One-shot command with generous timeout
cmd = " ".join(args)
asyncio.run(send_cmd(cmd, timeout=300))
if __name__ == "__main__":
main()