""" Wrapper for bbpPairings — calls the external binary, parses output. """ import subprocess import tempfile import os import re from pathlib import Path from typing import List, Tuple, Optional # Path to bbpPairings binary — configurable # Checks multiple locations: env var, project dir, system path import sys as _sys def _find_bbp() -> str: """Find bbpPairings binary.""" # Check env var env_path = os.environ.get('BBP_PATH', '') if env_path and os.path.exists(env_path): return env_path # Check project root (where this file is) project_root = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) candidates = [ os.path.join(project_root, 'bbpPairings.exe'), '/usr/local/bin/bbpPairings.exe', '/tmp/bbpPairings/bbpPairings.exe', ] for path in candidates: if os.path.exists(path): return path return env_path or '/usr/local/bin/bbpPairings.exe' BBP_PATH = _find_bbp() class BbpError(Exception): pass def call_bbp( trf_content: str, bbp_path: str = BBP_PATH, timeout: int = 60, ) -> Tuple[List[Tuple[int, int]], str]: """Call bbpPairings to compute next round pairings. Args: trf_content: TRF file content. bbp_path: Path to bbpPairings binary. timeout: Timeout in seconds. Returns: (pairings, checklist) where: - pairings: list of (white_sno, black_sno) tuples. If black_sno==0, white_sno has a bye. - checklist: human-readable checklist text (only if -l flag used). Raises: BbpError: If bbpPairings fails. """ if not os.path.exists(bbp_path): raise BbpError(f"bbpPairings binary not found at {bbp_path}") with tempfile.NamedTemporaryFile( mode='w', suffix='.trf', delete=False, encoding='utf-8' ) as f: f.write(trf_content) trf_path = f.name try: result = subprocess.run( [bbp_path, '--dutch', trf_path, '-p'], capture_output=True, text=True, timeout=timeout, ) except subprocess.TimeoutExpired: raise BbpError("bbpPairings timed out") finally: os.unlink(trf_path) if result.returncode != 0: stderr = result.stderr.strip() raise BbpError(f"bbpPairings exited with {result.returncode}: {stderr}") pairings = parse_pairing_output(result.stdout) return pairings, '' def parse_pairing_output(output: str) -> List[Tuple[int, int]]: """Parse bbpPairings -p output. Format: ... 0 """ lines = output.strip().split('\n') if not lines: return [] try: count = int(lines[0].strip()) except ValueError: raise BbpError(f"Invalid pairing output: {output[:200]}") pairings = [] for line in lines[1:]: line = line.strip() if not line: continue parts = line.split() if len(parts) != 2: continue try: white = int(parts[0]) black = int(parts[1]) except ValueError: continue pairings.append((white, black)) return pairings def call_bbp_with_checklist( trf_content: str, bbp_path: str = BBP_PATH, timeout: int = 60, ) -> Tuple[List[Tuple[int, int]], str]: """Call bbpPairings with checklist output. Returns: (pairings, checklist_text) """ if not os.path.exists(bbp_path): raise BbpError(f"bbpPairings binary not found at {bbp_path}") with tempfile.NamedTemporaryFile( mode='w', suffix='.trf', delete=False, encoding='utf-8' ) as f: f.write(trf_content) trf_path = f.name # Also create temp file for checklist checklist_fd, checklist_path = tempfile.mkstemp(suffix='.txt') os.close(checklist_fd) try: result = subprocess.run( [bbp_path, '--dutch', trf_path, '-p', '-l', checklist_path], capture_output=True, text=True, timeout=timeout, ) except subprocess.TimeoutExpired: raise BbpError("bbpPairings timed out") finally: os.unlink(trf_path) if result.returncode != 0: os.unlink(checklist_path) stderr = result.stderr.strip() raise BbpError(f"bbpPairings exited with {result.returncode}: {stderr}") pairings = parse_pairing_output(result.stdout) # Read checklist with open(checklist_path, 'r') as f: checklist = f.read() os.unlink(checklist_path) return pairings, checklist