text:
from __future__ import annotations
from typing import List, Tuple, Union

from collections import deque
import numpy as np
import torch

from config import OthelloConfig
from utils.game import Othello
from utils.model import Network

class ReplayBuffer(object):
    def __init__(self, buffer: list):
        cfg = OthelloConfig()
        self._window_size = cfg.window_size
        self._batch_size = cfg.batch_size
        self._buffer = buffer

    def __len__(self) -> int:
        return len(self._buffer)

    def empty(self) -> bool:
        return len(self) < self._batch_size

    def save_training_data(self, training_data: List[Tuple[np.ndarray, np.ndarray, float, np.ndarray]]):
        for data_point in training_data:
            if len(self._buffer) >= self._window_size:
                self._buffer.pop(0)
            self._buffer.append(data_point)

    def sample_batch(self) -> Tuple[np.ndarray, np.ndarray, np.ndarray, np.ndarray]:
        indices = [int(i) for i in np.random.choice(range(len(self)), size=self._batch_size, replace=False)]
        samples = []
        for i in indices:
            samples.append(self._buffer[i])
        image, pi, z, actions_mask = zip(*samples)
        return (np.array(image).astype(np.bool), np.array(pi).astype(np.float32),
                np.expand_dims(np.array(z).astype(np.float32), -1), np.array(actions_mask).astype(np.bool))

class Node(object):
    def __init__(self, cfg: OthelloConfig, game: Othello, action_probs: np.ndarray):
        self._cfg = cfg
        self._game = game
        self._P = action_probs.copy()
        self._N = np.zeros(self._game.num_distinct_actions(), dtype=np.int64)
        self._Q = np.zeros(self._game.num_distinct_actions(), dtype=np.float32)
        self._W = np.zeros(self._game.num_distinct_actions(), dtype=np.float32)
        self._children = [None for _ in range(self._game.num_distinct_actions())]

    def child(self, action: int) -> Union[Node, None]:
        return self._children[action]

    def update_child(self, child: Union[Node, None], action: int):
        self._children[action] = child

    def game(self) -> Othello:
        return self._game

    def select_action(self) -> int:
        action_mask = self._game.legal_actions_mask()
        ucb = self._cfg.ucb_c_puct * self._P * (np.sqrt(self._N.sum()) / (1 + self._N))
        puct = self._Q + ucb
        puct = action_mask * (1 + puct)  # add 1 for case when all values of puct are zero
        action = int(puct.argmax())
        return action

    def select_optimal_action(self) -> int:
        target_policy = self.get_policy()
        action = int(target_policy.argmax())
        return action

    def get_policy(self) -> np.ndarray:
        tau = self._cfg.tau_initial if len(self._game) <= self._cfg.num_sampling_moves else self._cfg.tau_final
        n = self._N ** (1/tau)
        if n.sum() == 0:
            n += 1e-8
        policy = n / n.sum()
        return policy

    def add_returns(self, action: int, returns: np.ndarray):
        self._W[action] += returns[self._game.current_player()]
        self._N[action] += 1
        self._Q[action] = self._W[action] / self._N[action]

    @staticmethod
    def get_new_node(
            cfg: OthelloConfig, game: Othello, network: Network, device: torch.device
    ) -> Tuple[Node, np.ndarray, np.ndarray]:
        state_tensor = image_to_tensor(game.make_input_image(), device)
        with torch.no_grad():
            p, v = network.inference(state_tensor)
            actions_mask = torch.as_tensor(game.legal_actions_mask(), dtype=torch.float32).to(device)
            p = filter_legal_action_probs(p.unsqueeze(0), actions_mask.unsqueeze(0))
            p.squeeze(0)
        p, v = p.cpu().numpy(), v.cpu().numpy()
        new_node = Node(cfg, game, p)
        return new_node, p, v

def image_to_tensor(image: np.ndarray, device: torch.device) -> torch.Tensor:
    tensor = torch.as_tensor(image, dtype=torch.float32).to(device)
    return tensor

def filter_legal_action_probs(action_probs: torch.Tensor, action_mask: torch.Tensor) -> torch.Tensor:
    action_probs = action_probs + 0.01  # for case when all legal actions sum to zero
    masked_probs = action_mask * action_probs
    return masked_probs / masked_probs.sum(dim=1, keepdim=True)

