pbn_util.py

from collections import defaultdict
import logging
import bridgebots
from bridgebots import PlayerHand, Rank
from pathlib import Path
from datetime import datetime
import pandas as pd
import tempfile
from dds_util import get_dd_tricks
from typing import List, Dict
import os


DEALER_LIST = ['N', 'E', 'S', 'W']
VULNERABILITY_LIST = ["None","NS","EW","All","NS","EW","All","None","EW","All","None","NS","All","None","NS","EW"]
SUITS = [bridgebots.Suit.SPADES, bridgebots.Suit.HEARTS, bridgebots.Suit.DIAMONDS, bridgebots.Suit.CLUBS]


def _build_record_dict(record_strings: List[str]) -> Dict:
    """
    Parse the pbn line by line. When a block section like "Auction" or "Play" is encountered, collect all the content of
     the block into a single entry
    :param record_strings: List of string lines for a single board
    :return: A dictionary mapping keys from the pbn to strings or other useful values (e.g. list of strings for the
    bidding record)
    """
    record_dict = {}
    # Janky while loop to handle non bracketed lines
    i = 0
    while i < len(record_strings):
        record_string = record_strings[i]
        if not (record_string.startswith("[") or record_string.startswith("{")):
            i += 1
            continue
        if record_string.startswith("{"):
            commentary = ""
            while i < len(record_strings):
                record_string = record_strings[i]
                if record_string.startswith("["):
                    break
                commentary += record_string + " "
                i += 1
            record_dict["Commentary"] = commentary.strip()
            continue
        if record_string.startswith("[") and "]" not in record_string:
            while "]" not in record_string:
                i += 1
                record_string = record_string + record_strings[i]
        record_string = record_string.replace("[", "").replace("]", "")
        key, value = record_string.split(maxsplit=1)
        value = value.replace('"', "")
        if key == "Note":
            number, message = value.split(":", maxsplit=1)
            key = key + "_" + number
            value = message
        record_dict[key] = value
        if key == "Auction":
            auction_record = []
            i += 1
            while i < len(record_strings):
                auction_str = record_strings[i]
                if "[" in auction_str:
                    break
                auction_record.extend(auction_str.split())
                i += 1
            record_dict["bidding_record"] = auction_record

        elif key == "Play":
            play_record = []
            i += 1
            while i < len(record_strings):
                play_str = record_strings[i]
                if "[" in play_str or play_str == "*":
                    break
                play_record.append(play_str.split())
                i += 1
            record_dict["play_record"] = play_record
        elif key == "OptimumResultTable":
            dds_tricks_table = []
            i += 1
            while i < len(record_strings):
                dds_str = record_strings[i]
                if "[" in dds_str or dds_str == "*":
                    break
                dds_tricks_table.append(dds_str.split())
                i += 1
            record_dict["dds_tricks_table"] = dds_tricks_table
        else:
            i += 1
    return record_dict

def parse_single_pbn_record(record_strings):
    """
    :param record_strings: One string per line of a single PBN deal record
    :return: Deal and BoardRecord corresponding to the PBN record
    """
    record_dict = _build_record_dict(record_strings)
    try:
        deal = bridgebots.pbn.from_pbn_deal(record_dict["Dealer"], record_dict["Vulnerable"], record_dict["Deal"])
    except KeyError as e:
        # if previous_deal:
        #     deal = previous_deal
        # else:
        raise ValueError("Missing deal fields and no previous_deal provided") from e
    # board_record = _parse_board_record(record_dict, deal)
    return deal, record_dict


