pbn_util.py

from collections import defaultdict
import logging
import bridgebots
from pathlib import Path
from datetime import datetime
import pandas as pd
import tempfile
from collections import Counter


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 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 = bridgebots.pbn._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_single_pbn_string(
        data: pd.DataFrame,
        date=datetime.today(),
        board_no=1,
        event='',
        site='',
    ) -> str:
    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 = 'N:'
    deal += ' '.join(
        ['.'.join(data[col]) for col in data.columns[1:]]
    ) # 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'

    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:
                f.write('\n*\n')        
            f.write(v)
    return fn


def validate_pbn(pbn_string):
    try:
        deal, record = parse_single_pbn_record(pbn_string)
    except AssertionError:
        raise ValueError('Everyone should have 13 cards')
    except Exception as e:
        print('test')
        raise Exception(e)
    hands = deal.hands
    duplicated = set()
    missing = set()
    validation_dict = {}
    for suit in bridgebots.Suit:
        cards = [hands[direction].suits[suit] for direction in hands]
        # assert len(cards) == 13, 
        cards = [c for c in sum([hands[direction].suits[suit] for direction in hands], [])]
        duplicated.update([bridgebots.Card(suit,val) for val,cnt in Counter(cards).items() if cnt >1])

        cards_set = set(cards)
        missing.update([bridgebots.Card(suit,r) for r in bridgebots.Rank if not r in cards_set])

    err_msg = ''
    if len(duplicated) > 0:
        err_msg += f'Duplicated cards: {duplicated}. '
    if len(missing) > 0:
        err_msg += f'Missing cards: {missing}. '

    for direction in hands:
        num_cards = len(hands[direction].cards)
        if not num_cards == 13:
            err_msg += '{direction.name} has {num_cards} cards. '
    
    if err_msg:
        raise ValueError(err_msg)