さて
自分のサイトにやねうらおうさん提供の５００万局の詰将棋局面を任意に表示できるページを前回作成したわけだが、詰め手順まではついていない。 これはもともと提供される局面にも回答は添付されていないのであって、基本的にはじぶんで解くか（それほど難しい問題はそれほど無い）さもなくば将棋エンジン（やねうらおうなど）を自分のパソコンに導入して局面をsfenではりつけて検索させれば回答は導け出せますという使い方が前提になっているわけだ。
自分のサイトでもページからsfenを取得できるので詰将棋が解ける環境を自分のパソコン上に持っていればいいことになる。
ただ前にも書いたが、サイトのページ上で回答もそのまま見ることができるようになれば便利なことは確かである。
これを実現しようとすると最初に考えるのは詰将棋を解くAPIを用意してWeb PageからオンデマンドそのAPIを叩けに行けば良いという方法。 これにはAPIを供給するサーバーが必要になる。 すこし探してみたらClaude.ai関連のMCPサーバーの実装で将棋エンジンとブリッジするサーバー構築をexpressを使った形ですでにGitHubにあげている方がいた。azumausu/shogi-mcp
実際にこれでサーバーを作成し、やねうらおうを導入してみたら、うまく作動することは確認できた。 でもやはり、サーバーを常にスタンバイさせておかなければならない、という課題は残るし、リアルタイムに詰みを返せるような高速のエンジンを動かすためのバーチャルマシンはそれなりのスペックのものが必要になる。 こちらは最低限のリソースとマシンスペックでサイトを運営するのを旨としているので、方針がそぐわない。
よって考え方を切り替え、時間がかかってもよいので退職以来３年間眠っている業務用のノートパソコンでひたすら局面を解析させて５００万問の詰め手順を導き出し、これをデータベースのテーブル上に格納してここから答えを引き出す方針で進めることにした。
これを行うためには、ノートパソコンで将棋エンジンを起動し、これに局面を送って、詰みの回答を読み取ることを延々と繰り返すというスクリプトが必要になる。
なので、以下Pythonのお勉強の続きをすることになったわけです。
将棋のEngineと通信する部分のモジュールを以下のような構成で考える。
将棋エンジン（やねうらおう）をサブプロセスで起動する。
このプロセスにstdioを通じて詰将棋を解いてね、というコマンドを送り 答えをいただく。
これを問題の数だけ繰り返す。
すべて終わったらサブプロセスを終了
ちなみにこのエンジンを起動させたあと、コンソールで詰将棋を解くためには、
まず “usi” コマンドを送って、”usiok”が戻ってくるのをまってUSIプロトコルが働くことを確認し、そののち setoption name xxx value yyyというフォーマットでオプション類を設定する。
これが終わったら、以下のプロセスを問題の数だけ繰り返す。
”isready”を送ってエンジンの局面を初期化”readyok”と返ってくるのを待つ。
“position sfen sfen文字列” と送って局面を通達。
“go mate” で詰み検索を開始。
stdoutに返されてくるinfo文字列を読み解きこの中に mate xx (xx は例えば１１手詰めならば11) という文字が含まれていれば、所定手数の詰みを見つけたということなので ”stop”コマンドを送信して検討をストップさせ、文字列の中から詰め手順だけ抜き出して親プロセスに返す。 （親プロセスで画面表示するなりファイルにセーブするなりデータベースに書き込むなりの処理をする）
すべて終了したらサブプロセスを終了しスクリプトも終了
ということだと思うので、この考えをもとに実際に書いてみました
import re
def getMateString(line,mateLength):
"""
Take string and expected mate length. return True if mate in equal to or less than the expected length
:param line: string to be evaluated for "mate nn" substring
:param mateLength: expected mate length
:return: True if mate in equal to or less than the expected length. None otherwise
"""
pattern = r"mate\s(\d+)\s"
matches=re.findall(pattern,line)
if matches:
if int(matches[0])<=mateLength:
return matches[0]
else:
return None
ヘルパーFunction その１。 詰めたよ、という文字列が所定手数になっているかの確認。11手詰めのはずの詰めだと13手とか17手の詰めの解答が最初戻ってくるのでそれを無視するためのもの。 ただし、11手詰めのはずなに9手とか7手で詰みます、といってくることがあることがわかったので所定手数以下の文字列についても正解と判断するようになっている。
import subprocess
def start_engine(binaryExecutable, options):
"""
Executes the shogi engine binary and returns the subprocess object.
If log is True, writes standard I/O streams to log.txt and err.txt.
:param binaryExecutable: path to shogi engine
:param options: options passed to shogi engine
:return: subprocess object
"""
try:
proc= subprocess.Popen([binaryExecutable], stdin=subprocess.PIPE, stdout=subprocess.PIPE,
universal_newlines=True)
# Send 'usi' command and wait until 'usiok' is received
proc.stdin.write("usi\n")
proc.stdin.flush()
while True:
line = proc.stdout.readline().strip()
if line == "usiok":
break
for key, value in options.items():
proc.stdin.write('setoption name %s value %s\n' % (key, value))
proc.stdin.flush()
return proc
except Exception as e:
print(f"Error executing engine: {e}")
return None
start_engineという関数は将棋エンジンをsubProcessで起動し、option設定を終えたのちにこのプロセスオブジェクトをメインプログラムに返す。`
ちなみにEngineとOptionsについては以下をメインにて定義する。非力なノートパソコンで走らせるのでThreadをデフォルトの４から２にしないとエンジンが固まることがあった。FV_SCALEは水匠の推奨値。定跡ファイルは必要ないのでno_bookを指定
executable = "./engine/engine"
options = {"BookFile": "no_book", "FV_Scale": "24", "Threads": "2"}このプロセスオブジェクトででエンジンが動いているので、stdin/stdoutのパイプを使って交信することができるようになる。
Send_command関数でSFENの局面をエンジンに渡し回答を得るのだが、そのためのヘルパー関数を用意する
def initialize_engine(proc):
"""
send stop then 'isready' to initialize the engine for next command. Will block if "readyok" is not returned
It is assumed 'readyok' response is guaranteed after sending "isready" command. Otherwise, the function will hung
:param proc: chile process object
:return: None
"""
proc.stdin.write("stop\n")
proc.stdin.flush()
proc.stdin.write("isready\n")
proc.stdin.flush()
while True:
line = proc.stdout.readline()
if line == 'readyok\n':
break
def try_mate_again(proc):
"""
stop engine and try mate eval again.
:param proc: child process object
:return: None
"""
proc.stdin.write("stop\n")
proc.stdin.flush()
proc.stdin.write("go mate\n")
proc.stdin.flush()
def query_mate(proc,sfenString,mateLength):
proc.stdin.write(f"position sfen {sfenString}\n")
proc.stdin.flush()
proc.stdin.write(f"go mate {mateLength}\n")
proc.stdin.flush()
initialize_engine関数ではstopを送って探索を終了させたのち、isreadyを送って局面値を初期化する。 try_mate_againではisreadyを送らず、go mateを送るので、そのときメモリー上にある局面を改めて探索して詰みを探す。
query_mate関数はsfen文字列をpositionとしてセットし、go mateを送って詰みの探索開始するためのもの
send_commandsはこれらのヘルパーを使いながら詰みを検索。
import os
def send_commands(proc, sfenString, mateLength, problemNumber=0):
"""
Sends commands to the running shogi engine.
:param proc: child process object
:param sfenString: sfen string to send to shogi engine
:param mateLength: length of mate
:param problemNumber: optional problem number. Shogi Engine does not need it. this is used to identify the problem number for logging purpose.
:return: String for mating move
"""
try:
# Send 'isready' command and wait for 'readyok' response
initialize_engine(proc)
# Send 'position sfen' command and wait for 'info mate' response
timeout=20
line=""
time_limit =time.time()+timeout
retryCount=50
query_mate(proc, sfenString, mateLength)
count=0
os.set_blocking(proc.stdout.fileno(), False) #make stdout/in no blocking
while True:
line = proc.stdout.readline().strip()
if line:
if getMateString(line, mateLength):
if count > 0:
with open("log.txt", "a") as log:
log.write(f"retry attempted #{count} times at problem #{problemNumber}\n")
break
else:
#if stdout is blank, check for timeout condition
if time.time() > time_limit:
if count < retryCount:
#if timeout happened and retry count is not maxed, attempt a retry
count += 1
swipe_print(f"time limit reached, retrying {count} :")
time_limit = time.time() + timeout
# initialize_engine(proc)
# query_mate(proc, sfenString, mateLength)
try_mate_again(proc)
else:
swipe_print(f"time limit reached, giving up. Problem number: {problemNumber}")
with open('exception.txt', 'a') as f:
f.write(f"Giving up. Use the second best solution for problem id: {problemNumber}.\n")
break
os.set_blocking(proc.stdout.fileno(), True) #change back I/O to blocking mode
proc.stdin.write("stop\n")
proc.stdin.flush()
return line.split("pv ")[-1].strip()
swipe_printはコンソールで、画面がスクロールしないように同じ行に出力を重ねていくヘルパー関数
def swipe_print(text):
print("\r", end="")
print(text, end='')
で、send_commandを５００万回繰り返せばすべての解答が得られるはず
stop_engine関数でエンジンを終了しサブプロセスを終了。proc.communicateはプロセスの終了処理も行ってくれる。
def stop_engine(proc):
"""
Stops the running shogi engine.
"""
stdout, stderr = proc.communicate(input="quit\n", timeout=5)
if stderr:
raise Exception(stderr.decode("utf8"))
print("Engine stopped")
で、下のスクリプトでは上の関数を適宜に呼び出して、mate11.sfenの最初の500行を解いてみている。 その結果をみながら、上のsend_commandの中身は何度か修正している。
'''
Test program
'''
from swipe_print import swipe_print
# Execute the engine with logging
start_time = time.time()
executable = "./engine/engine"
options = {"BookFile": "no_book", "FV_Scale": "24", "Threads": "2"}
proc = start_engine(executable, options)
if proc:
print("Engine executed successfully.")
#soliving problem in text file (first 500 problems)
N=500
with open("mate11.sfen","r") as f:
sfens=f.readlines()[:N]
for index, sfen in enumerate(sfens):
pv=send_commands(proc, sfen, 11,index)
if pv:
swipe_print(f"problem#:{index}\tPrincipal variation: {pv}")
# Stop the engine
stop_engine(proc)
print (f"Time elapsed: {time.time()-start_time}")いつまで待っても所定手数の詰みが見つけられない、という可能性もあることがわかったのでタイムアウトを設けることにした。 最初はThreadingでタイムアウトを生じさせ、このときKeyboardInterruptを生成してこれを処理する手法を試してみたがCTL-Cが乗っ取られるのがどうも気に食わない。 async.ioライブラリーも読んでみたがブロックIOとどのように整合させてよいのかがよくわからない。 ので最初はどろくさくWhileループの中で時間をチェックして一定時間たっていたらbreakするという簡単な方法で書いてみた。 ただし、これポーリングで時間を見ているだけなので、ループ中のstdout.readline()がブロックしている限り時間切れにはならない。タイムアウト１５秒くらいまでなら、だらだらと文字列が出力されているようなのでとりあえずハングすることはないとたかをくくって書いてみたら問題なく動いている。 これで良しにしようと思っていたが、さらに日を置いて調べてみたら、os.set_blockingという関数を見つけた。Linuxのみに対応かと思ったら、WindowsのpipeにもPython3.12から対応しました、となっていて、これはラッキー。 これをつかうとstdioがブロックされなくなり、readline()で何もないときには空の文字列が返ってくるようになるので、Timeoutも割り込みなどの処理を使うことなく検知できるようになった。 また、エンジン、検討途中でドツボにはまって正解がなかなか出せなくなる、ということも起こるということを経験的に知った。この場合は出力を待つよりも一度ご破算にして解答を聞き直すとすぐに正解にたどり着くような現象が多くみられた。（実に人間のようである）のでタイムアウトをやたら長くするより、リトライと組み合わせたほうが効率が良いと考えリトライも追加。たいてい２回か３回のリトライで正解にたどり着くようだが、中には３０回目４０回目のリトライで正解、というケースも５００問中２問や３問発生し、それでも正解にたどり着かなかった場合はあきらめて、最後に読んだ文字列を取得するようにしてみた。何回か試してみたが、試すごとにエンジンがドツボにはまる問題には、ある程度の一貫性はあるものの、前回３０回以上リトライした問題が次のRunでは最初のトライで解答がでてきたり、このあたりのランダム性は不思議だなあと思うのである。この設定による５００問１１手詰めの探索で、全問に11手の（またはそれ以下）詰め手順が返されてくるのはタタイムアウトを１５秒以上、リトライ回数を５０回マックスくらいにすればよさそうだ。タイムアウトをもっと長くすれば確実性は増しそうだが、全部の問題を解く時間がどんどん長くなる。プログラムの出力を見ていると、９割以上の問題は２，３秒で指定手数の正解が出てしまうのであるが、つまずいてリトライが３０回発生する問題は解くのに１０分かかってしまうということである。またリトライ無しにすれば全問にかかる時間は少なくなるが不完全な（１３手詰め以上の）解答が増えてくる。とりあえず時間節約のため、全部解かせたあとで、１１手詰めになっていない問題をあとから再度検索させる方法も考えたが、２度手間になってしまうので時間がかかっても全問指定手数の解答までたどり着く方向で考えることにしたのだが、 この実験結果から試算すると１１手詰めの問題100万個すべてに11手以下の詰みをみつけるのには最低２か月はかかりそうだ。 ちなみに3手詰めのほうはリトライなどはさすがに発生してしないので４日ほどで終わりそうである。
なにせノートブック用２コア４スレッドのスカイレークという非力なCPUの上にRAMも８ギガバイトである。遅いのは致し方がない。改めて５００万問という数字の理不尽さをしみじみ感じてしまうと同時にCPUの性能のちがいが顕著に見えてくる将棋エンジンのコンピューターパワーの使いたおし方は凄いなあと思う。
とにかく、mysqlのテーブルに出力された文字列を書き込むパイソンスクリプトも別途作成し、上記のスクリプトを組み合わせて、５００万問を３手詰めから順番になめていくメインプログラムを起動させたのが三日前。画面をブランクにし、ひっそりたたずむノートパソコン。 ときおり、画面を開いてはコンソール上に出てくる問題番号を見て進行状況を確認しております。 うんうん、まだ３手詰のあたりは順調だが道は長い。（長すぎる！）