def calculate_loss(
        predicted_action_probs: torch.Tensor, predicted_values: torch.Tensor,
        target_action_probs: torch.Tensor, target_values: torch.Tensor
) -> Tuple[torch.Tensor, torch.Tensor, torch.Tensor]:
    value_loss = torch.square(target_values - predicted_values)
    policy_loss = (target_action_probs * torch.log(predicted_action_probs)).sum(dim=1)
    final_loss = value_loss - policy_loss
    return -policy_loss.mean(), value_loss.mean(), final_loss.mean()

def generate_training_data(
        cfg: OthelloConfig, game: Othello, target_policies: np.ndarray, final_returns: np.ndarray
) -> List[Tuple[np.ndarray, np.ndarray, float, np.ndarray]]:
    assert len(target_policies) == len(game)
    dq = deque(maxlen=cfg.total_input_channels // 2)
    training_data = []  # list of (input_image, pi, z, action_mask)
    for _ in range(cfg.total_input_channels // 2):
        dq.appendleft(np.zeros((2, 8, 8), dtype=np.bool))
    for i in range(len(game)):
        img = game.history_state(i)
        player = game.history_player(i)
        action_mask = game.history_actions_mask(i)
        dq.appendleft(img)
        x = np.zeros((cfg.total_input_channels, 8, 8), dtype=np.bool)
        for ch, img in enumerate(dq):
            x[ch] += img[0]
            x[(cfg.total_input_channels // 2) + ch] += img[1]
        x[-1] += bool(player)
        training_data.append((x, target_policies[i], float(final_returns[player]), action_mask))
    return training_data

def mcts(node: Node, cfg: OthelloConfig, network: Network, device: torch.device) -> np.ndarray:
    if node.game().is_terminal():
        return node.game().returns()
    action = node.select_action()
    child = node.child(action)
    if child is not None:
        returns = mcts(child, cfg, network, device)
        node.add_returns(action, returns)
        return returns
    child_game = node.game().clone()
    child_game.apply_action(action)
    if child_game.is_terminal():  # if new created child node is terminal node
        returns = child_game.returns()
        node.add_returns(action, returns)
        p = np.zeros(child_game.num_distinct_actions(), dtype=np.float32)
        child = Node(cfg, child_game, p)  # create an empty node. It will never be in training data.
        node.update_child(child, action)
        return returns
    child, p, v = Node.get_new_node(cfg, child_game, network, device)
    node.update_child(child, action)
    returns = np.empty(2, dtype=np.float32)
    returns[child_game.current_player()] = v
    returns[1 - child_game.current_player()] = -v
    node.add_returns(action, returns)
    return returns

max_stars_repo_name: kushwahashivam/alphazero-othello

max_issues_repo_name: kushwahashivam/alphazero-othello

max_forks_repo_name: kushwahashivam/alphazero-othello


text:
'''
Deep neural network that plays omok
'''
import numpy as np
from omok.core.board import Board

class Network:
    EMPTY_SLOT = 0.0
    PLAYER_SLOT = 1.0
    ENEMY_SLOT = -1.0

    WIDTH = 40
    HEIGHT = 30
    STRUCTURE = [200]

    MODEL_DIR = 'omok/ai/models/'

    @staticmethod
    def sigmoid(X):
        X = 1.0 / (1.0 + np.exp(-X))
        return X

    @staticmethod
    def sigmoid_derivative(Y):
        Y = Y * (1 - Y)
        return Y

    @staticmethod
    def ReLU(X):
        X[X < 0] = 0
        return X

    @staticmethod
    def ReLU_derivative(Y):
        Y[Y > 0] = 1
        Y[Y < 0] = 0
        return Y

    def __init__(self, filename=None):
        self.size = Network.WIDTH * Network.HEIGHT
        if filename==None:
            network_builder = [self.size] + Network.STRUCTURE + [self.size]
            self.model = [np.random.randn(network_builder[i + 1], network_builder[i])\
                                for i in range(len(network_builder) - 1)]
        else:
            self.model = list(np.load(filename, allow_pickle=True))

    def decide_next_move(self, board_instance, training=False):
        board = board_instance.board
        condition = board_instance.status
        preprocessed_board, original_dimension = self.preprocess(board, condition)
        prediction, _ = self.feed_forward(preprocessed_board)

        (i_start, i_end), (j_start, j_end) = original_dimension

        prediction[:i_start * Network.WIDTH] = -1
        prediction[i_end * Network.WIDTH:] = -1
        for i in range(i_start, i_end):
            prediction[i * Network.WIDTH:i * Network.WIDTH + j_start] = -1
            prediction[i * Network.WIDTH + j_end:(i + 1) * Network.WIDTH] = -1
            for j in range(j_start, j_end):
                if board[i - i_start][j - j_start] != Board.EMPTY_SLOT:
                    prediction[i * Network.WIDTH + j] = -1

        prediction_max = np.argmax(prediction)
        i = prediction_max // Network.WIDTH - i_start
        j = prediction_max % Network.WIDTH - j_start
        if not training:
            return (i, j)
        else:
            return (i, j), _, prediction

    def preprocess(self, board, condition):
        width = len(board[0])
        height = len(board)

        preprocessed_board = np.zeros(self.size)
        i_start, j_start = (Network.HEIGHT - height) // 2, (Network.WIDTH - width) // 2
        i_end, j_end = i_start + height, j_start + width
        player_slot = Board.BLACK_SLOT if condition == Board.BLACK_TURN else Board.WHITE_SLOT

        for i in range(i_start, i_end):
            for j in range(j_start, j_end):
                preprocessed_board[i * Network.WIDTH + j] =\
                    Network.PLAYER_SLOT if board[i - i_start][j - j_start] == player_slot else (
                        Network.EMPTY_SLOT if board[i - i_start][j - j_start] == Board.EMPTY_SLOT else
                            Network.ENEMY_SLOT
                    )

        original_dimension = (i_start, i_end), (j_start, j_end)
        return preprocessed_board, original_dimension

    def feed_forward(self, preprocessed_board):
        hidden_states = []
        hidden_states.append(preprocessed_board)
        for layer in self.model[:-1]:
            hidden_states.append(np.dot(layer, hidden_states[-1]))
            hidden_states[-1] = Network.sigmoid(hidden_states[-1])
        prediction = np.dot(self.model[-1], hidden_states[-1])
        prediction = Network.sigmoid(prediction)
        return prediction, hidden_states

    def calculate_gradients(self, hidden_states, gradients):
        full_gradients = [np.empty_like(self.model[i]) for i in range(len(self.model))]
        current_gradients = gradients
        for i in range(len(self.model) - 1, -1, -1):
            for j in range(len(self.model[i])):
                full_gradients[i][j] = current_gradients[j] * hidden_states[i]
            current_gradients = np.dot(current_gradients, self.model[i])
            current_gradients = Network.sigmoid_derivative(current_gradients)
        return full_gradients

    def feed_backward(self, full_gradients, learning_rate):
        for i in range(len(self.model)):
            self.model[i] += learning_rate * full_gradients[i]

    def save_model(self, filename):
        np.save(filename, self.model)

ext: py

reason: import numpy


text:
import numpy as np
import pytest

import psyneulink.core.components.functions.statefulfunctions.integratorfunctions as Functions
import psyneulink.core.llvm as pnlvm
from psyneulink.core.components.functions.statefulfunctions.integratorfunctions import AdaptiveIntegrator
from psyneulink.core.components.functions.function import FunctionError

SIZE=1000
test_var = np.random.rand(SIZE)
test_initializer = np.random.rand(SIZE)
test_noise_arr = np.random.rand(SIZE)

RAND0_1 = np.random.random()
RAND2 = np.random.rand()
RAND3 = np.random.rand()

def SimpleIntFun(init, value, iterations, rate, noise, offset, **kwargs):
    val = np.full_like(value, init)
    for i in range(iterations):
        val = val + (rate * value) + noise + offset
    return val

def AdaptiveIntFun(init, value, iterations, rate, noise, offset, **kwargs):
    val = np.full_like(value, init)
    for i in range(iterations):
        val = (1 - rate) * val + rate * value + noise + offset
    return val

test_data = [
    (Functions.AdaptiveIntegrator, test_var, {'rate':RAND0_1, 'noise':RAND2, 'offset':RAND3}, AdaptiveIntFun),
    (Functions.AdaptiveIntegrator, test_var, {'rate':RAND0_1, 'noise':test_noise_arr, 'offset':RAND3}, AdaptiveIntFun),
    (Functions.AdaptiveIntegrator, test_var, {'initializer':test_initializer, 'rate':RAND0_1, 'noise':RAND2, 'offset':RAND3}, AdaptiveIntFun),
    (Functions.AdaptiveIntegrator, test_var, {'initializer':test_initializer, 'rate':RAND0_1, 'noise':test_noise_arr, 'offset':RAND3}, AdaptiveIntFun),
    (Functions.SimpleIntegrator, test_var, {'rate':RAND0_1, 'noise':RAND2, 'offset':RAND3}, SimpleIntFun),
    (Functions.SimpleIntegrator, test_var, {'rate':RAND0_1, 'noise':test_noise_arr, 'offset':RAND3}, SimpleIntFun),
    (Functions.SimpleIntegrator, test_var, {'initializer':test_initializer, 'rate':RAND0_1, 'noise':RAND2, 'offset':RAND3}, SimpleIntFun),
    (Functions.SimpleIntegrator, test_var, {'initializer':test_initializer, 'rate':RAND0_1, 'noise':test_noise_arr, 'offset':RAND3}, SimpleIntFun),
]

# use list, naming function produces ugly names
names = [
    "AdaptiveIntegrator",
    "AdaptiveIntegrator Noise Array",
    "AdaptiveIntegrator Initializer",
    "AdaptiveIntegrator Initializer Noise Array",
    "SimpleIntegrator",
    "SimpleIntegrator Noise Array",
    "SimpleIntegrator Initializer",
    "SimpleIntegrator Initializer Noise Array",
]

GROUP_PREFIX="IntegratorFunction "

@pytest.mark.function
@pytest.mark.integrator_function
@pytest.mark.parametrize("func, variable, params, expected", test_data, ids=names)
@pytest.mark.benchmark
def test_basic(func, variable, params, expected, benchmark):
    f = func(default_variable=variable, **params)
    benchmark.group = GROUP_PREFIX + func.componentName
    f(variable)
    f(variable)
    res = f(variable)
    assert np.allclose(res, expected(f.initializer, variable, 3, **params))
    benchmark(f, variable)

@pytest.mark.llvm
@pytest.mark.function
@pytest.mark.integrator_function
@pytest.mark.parametrize("func, variable, params, expected", test_data, ids=names)
@pytest.mark.benchmark
def test_llvm(func, variable, params, expected, benchmark):
    benchmark.group = GROUP_PREFIX + func.componentName
    f = func(default_variable=variable, **params)
    m = pnlvm.execution.FuncExecution(f)
    m.execute(variable)
    m.execute(variable)
    res = m.execute(variable)
    assert np.allclose(res, expected(f.initializer, variable, 3, **params))
    benchmark(m.execute, variable)

@pytest.mark.llvm
@pytest.mark.cuda
@pytest.mark.function
@pytest.mark.integrator_function
@pytest.mark.parametrize("func, variable, params, expected", test_data, ids=names)
@pytest.mark.benchmark
def test_ptx_cuda(func, variable, params, expected, benchmark):
    benchmark.group = GROUP_PREFIX + func.componentName
    f = func(default_variable=variable, **params)
    m = pnlvm.execution.FuncExecution(f)
    m.cuda_execute(variable)
    m.cuda_execute(variable)
    res = m.cuda_execute(variable)
    assert np.allclose(res, expected(f.initializer, variable, 3, **params))
    benchmark(m.cuda_execute, variable)

def test_integrator_function_no_default_variable_and_params_len_more_than_1():
    I = AdaptiveIntegrator(rate=[.1, .2, .3])
    I.defaults.variable = np.array([0,0,0])

def test_integrator_function_default_variable_len_1_but_user_specified_and_params_len_more_than_1():
    with pytest.raises(FunctionError) as error_text:
        AdaptiveIntegrator(default_variable=[1], rate=[.1, .2, .3])
    error_msg_a = 'The length (3) of the array specified for the rate parameter'
    error_msg_b = 'must match the length (1) of the default input ([1])'
    assert error_msg_a in str(error_text.value)
    assert error_msg_b in str(error_text.value)

def test_integrator_function_default_variable_and_params_len_more_than_1_error():
    with pytest.raises(FunctionError) as error_text:
        AdaptiveIntegrator(default_variable=[0,0], rate=[.1, .2, .3])
    error_msg_a = 'The length (3) of the array specified for the rate parameter'
    error_msg_b = 'must match the length (2) of the default input ([0 0])'
    assert error_msg_a in str(error_text.value)
    assert error_msg_b in str(error_text.value)

def test_integrator_function_with_params_of_different_lengths():
    with pytest.raises(FunctionError) as error_text:
        AdaptiveIntegrator(rate=[.1, .2, .3], offset=[.4,.5])
    error_msg_a = "The parameters with len>1 specified for AdaptiveIntegrator Function"
    error_msg_b = "(['rate', 'offset']) don't all have the same length"
    assert error_msg_a in str(error_text.value)
    assert error_msg_b in str(error_text.value)

def test_integrator_function_with_default_variable_and_params_of_different_lengths():
    with pytest.raises(FunctionError) as error_text:
        AdaptiveIntegrator(default_variable=[0,0,0], rate=[.1, .2, .3], offset=[.4,.5])
    error_msg_a = "The following parameters with len>1 specified for AdaptiveIntegrator Function"
    error_msg_b = "don't have the same length as its 'default_variable' (3): ['offset']."
    assert error_msg_a in str(error_text.value)
    assert error_msg_b in str(error_text.value)

max_stars_repo_name: AlirezaFarnia/PsyNeuLink

max_issues_repo_name: AlirezaFarnia/PsyNeuLink

max_forks_repo_name: AlirezaFarnia/PsyNeuLink


text:
import json
import numpy
from azureml.core.model import Model
import joblib

def init():
    global model

    # load the model from file into a global object
    model_path = Model.get_model_path(
        model_name="sklearn_regression_model.pkl")
    model = joblib.load(model_path)

def run(raw_data):
    data = json.loads(raw_data)["data"]
    data = numpy.array(data)
    result = model.predict(data)
    return {"result": result.tolist()}

if __name__ == "__main__":
    # Test scoring
    init()
    test_row = '{"data":[[1,2,3,4,5,6,7,8,9,10],[10,9,8,7,6,5,4,3,2,1]]}'
    prediction = run(test_row)
    print("Test result: ", prediction)

max_stars_repo_name: algattik/azure-mlops-terraform

max_issues_repo_name: algattik/azure-mlops-terraform

max_forks_repo_name: algattik/azure-mlops-terraform


text:
""" File where testing references are stores """

import pandas as pd
import numpy as np
import pytest
from Assignment2.helper_functions import CleanData, WarpData, WarpAddress

# Reference Dataframe for CleanData, WarpData testing
df = pd.DataFrame(
    np.array(
        [
            [1, 2, 3, 4, 5],
            [4, np.nan, 6, 24, 26],
            [7, 8, 9, 123, np.nan]
        ]
    ),
    columns=['a', 'b', 'c', 'd', 'e'])

# Testing CleanData class

def test_type_int():
    '''Check that null_count returns integer'''
    null_count = CleanData().null_count(df)
    assert isinstance(null_count, np.int64)

def test_count_null():
    '''Check that null_count returns correct number of nulls'''
    null_count = CleanData().null_count(df)
    assert null_count == 2

# Testing WarpData class
def test_type_int():
    '''Check training and test sets length'''
    train, test = WarpData().train_test_split(df, 0.4)
    assert len(train) < len(df)
    assert len(test) < len(df)
    assert len(train) < len(test)

def test_randomize_len():
    '''Check len of randomize df'''
    df_rand = WarpData().randomize(df, 42)
    assert len(df_rand) == len(df)

# Address series for WarpAddress class testing
addresses = [
    '890 Jennifer Brooks\nNorth Janet, WY 24785',
    '8394 Kim Meadow\nDarrenville, AK 27389',
    '379 Cain Plaza\nJosephburgh, WY 06332',
    '5303 Tina Hill\nAudreychester, VA 97036'
]

# # Import State_Abbr CSV as df
# state_abb = pd.read_csv('State_Abbr.csv')

# state_abb