# -*- coding: utf-8 -*- """皮卡鱼异步封装(asyncio 子进程, 替代旧的同步实现)。""" import asyncio import os ENGINE_DIRS = [ os.path.join(os.path.dirname(os.path.dirname(os.path.abspath(__file__))), 'Pikafish-2026-09-06'), ] NNUE_FILE = 'pikafish.nnue' def find_engine(): """按当前平台挑选引擎二进制: Windows 优先 .exe, 其次 Linux/Mac。""" import platform system = platform.system() # Windows / Linux / Darwin for base in ENGINE_DIRS: if not os.path.isdir(base): continue candidates = [] for name in sorted(os.listdir(base)): path = os.path.join(base, name) if not os.path.isfile(path): continue lower = name.lower() if 'pikafish' not in lower or lower.endswith(('.nnue', '.txt', '.md')): continue if system == 'Windows': ok = lower.endswith('.exe') and 'android' not in lower elif system == 'Darwin': ok = 'macos' in lower else: ok = 'linux' in lower and 'android' not in lower if ok: candidates.append(path) if candidates: return candidates[0], base return None, None class Pikafish: """异步 UCI 引擎进程。非法局面会导致引擎退出, 调用方负责 restart()。""" def __init__(self, threads=None, hash_mb=256, movetime=800, log=None): found, workdir = find_engine() if not found: raise RuntimeError('未找到皮卡鱼引擎二进制') self.path, self.workdir = found, workdir self.threads = threads or os.cpu_count() or 1 self.hash_mb = hash_mb self.movetime = movetime self.log = log or (lambda m: None) self._proc = None self._lock = asyncio.Lock() async def start(self): self._proc = await asyncio.create_subprocess_exec( self.path, cwd=self.workdir, stdin=asyncio.subprocess.PIPE, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.DEVNULL) await self._send('uci') await self._wait_for('uciok') nnue = os.path.join(self.workdir, NNUE_FILE) opts = [('Threads', str(self.threads)), ('Hash', str(self.hash_mb))] if os.path.isfile(nnue): opts.append(('EvalFile', NNUE_FILE)) for k, v in opts: await self._send(f'setoption name {k} value {v}') await self._send('isready') await self._wait_for('readyok') await self._send('ucinewgame') self.log(f'引擎已启动: {os.path.basename(self.path)} ' f'(threads={self.threads}, hash={self.hash_mb}MB)') return self async def restart(self): await self.quit() await self.start() async def _send(self, cmd): self._proc.stdin.write((cmd + '\n').encode()) await self._proc.stdin.drain() async def _readline(self): line = await self._proc.stdout.readline() if not line: raise RuntimeError('引擎进程已退出') text = line.decode(errors='replace').strip() self.log('<< ' + text) return text async def _wait_for(self, token): while True: if token in await self._readline(): return async def bestmove(self, moves=None, fen=None, movetime=None, depth=None): """返回 (move, score)。fen 优先于 moves; depth 优先于 movetime。""" async with self._lock: if fen: await self._send(f'position fen {fen}') else: pos = 'position startpos' if moves: pos += ' moves ' + ' '.join(moves) await self._send(pos) if depth: await self._send(f'go depth {depth}') else: await self._send(f'go movetime {movetime or self.movetime}') score = None while True: line = await self._readline() if line.startswith('info') and ' pv ' in line: i = line.find('score ') if i >= 0: parts = line[i:].split() if len(parts) >= 3: score = f'{parts[1]} {parts[2]}' if line.startswith('bestmove'): mv = line.split()[1] if mv in ('(none)', '(null)'): raise ValueError('该局面无可走之子') return mv, score async def quit(self): if self._proc and self._proc.returncode is None: try: self._proc.stdin.write(b'quit\n') await self._proc.stdin.drain() await asyncio.wait_for(self._proc.wait(), timeout=5) except Exception: self._proc.kill()