niklasf
commented
4 years ago It's possible, but requires a bit of plumbing. For example, say Stockfish is tethered to port 4000:
#!/bin/bash
coproc stockfish
nc -l -p 4000 <&"${COPROC[0]}" >&"${COPROC[1]}"Then a client would look like this:
import logging
import asyncio
import chess.engine
logging.basicConfig(level=logging.DEBUG)
class ProtocolAdapter(asyncio.Protocol):
def __init__(self, protocol):
self.protocol = protocol
def connection_made(self, transport):
self.transport = TransportAdapter(transport)
self.protocol.connection_made(self.transport)
def connection_lost(self, exc):
self.transport.alive = False
self.protocol.connection_lost(exc)
def data_received(self, data):
self.protocol.pipe_data_received(1, data)
class TransportAdapter(asyncio.SubprocessTransport, asyncio.ReadTransport, asyncio.WriteTransport):
def __init__(self, transport):
self.alive = True
self.transport = transport
def get_pipe_transport(self, fd):
return self
def write(self, data):
self.transport.write(data)
def get_returncode(self):
return None if self.alive else 0
def get_pid(self):
return None
def close(self):
self.transport.close()
# Unimplemented: kill(), send_signal(signal), terminate(), and various flow
# control methods.
async def main():
loop = asyncio.get_running_loop()
_, adapter = await loop.create_connection(
lambda: ProtocolAdapter(chess.engine.UciProtocol()),
"127.0.0.1", 4000)
engine = adapter.protocol
await engine.initialize()
# Example: isready
await engine.ping()
# Example: go
print(await engine.play(chess.Board(), chess.engine.Limit(time=2.0)))
if __name__ == "__main__":
asyncio.run(main())The transport adaper is required to provide get_returncode() and get_pid(), which usually do not make sense for sockets. It also has to provide get_pipe_transport() and pretend the socket is stdin/stdout of a process.
Mk-Chan
sshivaji
PedanticHacker
I wanted to request a feature: Allow communicating using the UCI/Xboard protocol over a TCP socket.
The use-case I had in mind was to be able to essentially decouple the engine provider from python-chess. For example I have written an application that enables votechess over irc (https://github.com/Mk-Chan/votechess-bot). It communicates with https://github.com/ShailChoksi/lichess-bot. I had to hack together a custom UCIEngine wrapper which just sends UCI commands over a socket (and similarly reads it for responses) and causes the votechess-bot to initiate a vote in the irc channel.
Although that's using the older UCI/Xboard engine APIs so maybe there is a different way with the current engine API or maybe you can suggest an alternative. I looked at the AsyncSSH example in the docs but I'm not sure if that's the same thing.
I was thinking something very similar to
popenbut instead just aconnectto a server-socket and listen until connection is closed by the server or the python-chess client