Initial commit

Set up a UCI engine that plays a random move

.gitignore

@@ -0,0 +1,2 @@
+.eggs
+badchess.log

.gitmodules

@@ -0,0 +1,3 @@
+[submodule "lichess-bot"]
+	path = lichess-bot
+	url = https://github.com/lichess-bot-devs/lichess-bot.git

.python-version

@@ -0,0 +1 @@
+badchess

README.md

@@ -0,0 +1,38 @@
+# badchess
+
+A bad chess engine for learning. Implements the [UCI Protocol](https://backscattering.de/chess/uci/) for communication with chess GUIs.
+
+Currently just plays a random legal move!
+
+## Installation
+
+Clone this repo, including git submodules:
+
+    git clone --recurse-submodules https://github.com/chaconinc/MainProject
+
+Install the python dependencies (I use pyenv, but adapt as you wish):
+
+    pyenv virtualenv badchess
+    pyenv local badchess
+    pip install -r requirements.txt
+
+## Usage
+
+### Locally, in `xboard`
+
+You can play against the engine in `xboard`, a chess GUI. From this repo's directory, start `xboard` with the following arguments:
+
+    xboard -fcp badchess/badchess.py -fd . -fUCI
+
+### As a bot on lichess.org
+
+You can also set up your engine as a bot in [lichess](https://lichess.org). The engine will run on your computer but communicate
+
+1. Make sure you have the files within the git submodule `lichess-bot/`. If not, try `git submodule update --init`.
+2. cd into the lichess-bot directory: `cd lichess-bot`
+3. Create a [lichess OAuth token](https://github.com/lichess-bot-devs/lichess-bot/wiki/How-to-create-a-Lichess-OAuth-token), storing the token as environment variable `LICHESS_BOT_TOKEN`.
+4. Update your lichess account to a bot account by running `python lichess-bot.py -u`
+5. Run the bot using the provided configuration file: `python lichess-bot.py --config ../lichess-bot-config.yml`
+6. You should be able to find your bot on lichess.org and challenge it!
+
+

badchess/badchess.py

@@ -0,0 +1,153 @@
+#!/usr/bin/env python
+
+from concurrent.futures import ThreadPoolExecutor
+import logging
+import random
+import selectors
+import sys
+import queue
+
+import chess
+
+logging.basicConfig(
+    filename="badchess.log",
+    encoding="utf-8",
+    level=logging.DEBUG,
+    format="%(levelname)s %(asctime)s %(message)s",
+)
+
+should_exit = False
+
+
+def producer(q):
+    """
+    Read a line from stdin. Check to see whether we should exit every second.
+    """
+
+    def read_input(stdin, mask):
+        line = stdin.readline()
+        if not line:
+            sel.unregister(sys.stdin)
+            return None
+        else:
+            return line.strip()
+
+    sel = selectors.DefaultSelector()
+    sel.register(sys.stdin, selectors.EVENT_READ, read_input)
+
+    while not should_exit:
+        logging.debug(f"Producer: {should_exit}")
+        for key, events in sel.select(timeout=1):
+            callback = key.data
+            line = callback(sys.stdin, events)
+            if line:
+                q.put(line)
+                logging.debug(f"Producer: {line}")
+    logging.debug(f"Producer done")
+
+
+def consumer(q, board, i):
+    while not should_exit:
+        logging.debug(f"Consumer {i}: {should_exit}")
+        try:
+            line = q.get(timeout=1)
+            if line:
+                logging.debug(f"Consumer: {line}")
+                process_command(line, board)
+        except queue.Empty:
+            pass
+    logging.debug(f"Consumer {i} done")
+
+
+def send_command(command):
+    logging.debug(f"Command sent: {command}")
+    sys.stdout.write(f"{command}\n")
+    sys.stdout.flush()
+
+
+def process_command(command, board):
+    logging.debug(f"Command received: {command}")
+    words = command.strip().split(" ")
+
+    if words[0] == "uci":
+        process_uci(words, board)
+    elif words[0] == "setoption":
+        process_setoption(words, board)
+    elif words[0] == "isready":
+        process_isready(words, board)
+    elif words[0] == "ucinewgame":
+        process_ucinewgame(words, board)
+    elif words[0] == "position":
+        process_position(words, board)
+    elif words[0] == "go":
+        process_go(words, board)
+    elif words[0] == "stop":
+        process_stop(words, board)
+    elif words[0] == "quit":
+        process_quit(words, board)
+
+
+def process_setoption(words, board):
+    logging.warning("setoption ignored")
+
+
+def process_uci(words, board):
+    send_command("id name badchess")
+    send_command("id author Kevin Schaul")
+    send_command("uciok")
+
+
+def process_isready(words, board):
+    send_command("readyok")
+
+
+def process_ucinewgame(words, board):
+    board.reset()
+
+
+def process_position(words, board):
+    """
+    position [fen <fenstring> | startpos ]  moves <move1> .... <movei>
+
+    set up the position described in fenstring on the internal board and
+    play the moves on the internal chess board.
+    if the game was played  from the start position the string "startpos" will be sent
+    """
+    if words[1] == "startpos" and words[2] == "moves":
+        board.reset()
+        [board.push_uci(move) for move in words[3:]]
+    elif words[1] == "fen":
+        board = chess.Board(" ".join(words[2:8]))
+        if len(words) >= 9 and words[8] == "moves":
+            [board.push_uci(move) for move in words[9:]]
+
+
+def process_go(words, board):
+    move = random.choice([move for move in board.legal_moves])
+    send_command(f"bestmove {move.uci()}")
+
+
+def process_stop(words, board):
+    # TODO stop other consumer threads
+    pass
+
+
+def process_quit(words, board):
+    global should_exit
+    should_exit = True
+    logging.debug(f"exit set")
+
+
+def main():
+    q = queue.Queue()
+    board = chess.Board()
+
+    n_consumers = 2
+
+    with ThreadPoolExecutor(max_workers=n_consumers + 1) as executor:
+        executor.submit(producer, q)
+        [executor.submit(consumer, q, board, i) for i in range(n_consumers)]
+
+
+if __name__ == "__main__":
+    main()