pflow-tensors

generate_data.py

import math
import random

import torchflow
import torch
import examples.tictactoe


win_moves = [
    ["00", "01", "02"],
    ["10", "11", "12"],
    ["20", "21", "22"],
    ["00", "10", "20"],
    ["01", "11", "21"],
    ["02", "12", "22"],
    ["00", "11", "22"],
    ["20", "11", "02"],
]

empty_board = {
    '00': '',
    '01': '',
    '02': '',
    '10': '',
    '11': '',
    '12': '',
    '20': '',
    '21': '',
    '22': ''
}

m = torchflow.Model(
    schema='tic-tac-toe',
    machine=examples.tictactoe.machine
)


other_player = {'X': 'O', 'O': 'X'}  # alternate players
win_encode = {'tie': 0, 'X': 1, 'O': 2}

def game_over(board: dict):
    for _, value in board:
        if not value:
            return False
    return True


def is_winner(player, board: dict):
    for moves in win_moves:
        if board[moves[0]][:1] != player:
            continue
        if board[moves[1]][:1] != player:
            continue
        if board[moves[2]][:1] != player:
            continue
        return True

    return False


def compute_games(savefile: str = ''):
    """
    recursively create all games of tic-tac-toe
    """
    results = {'counter': 0, 'tie': 0, 'win': {'X': 0, 'O': 0}}
    initial = m.initial_vector()

    vector_space = len(m.transitions) + 1  # ['X00', 'O00',...'X22', 'O22'] + result
    game_space = 255168  # total games
    state_space = torch.zeros(game_space, vector_space)  # all games stored as tensors

    def store_game(game_board: dict, winner: str):
        """ construct a tensor to represent a given sequence of gameplay """
        game_history = torch.zeros(vector_space)
        for k, v in game_board.items():
            player_symbol = v[:1] # X
            move_num = v[1:] # 11
            if move_num:
                move_key = player_symbol+k
                game_history[m.transitions[move_key]['offset']] = int(move_num)+1  # count starting with 1

        game_history[-1] = win_encode[winner]
        state_space[results['counter']] = game_history
        results['counter'] += 1
        # print(game_history)

    # REVIEW: this is a forward step!! 
    def play_game(move=0, state=initial, board=empty_board):
        if math.remainder(move, 2) == 0.0:
            player = 'X'
        else:
            player = 'O'

        if move > 3 and is_winner(other_player[player], board):
            store_game(board, other_player[player])
            results['win'][other_player[player]] += 1
            return

        available_moves = m.valid_transitions(state)

        # random.shuffle(available_moves)
        if len(available_moves) == 0:
            store_game(board, 'tie')
            results['tie'] += 1
            return

        for row in available_moves:
            action = row['action']
            next_state = row['next_state']
            next_board = board.copy()
            next_board[action[1:]] = '%s%i' % (action[:1], move)  # board['11'] = 'X0'

            play_game(
                move + 1,
                next_state,
                next_board
            )
            # break # show only 1 game

    play_game()

    if savefile:
        torch.save(state_space, savefile)

    return results  # {'counter': 255168, 'win': {'X': 49392, 'O': 77904}}

pflow_tensors.py

import metamodel
import metamodel.error
import torch


# Extend pflow-metamodel package to support tensors
# https://pypi.org/project/pflow-metamodel/
class Model(object):

    places = {}
    transitions = {}

    def __init__(self, **kwargs):
        """
        convert metamodel vectors to tensors
        """
        self.m = metamodel.Model(kwargs['schema'], kwargs['machine'])
        self.places = self.m.places

        for label, txn in self.m.transitions.items():
            txn['delta'] = torch.tensor(txn['delta'], dtype=torch.int8)

            # TODO: include guards
            # for guardLabel, guard in txn['guards']:
            #     txn['guards'][guardLabel]['delta'] = torch.tensor(
            #         guard['delta'], dtype=torch.int8)

            self.transitions[label] = txn

    def initial_vector(self):
        return torch.tensor(self.m.initial_vector())

    def valid_transitions(self, state: torch.Tensor, limit=0):
        out = []
        for label, tx in self.transitions.items():
            ok, next_state, _ = self.transform(state, label)
            if ok:
                out.append({'action': label, 'next_state': next_state})
                if limit > 0 and len(out) >= limit:
                    break
        return out

    def transform(self, state: torch.Tensor, action: str, multiple: int = 1):

        if multiple < 0:
            raise metamodel.error.InvalidInput(
                'invalid multiple %s' % multiple)

        if action not in self.transitions:
            raise metamodel.error.InvalidInput('unknown action: %s' % action)

        t = self.transitions[action]
        out = state + t['delta']
        ok = True

        for _, p in self.places.items():
            i = p['offset']
            if out[i] < 0:  # underflow
                ok = False

            if 0 < p["capacity"] < out[i]:  # overflow
                ok = False

        return ok, out, t['role']