def parse_pbn(file_path):
    """
    Split PBN file into boards then decompose those boards into Deal and BoardRecord objects. Only supports PBN v1.0
    See https://www.tistis.nl/pbn/pbn_v10.txt

    :param file_path: path to a PBN file
    :return: A list of DealRecords representing all the boards played
    """
    records_strings = bridgebots.pbn._split_pbn(file_path)
    # Maintain a mapping from deal to board records to create a single deal record per deal
    records = defaultdict(list)
    # Some PBNs have multiple board records per deal
    previous_deal = None
    for record_strings in records_strings:
        try:
            deal, board_record = parse_single_pbn_record(record_strings)
            records[deal].append(board_record)
            # previous_deal = deal
        except (KeyError, ValueError) as e:
            logging.warning(f"Malformed record {record_strings}: {e}")
    return [(deal, board_records) for deal, board_records in records.items()]


def create_pbn_file(
        data: pd.DataFrame,
        save_path = None,
        date=datetime.today(),
        board_no=1,
        event='',
        site='',
    ) -> str:
    if not save_path:
        fd, save_path = tempfile.mkstemp(suffix='.pbn')
        os.close(fd)

    year = date.strftime("%y")
    month = date.strftime("%m")
    day = date.strftime("%d")
    date_print = day + "." + month + "." + year
    dealer = DEALER_LIST[(board_no-1) % 4]
    vulnerability=VULNERABILITY_LIST[(board_no-1) % 16]
    deal = pbn_deal_string(data, dealer) # sss.hhh.ddd.ccc sss.hhh.ddd.ccc......

    file = ''
    file += ("%This PBN was generated by Bridge Hand Scanner\n")
    file += f'[Event "{event}"]\n'
    file += f'[Site "{site}"]\n'
    file += f'[Date "{date_print}"]\n'
    file += f'[Board "{str(board_no)}"]\n'
    file += f'[Dealer "{dealer}"]\n'
    file += f'[Vulnerable "{vulnerability}"]\n'
    file += f'[Deal "{deal}"]\n'
    file += pbn_optimum_table(deal)
    file += '\n' # End of board

    with open(save_path, 'w') as f:
        f.write(file)

    return file

def merge_pbn(pbn_paths):
    fd, fn = tempfile.mkstemp(suffix='.pbn', text=True)
    board_dict = {}
    for i, pbn_path in enumerate(pbn_paths):
        result = parse_pbn(pbn_path)[0]
        with open(pbn_path, 'r') as f2:
            pbn_str = f2.read()
        board_no = int(result[1][0]['Board'])
        board_dict[board_no] = pbn_str
        
    ordered_board_dict = dict(sorted(board_dict.items()))
    with open(fd, 'w') as f:
        for i, (k,v) in enumerate(ordered_board_dict.items()):
            if i != 0:
                if v[-1] != "\n":
                    f.write('\n')        
            f.write(v)
    return fn
    
def pbn_deal_string(df, dealer):
    assert dealer in DEALER_LIST, f'Dealer {dealer} is not valid'
    dealer_idx = DEALER_LIST.index(dealer)
    dealer_list_rolled = DEALER_LIST[dealer_idx:] + DEALER_LIST[:dealer_idx]
    deal = f'{dealer}:'
    hand_reprs = []
    for direction in dealer_list_rolled:
        hand = PlayerHand.from_string_lists(*df[direction])
        hand_reprs.append('.'.join([''.join([r.abbreviation() for r in hand.suits[suit]]) for suit in SUITS]))
    deal += ' '.join(hand_reprs)
    return deal

DD_DIRECTIONS = ['N', 'S', 'E', 'W']
DD_SUITS = ['NT', 'S', 'H', 'D', 'C']
def pbn_optimum_table(pbn_deal_string):
    try:
        dd_tricks_string = get_dd_tricks(pbn_deal_string)
    except Exception as e:
        print(e)
        raise ValueError(e)
    s = '[OptimumResultTable "Declarer;Denomination\\2R;Result\\2R"]\n'
    for i, char in enumerate(dd_tricks_string):
        direction = DD_DIRECTIONS[i//5]
        suit = DD_SUITS[i%5]
        tricks = int(char, 16)
        s += f'{direction} {suit} {tricks}\n'
    return s