text:
import re
import unittest

import numpy
from sklearn.utils.validation import check_random_state

from classifiers.utils import split_train_data, iterate

class TestUtils(unittest.TestCase):
    def setUp(self):
        self.random_state = check_random_state(0)
        self.y = numpy.array([0, 0, 0, 1, 1, 1, 1, 1, 0, 0, 2, 2, 2, 2, 2, 1, 1, 1, 1, 0], dtype=numpy.int32)

    def tearDown(self):
        del self.random_state
        del self.y

    def test_split_train_data_positive01(self):
        """ Корректность разделения на обучающее и тестовое множества на корректных данных при заданном рандом-сиде. """
        n_validation = 6
        true_indices_for_training = [18, 1, 19, 8, 10, 17, 6, 13, 4, 2, 5, 14, 9, 7]
        true_indices_for_testing = [16, 11, 3, 0, 15, 12]
        indices_for_training, indices_for_testing = split_train_data(self.y, n_validation, self.random_state)
        self.assertIsInstance(indices_for_training, numpy.ndarray)
        self.assertIsInstance(indices_for_testing, numpy.ndarray)
        self.assertEqual((n_validation,), indices_for_testing.shape)
        self.assertEqual((self.y.shape[0] - n_validation,), indices_for_training.shape)
        self.assertEqual(true_indices_for_training, indices_for_training.tolist())
        self.assertEqual(true_indices_for_testing, indices_for_testing.tolist())

    def test_split_train_data_negative01(self):
        """ Корректность разделения на обучающее и тестовое множества, если n_validation нулевое. """
        n_validation = 0
        true_err_msg = re.escape('Train data cannot be split into train and validation subsets!')
        with self.assertRaisesRegex(ValueError, true_err_msg):
            _ = split_train_data(self.y, n_validation, self.random_state)

    def test_split_train_data_negative02(self):
        """ Корректность разделения на обучающее и тестовое множества, если n_validation равно числу всех примеров. """
        n_validation = self.y.shape[0]
        true_err_msg = re.escape('Train data cannot be split into train and validation subsets!')
        with self.assertRaisesRegex(ValueError, true_err_msg):
            _ = split_train_data(self.y, n_validation, self.random_state)

    def test_split_train_data_negative03(self):
        """ Корректность разделения на обучающее и тестовое множества, если n_validation меньше числа классов. """
        n_validation = 2
        true_err_msg = re.escape('Train data cannot be split into train and validation subsets!')
        with self.assertRaisesRegex(ValueError, true_err_msg):
            _ = split_train_data(self.y, n_validation, self.random_state)

    def test_split_train_data_negative04(self):
        """ Корректность разделения на обучающее и тестовое множества, если n_validation слишком велико. """
        n_validation = self.y.shape[0] - 2
        true_err_msg = re.escape('Train data cannot be split into train and validation subsets!')
        with self.assertRaisesRegex(ValueError, true_err_msg):
            _ = split_train_data(self.y, n_validation, self.random_state)

    def test_iterate_positive01(self):
        """ Входные данные корректны, длина датасета делится на размер батча нацело, итерируемся последовательно. """
        source_indices = numpy.arange(3, self.y.shape[0] + 3, 1, dtype=numpy.int32)
        batch_size = 5
        true_batches = [
            numpy.arange(3 + 0, 3 + 5, 1, dtype=numpy.int32),
            numpy.arange(3 + 5, 3 + 10, 1, dtype=numpy.int32),
            numpy.arange(3 + 10, 3 + 15, 1, dtype=numpy.int32),
            numpy.arange(3 + 15, 3 + 20, 1, dtype=numpy.int32)
        ]
        calculated_batches = list(iterate(source_indices, batchsize=batch_size, shuffle=False,
                                          random_state=self.random_state))
        self.assertEqual(len(true_batches), len(calculated_batches))
        for batch_ind in range(len(true_batches)):
            self.assertIsInstance(calculated_batches[batch_ind], numpy.ndarray)
            self.assertEqual((batch_size,), calculated_batches[batch_ind].shape)
            self.assertEqual(true_batches[batch_ind].tolist(), calculated_batches[batch_ind].tolist())

    def test_iterate_positive02(self):
        """ Входные данные корректны, длина датасета НЕ делится на размер батча нацело, итерируемся последовательно. """
        source_indices = numpy.array([3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22],
                                     dtype=numpy.int32)
        batch_size = 3
        true_batches = [
            numpy.array([3, 4, 5], dtype=numpy.int32),
            numpy.array([6, 7, 8], dtype=numpy.int32),
            numpy.array([9, 10, 11], dtype=numpy.int32),
            numpy.array([12, 13, 14], dtype=numpy.int32),
            numpy.array([15, 16, 17], dtype=numpy.int32),
            numpy.array([18, 19, 20], dtype=numpy.int32),
            numpy.array([21, 22, 3], dtype=numpy.int32)
        ]
        calculated_batches = list(iterate(source_indices, batchsize=batch_size, shuffle=False,
                                          random_state=self.random_state))
        self.assertEqual(len(true_batches), len(calculated_batches))
        for batch_ind in range(len(true_batches)):
            self.assertIsInstance(calculated_batches[batch_ind], numpy.ndarray)
            self.assertEqual((batch_size,), calculated_batches[batch_ind].shape)
            self.assertEqual(true_batches[batch_ind].tolist(), calculated_batches[batch_ind].tolist())

    def test_iterate_positive03(self):
        """ Входные данные корректны, длина датасета делится на размер батча нацело, итерируемся случайно. """
        source_indices = numpy.arange(3, self.y.shape[0] + 3, 1, dtype=numpy.int32)
        batch_size = 5
        true_batches = [
            numpy.array([21, 4, 22, 11, 13], dtype=numpy.int32),
            numpy.array([20, 9, 16, 7, 5], dtype=numpy.int32),
            numpy.array([8, 17, 12, 10, 19], dtype=numpy.int32),
            numpy.array([14, 6, 3, 18, 15], dtype=numpy.int32),
        ]
        calculated_batches = list(iterate(source_indices, batchsize=batch_size, shuffle=True,
                                          random_state=self.random_state))
        self.assertEqual(len(true_batches), len(calculated_batches))
        for batch_ind in range(len(true_batches)):
            self.assertIsInstance(calculated_batches[batch_ind], numpy.ndarray)
            self.assertEqual((batch_size,), calculated_batches[batch_ind].shape)
            self.assertEqual(true_batches[batch_ind].tolist(), calculated_batches[batch_ind].tolist())

    def test_iterate_positive04(self):
        """ Входные данные корректны, длина датасета НЕ делится на размер батча нацело, итерируемся случайно. """
        source_indices = numpy.array([3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22],
                                     dtype=numpy.int32)
        batch_size = 3
        true_batches = [
            numpy.array([21, 4, 22], dtype=numpy.int32),
            numpy.array([11, 13, 20], dtype=numpy.int32),
            numpy.array([9, 16, 7], dtype=numpy.int32),
            numpy.array([5, 8, 17], dtype=numpy.int32),
            numpy.array([12, 10, 19], dtype=numpy.int32),
            numpy.array([14, 6, 3], dtype=numpy.int32),
            numpy.array([18, 15, 21], dtype=numpy.int32)
        ]
        calculated_batches = list(iterate(source_indices, batchsize=batch_size, shuffle=True,
                                          random_state=self.random_state))
        self.assertEqual(len(true_batches), len(calculated_batches))
        for batch_ind in range(len(true_batches)):
            self.assertIsInstance(calculated_batches[batch_ind], numpy.ndarray)
            self.assertEqual((batch_size,), calculated_batches[batch_ind].shape)
            self.assertEqual(true_batches[batch_ind].tolist(), calculated_batches[batch_ind].tolist())

    def test_iterate_negative01(self):
        """ Входные данные некорректны: индексы не являются numpy.ndarray-массивом. """
        source_indices = numpy.arange(3, self.y.shape[0] + 3, 1, dtype=numpy.int32).tolist()
        batch_size = 5
        true_err_msg = re.escape('`indices` must be a numpy.ndarray!')
        with self.assertRaisesRegex(ValueError, true_err_msg):
            _ = list(iterate(source_indices, batchsize=batch_size, shuffle=False, random_state=self.random_state))

    def test_iterate_negative02(self):
        """ Входные данные некорректны: индексы являются numpy.ndarray-массивом, но двумерным, а не одномерным. """
        source_indices = numpy.array(
            [
                [3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22],
                [23, 24, 25, 26, 27, 28, 29, 30, 31, 32, 33, 34, 35, 36, 37, 38, 39, 40, 41, 42]
            ],
            dtype=numpy.int32
        )
        batch_size = 5
        true_err_msg = re.escape('`indices` must be a 1-D array!')
        with self.assertRaisesRegex(ValueError, true_err_msg):
            _ = list(iterate(source_indices, batchsize=batch_size, shuffle=False, random_state=self.random_state))

    def test_iterate_negative03(self):
        """ Входные данные некорректны: индексы заданы 1-мерным numpy.ndarray-массивом, но как вещественные числа. """
        source_indices = numpy.array([3, 4, 5, 6, 7, 8, 9, 10, 11, 12, 13, 14, 15, 16, 17, 18, 19, 20, 21, 22],
                                     dtype=numpy.float32)
        batch_size = 5
        true_err_msg = re.escape('Items of the `indices` array must be integer numbers!')
        with self.assertRaisesRegex(ValueError, true_err_msg):
            _ = list(iterate(source_indices, batchsize=batch_size, shuffle=False, random_state=self.random_state))

if __name__ == '__main__':
    unittest.main(verbosity=2)

max_stars_repo_name: polywaves/soroka

max_issues_repo_name: polywaves/soroka

max_forks_repo_name: polywaves/soroka


text:
import sys
import pandas as pd
from scipy import stats

f = sys.argv[1]
df_tx = pd.read_csv(f,sep='\t')
import pdb;pdb.set_trace()
tx_r,tx_p = stats.pearsonr(df_tx['True'], df_tx['Predict'])
print("{:.3f}\t{:.3e}".format(tx_r,tx_p))

ext: py

max_stars_repo_name: Tsinghua-gongjing/StructureImpute

max_issues_repo_name: Tsinghua-gongjing/StructureImpute

max_forks_repo_name: Tsinghua-gongjing/StructureImpute

reason: from scipy


text:
"""Implement SVM that predicts genre from an audio file."""

import pickle
from typing import Optional, List, Union, Tuple
import numpy as np
import pandas as pd
import librosa
import audioread

def svm_prediction(filename: str, genre_df: pd.DataFrame,
                   model_filename: str =
                   "data/svm_model.pkl") -> Optional[str]:
    """
    Return SVM prediction of genre from an audio file.

    Args:
       filename (str): Path to music mp3 file
       genre_df (pd.DataFrame): Dataframe containing genre information.
       model_filename (str): Path to the model file.
       Default is "data/svm_model.pkl".
    Returns:
        genre (str): Predicted genre of the mp3 file.
    """
    features = []

    try:
        y_audio, sample_rate = librosa.load(filename, sr=None, mono=True)
    except (RuntimeError, audioread.NoBackendError):
        print("Failed to load ", filename)
        return None
    except FileNotFoundError:
        print("File not found", filename)
        return None

    mel = librosa.feature.mfcc(y=y_audio, sr=sample_rate)
    mel = (mel - np.min(mel)) / (np.max(mel) - np.min(mel))

    spectral_contrast = librosa.feature.spectral_contrast(
        y_audio, sr=sample_rate, n_fft=2048, hop_length=1024
    )

    features.append((mel, spectral_contrast))

    test_data_array = np.array(
        [
            np.concatenate(
                [
                    np.average(features[0][0], axis=1),
                    np.median(features[0][0], axis=1),
                    np.std(features[0][0], axis=1),
                    [np.average(features[0][1])],
                    [np.median(features[0][1])],
                    [np.std(features[0][1])],
                ]
            )
        ]
    )

    with open(model_filename, "rb") as file:
        clf = pickle.load(file)

    label = clf.predict(test_data_array)

    genre = genre_df.iloc[label]["title"].item()

    return genre

def svm_accuracy_report(true_labels: Union[List[int], np.ndarray],
                        pred_labels: Union[List[int], np.ndarray]
                        ) -> Tuple[List[float], List[float], float]:
    """
    Return accuracy report of trained SVM.

    Args:
        true_labels: (Union[List[int], np.ndarray]) True labels for genre.
        pred_labels: (Union[List[int], np.ndarray]) Predicted labels for genre.

    Outputs:
        tp_rate_list: (List[float]) List of true positive rate of
        predictions per genre.
        fp_rate_list: (List[float]) List of false positive rate of
        predictions per genre.
        accuracy: (float) Overall accuracy of predicted labels
        against true labels.
    """

    assert true_labels is not None, "True labels cannot be a none type object."
    assert pred_labels is not None, "Pred labels cannot be a none type object."

    assert len(true_labels) == len(pred_labels), \
        "True and predicted labels must be the same length. "

    if isinstance(true_labels, list):
        true_labels = np.array(true_labels)
    if isinstance(pred_labels, list):
        pred_labels = np.array(pred_labels)

    all_ind = np.arange(len(true_labels))
    ind = np.where(pred_labels == true_labels)[0]
    accuracy = (len(ind) / len(pred_labels)) * 100
    tp_rate_list = []
    fp_rate_list = []

    print(len(ind), ' test files of a total of ', len(pred_labels),
          'are predicted correctly for an accuracy of ',
          accuracy, '%\n\n')

    for i in np.unique(true_labels):
        true_ind = np.where(true_labels == i)[0]
        true_neg = np.setdiff1d(all_ind, true_ind)

        pred_ind = np.where(pred_labels == i)[0]
        pred_neg = np.setdiff1d(all_ind, pred_ind)

        percentage_ind = (len(
            np.intersect1d(true_ind, pred_ind)) / len(true_ind)) * 100

        false_positive = len(np.setdiff1d(pred_ind, true_ind))
        true_positive = len(np.intersect1d(true_ind, pred_ind))
        true_negative = len(np.intersect1d(true_neg, pred_neg))
        false_negative = len(np.setdiff1d(pred_neg, true_neg))

        fp_rate = false_positive / (false_positive + true_negative)
        tp_rate = true_positive / (false_negative + true_positive)

        tp_rate_list.append(tp_rate)
        fp_rate_list.append(fp_rate)

        print('True Positive Rate: %.3f False Positive Rate: %.3f '
              'Percent Correct for '
              'genre %s: %.3f' % (tp_rate, fp_rate, i, percentage_ind))

    return tp_rate_list, fp_rate_list, accuracy

reason: import numpy


text:
import matplotlib
import mxnet as mx
import nltk
import numpy as np
import rpy2
import sklearn
import sparknlp
import tensorflow as tf
import tensorflow_datasets
import tensorflow_estimator
import tensorflow_hub
import tensorflow_probability
import torch
import torchvision
import xgboost

import os
if os.getenv("DATAPROC_VERSION") >= "2.0":
  import spark_tensorflow_distributor


text:
# -*- coding: utf-8 -*-
"""
Created on Wed Nov 27 19:29:25 2019

@author: yanik
"""
import cv2
import pyautogui
import time
import numpy as np
from part1 import sobel_filter

def min_eigenvalue_corner_detector(image):

    row, col, ch = image.shape

    #Apply sobel filter to take derivatives
    sobelx, sobely = sobel_filter(image)

    #To be able to get image structure tensor(2x2), convolve each values with Gaussian kernel
    M_00 = cv2.GaussianBlur(sobelx*sobelx,(5,5),0)
    M_01 = cv2.GaussianBlur(sobelx*sobely,(5,5),0)
    M_11 = cv2.GaussianBlur(sobely*sobely,(5,5),0)

    #minimum eigen value calculation (I got a reference from our lecture slides)
    min_eigen_values = 0.5 * ((M_00 + M_11) - ((M_00 - M_11)**2 + 4*(M_01**2))**0.5)
    #min_eigen_values = cv2.dilate(min_eigen_values,None)

    image = cv2.cvtColor(image,cv2.COLOR_BGR2GRAY)
    image = cv2.cvtColor(image,cv2.COLOR_GRAY2BGR)

    #I applied a threshold to take sharp corners and changed the colors
    image[min_eigen_values>0.15*min_eigen_values.max()]=[0,0,255]

    #Holds corners coordinates
    corner_x = []
    corner_y = []

    #To be able to see corners clearly, I traversed pixels, found red pixels, took coordinates and drew rectangles
    for i in range(row):
        for j in range(col):
            if (image[i][j][0] == 0 and image[i][j][1] == 0 and image[i][j][2] == 255):
                corner_x.append(j)
                corner_y.append(i)
                image = cv2.rectangle(image, (j, i), (j + 5, i + 5), (0, 255, 0), -1)

    '''
    #You can open comments to show and save output
    cv2.imshow('Output', image)
    #cv2.imwrite("Part3.jpg", image)
    cv2.waitKey()
    '''
    return corner_x, corner_y

'''
#This part is for testing
time.sleep(5)

myScreenshot = pyautogui.screenshot()
myScreenshot.save('test.png')

image = cv2.imread("./test.png")
row, col, ch = image.shape

#This part crops desired area from the whole screenshot
upLeft = (int(col / 4), int(row / 4.5))
bottomRight = (int(col * 3 / 4), int(row * 3 / 4))

image = image[upLeft[1]:bottomRight[1], upLeft[0]:bottomRight[0]]

corner_x, corner_y = min_eigenvalue_corner_detector(image)
'''

max_stars_repo_path: Sobel Filter, Edge Detector and Corner Detector/Code/part3.py

max_stars_repo_name: cansuynk/ComputerVision

max_issues_repo_name: cansuynk/ComputerVision

max_forks_repo_name: cansuynk/ComputerVision


text:
import cv2
import numpy as np
import os
import sys

this_dir = os.path.dirname(os.path.abspath(__file__))
needed_dir = os.path.abspath(os.path.join(this_dir, '../.'))
sys.path.insert(0, needed_dir)

from helpers.detect_people import Detect_people

# https://www.pyimagesearch.com/2015/11/09/pedestrian-detection-opencv/
# https://www.pyimagesearch.com/2014/11/10/histogram-oriented-gradients-object-detection/
# https://github.com/alexander-hamme/Pedestrian_Detector

detector = Detect_people()
currentDirectory = os.path.dirname(os.path.abspath(__file__))
img = cv2.imread(os.path.join(currentDirectory, "..", "images", "1_original.jpg"))

detector.DRAW_RAW_RECT = True
detector.DRAW_RECT = True
detector.SHOW_IMAGES = True
detector.set_calibration(2)
image, _, _ = detector.find_people(img)
detector.draw_image(image)

# detector.try_all_calibration_modes(img)


text:
import numpy as np
import math
import matplotlib.pyplot as plt

if __package__ is None or __package__ == '':
    import interpolation
else:
    from . import interpolation

l0 = 0.045
l1 = 0.065
l2 = 0.087

def sandbox(t):
    """
    python simulator.py -m sandbox

    Un premier bac à sable pour faire des expériences

    La fonction reçoit le temps écoulé depuis le début (t) et retourne une position cible
    pour les angles des 12 moteurs

    - Entrée: t, le temps (secondes écoulées depuis le début)
    - Sortie: un tableau contenant les 12 positions angulaires cibles (radian) pour les moteurs
    """

    # Par exemple, on envoie un mouvement sinusoidal
    targets = [0] * 12
    targets[0] = np.sin(t)

    return targets

def direct(a, b, c):
    """
    python simulator.py -m direct

    Le robot est figé en l'air, on ne contrôle qu'une patte

    Reçoit en argument les angles des moteurs (a, b, c) et retourne la position du bout
    de la patte (x, y, z)

    - Sliders: angles des trois moteurs de la patte
    - Entrée: a, b, c, angles des trois moteurs
    - Sortie: x, y, z, la position du bout de la patte
    """

    T0 = a  # Theta 0
    T1 = np.pi + b  # Theta 1
    T2 = c + np.pi  # Theta 2

    # point O
    O = np.array([[0], [0], [0]])
    # print("O:", O)

    # point A
    A = np.array([[l0 * math.cos(T0)], [l0 * math.sin(T0)], [0]]) + O

    # print("A:", A)

    # point B
    def rotationZ(t):
        return np.array([[np.cos(t), -np.sin(t), 0], [np.sin(t), np.cos(t), 0], [0, 0, 1]])

    def rotationX(t):
        return np.array([[1, 0, 0], [0, np.cos(t), -np.sin(t)], [0, np.sin(t), np.cos(t)]])

    B = np.dot(rotationZ(T0), np.array([[l1 * np.cos(T1 - np.pi)], [0], [l1 * np.sin(T1 - np.pi)]])) + A

    # print("B:", B)

    # point C
    def rotationY(t):
        return np.array([[np.cos(t), 0, np.sin(t)], [0, 1, 0], [-np.sin(t), 0, np.cos(t)]])

    # point M
    M = np.dot(rotationZ(T0),
               np.dot(rotationY(np.pi - T1), np.array([[l2 * np.cos(np.pi - T2)], [0], [l2 * np.sin(np.pi - T2)]]))) + B
    # print("M:", M)

    # points = [O, A, B, M]

    # X, Y, Z = np.zeros(4), np.zeros(4), np.zeros(4)
    # for i in range(len(points)):
    #     X[i] = points[i][0]
    #     Y[i] = points[i][1]
    #     Z[i] = points[i][2]

    return [M[0], M[1], M[2]]

def alKashi(a, b, c):
    angle = (a ** 2 + b ** 2 - c ** 2) / (2 * a * b)
    if angle > 1:
        angle = 1
    elif angle < -1:
        angle = -1
    return math.acos(angle)

def inverse(x, y, z):
    """
    python simulator.py -m inverse

    Le robot est figé en l'air, on ne contrôle qu'une patte

    Reçoit en argument une position cible (x, y, z) pour le bout de la patte, et produit les angles
    (alpha, beta, gamma) pour que la patte atteigne cet objectif

    - Sliders: La position cible x, y, z du bout de la patte
    - Entrée: x, y, z, une position cible dans le repère de la patte (mètres), provenant du slider
    - Sortie: un tableau contenant les 3 positions angulaires cibles (en radians)
    """

    side = 1

    theta0 = math.atan2(y, x)
    AH = math.sqrt(x ** 2 + y ** 2) - l0
    AM = math.sqrt(AH ** 2 + z ** 2)
    theta2 = side * (np.pi - alKashi(l1, l2, AM))

    if (AM == 0):
        theta1 = 0  # Peu importe
    else:
        theta1 = side * alKashi(AM, l1, l2) + math.atan2(z, AH)
    return [theta0, theta1, theta2]

def draw(t):
    interpolator = interpolation.LinearSpline3D()
    interpolator.add_entry(0, 0.05, 0, 0)
    interpolator.add_entry(1, 0.05, 0, 0.1)
    interpolator.add_entry(2, 0.05, 0.1, 0.1)
    interpolator.add_entry(3, 0.05, 0, 0)
    x, y, z = interpolator.interpolate(t % 3)

    """
    python simulator.py -m draw

    Le robot est figé en l'air, on ne contrôle qu'une patte

    Le but est, à partir du temps donné, de suivre une trajectoire de triangle. Pour ce faire, on
    utilisera une interpolation linéaire entre trois points, et la fonction inverse précédente.

    - Entrée: t, le temps (secondes écoulées depuis le début)
    - Sortie: un tableau contenant les 3 positions angulaires cibles (en radians)
    """

    return inverse(x, y, z)

def step(t, patte_num, speed_x, speed_y):
    d = 0.120
    z0 = -0.05

    initiale = [(-d, d, z0), (-d, -d, z0), (d, -d, z0), (d, d, z0)]
    start = (initiale[patte_num][0] + speed_x / 2, initiale[patte_num][1] + speed_y / 2, initiale[patte_num][2])
    high = (initiale[patte_num][0] + speed_x / 2, initiale[patte_num][1] + speed_y / 2, initiale[patte_num][2] + 0.05)
    end = (initiale[patte_num][0] - speed_x / 2, initiale[patte_num][1] - speed_y / 2, initiale[patte_num][2])

    interpolator = interpolation.LinearSpline3D()

    interpolator.add_entry(0.0, * start)
    interpolator.add_entry(0.5, * end)
    interpolator.add_entry(0.75, * high)
    interpolator.add_entry(1.0, * start)
    x, y, z = interpolator.interpolate(t % 1.0)

    print(t)

    return x, y, z

def legs(leg1, leg2, leg3, leg4):
    """
    python simulator.py -m legs

    Le robot est figé en l'air, on contrôle toute les pattes

    - Sliders: les 12 coordonnées (x, y, z) du bout des 4 pattes
    - Entrée: des positions cibles (tuples (x, y, z)) pour le bout des 4 pattes
    - Sortie: un tableau contenant les 12 positions angulaires cibles (radian) pour les moteurs
    """
    targets = [0] * 12
    offset = math.sqrt(0.040 * 0.040 / 2)

    def rotation(t, x, y):
        return np.dot(np.array([[np.cos(t), -np.sin(t)], [np.sin(t), np.cos(t)]]), np.array([[x], [y]]))

    def inverse_tuple_rotation_translation(my_tuple, t, x, y):
        x_ = my_tuple[0]
        y_ = my_tuple[1]
        z_ = my_tuple[2]

        # translation
        x_ += x * offset
        y_ += y * offset

        # rotation
        [x_, y_] = rotation(t, x_, y_)

        return inverse(x_, y_, z_)

    # Patte 1 +x,+y
    temp_x, temp_y, temp_z = inverse_tuple_rotation_translation(leg1, - 3 * np.pi / 4, 1, -1)
    targets[0:2] = [temp_x, temp_y, temp_z]
    # Patte 2 +x,+y
    temp_x, temp_y, temp_z = inverse_tuple_rotation_translation(leg2, 3 * np.pi / 4, 1, 1)
    targets[3:5] = [temp_x, temp_y, temp_z]
    # Patte 3 +x,+y
    temp_x, temp_y, temp_z = inverse_tuple_rotation_translation(leg3, np.pi / 4, -1, 1)
    targets[6:8] = [temp_x, temp_y, temp_z]
    # Patte 4 +x,+y
    temp_x, temp_y, temp_z = inverse_tuple_rotation_translation(leg4, -np.pi / 4, -1, -1)
    targets[9:] = [temp_x, temp_y, temp_z]

    return targets

leg1 = []

def walk_2(t, speed_x, speed_y, speed_rotation):

    # Déterminer la position des pattes en fonction de t
    # print("avancer_x = {0}\navancer_y = {1}".format(speed_x, speed_y))
    patte1 = step(t + 0.5, 0, speed_x, speed_y)
    patte2 = step(t, 1, speed_x, speed_y)
    patte3 = step(t + 0.5, 2, speed_x, speed_y)
    patte4 = step(t, 3, speed_x, speed_y)

    targets = legs(patte1, patte2, patte3, patte4)

    if False:
        x1, y1, z1 = patte3
        leg1.append([x1, y1, z1])
        tmp = np.array(leg1)
        plt.clf()
        plt.plot(tmp.T[0], color='red')
        plt.plot(tmp.T[1], color='green')
        plt.plot(tmp.T[2], color='blue')
        plt.grid()
        plt.pause(0.01)

    return targets

def walk(t, speed_x, speed_y, speed_rotation):
    return walk_2(t, speed_x, speed_y, speed_rotation)
    d = 0.120
    z0 = -0.05
    """
    python simulator.py -m walk

    Le but est d'intégrer tout ce que nous avons vu ici pour faire marcher le robot

    - Sliders: speed_x, speed_y, speed_rotation, la vitesse cible du robot
    - Entrée: t, le temps (secondes écoulées depuis le début)
            speed_x, speed_y, et speed_rotation, vitesses cibles contrôlées par les sliders
    - Sortie: un tableau contenant les 12 positions angulaires cibles (radian) pour les moteurs
    """
    targets = [0] * 12
    # Position initiale
    initiale = [(-d, d, z0), (-d, -d, z0), (d, -d, z0), (d, d, z0)]

    if t == 0:
        return legs(initiale[0], initiale[1], initiale[2], initiale[3])

    # Déterminer la position des pattes en fonction de t
    avancer_x  = speed_x
    avancer_y = speed_y
    print("avancer_x = {0}\n avancer_y = {1}".format(avancer_x, avancer_y))
    avancer_z = 0
    if speed_x != 0 or speed_y != 0:
        avancer_z = 0.03
    step = int((t * 15) % 4)
    # print("t : {0}".format(t))

    if step == 0:
        print("step {0}".format(step))
        patte1 = (initiale[0][0] - avancer_x, initiale[0][1] - avancer_y, initiale[0][2] + 0)
        patte2 = (initiale[1][0] + avancer_x / 2, initiale[1][1] + avancer_y / 2, initiale[1][2])
        patte3 = (initiale[2][0] - avancer_x, initiale[2][1] - avancer_y, initiale[2][2] + 0)
        patte4 = (initiale[3][0] + avancer_x / 2, initiale[3][1] + avancer_y / 2, initiale[3][2])

    elif step == 1:
        print("step {0}".format(step))
        patte1 = (initiale[0][0] + avancer_x / 2, initiale[0][1] + avancer_y / 2, initiale[0][2] + avancer_z)
        patte2 = initiale[1]
        patte3 = (initiale[2][0] + avancer_x / 2, initiale[2][1] + avancer_y / 2, initiale[2][2] + avancer_z)
        patte4 = initiale[3]

    elif step == 2:
        # print("step {0}".format(step))
        patte1 = (initiale[0][0] + avancer_x / 2, initiale[0][1] + avancer_y / 2, initiale[0][2])
        patte2 = (initiale[1][0] - avancer_x, initiale[1][1] - avancer_y, initiale[1][2])
        patte3 = (initiale[2][0] + avancer_x / 2, initiale[2][1] + avancer_y / 2, initiale[2][2])
        patte4 = (initiale[3][0] - avancer_x, initiale[3][1] - avancer_y, initiale[3][2])

    else:  # step == 3
        # print("step {0}".format(step))
        patte1 = initiale[0]
        patte2 = (initiale[1][0] + avancer_x / 2, initiale[1][1] + avancer_y / 2, initiale[1][2] + avancer_z)
        patte3 = initiale[2]
        patte4 = (initiale[3][0] + avancer_x / 2, initiale[3][1] + avancer_y / 2, initiale[3][2] + avancer_z)

    if False:
        x1, y1, z1 = patte1
        leg1.append([x1, y1, z1])
        tmp = np.array(leg1)
        plt.clf()
        plt.plot(tmp.T[0], color = 'red')
        plt.plot(tmp.T[1], color='green')
        plt.plot(tmp.T[2], color='blue')
        plt.grid()
        plt.pause(0.05)

    targets = legs(patte1, patte2, patte3, patte4)
    # print('\np1:', patte1, '\np2:', patte2, '\np3:', patte3, '\np4:', patte4)
    return targets

if __name__ == "__main__":
    print("N'exécutez pas ce fichier, mais simulator.py")

max_stars_repo_name: Sdelpeuch/IntroductionRobotique

max_issues_repo_name: Sdelpeuch/IntroductionRobotique

max_forks_repo_name: Sdelpeuch/IntroductionRobotique


text:
import numpy as np
import sys
from scipy.stats import norm

import regreg.api as rr

from .credible_intervals import projected_langevin
from .lasso_reduced import nonnegative_softmax_scaled, neg_log_cube_probability

class selection_probability_ms(rr.smooth_atom):
    def __init__(self,
                 X,
                 feasible_point,
                 active,  # the active set chosen by randomized marginal screening
                 active_signs,  # the set of signs of active coordinates chosen by ms
                 threshold,  # in R^p
                 mean_parameter,
                 noise_variance,
                 randomizer,
                 coef=1.,
                 offset=None,
                 quadratic=None,
                 nstep=10):

        n, p = X.shape
        self._X = X

        E = active.sum()
        self.q = p - E
        sigma = np.sqrt(noise_variance)

        self.active = active

        self.noise_variance = noise_variance
        self.randomization = randomizer
        self.inactive_conjugate = self.active_conjugate = randomizer.CGF_conjugate
        if self.active_conjugate is None:
            raise ValueError(
                'randomization must know its CGF_conjugate -- currently only isotropic_gaussian and laplace are implemented and are assumed to be randomization with IID coordinates')

        initial = np.zeros(n + E, )
        initial[n:] = feasible_point
        self.n = n

        rr.smooth_atom.__init__(self,
                                (n + E,),
                                offset=offset,
                                quadratic=quadratic,
                                initial=initial,
                                coef=coef)

        self.coefs[:] = initial
        nonnegative = nonnegative_softmax_scaled(E)

        opt_vars = np.zeros(n + E, bool)
        opt_vars[n:] = 1

        self._opt_selector = rr.selector(opt_vars, (n + E,))
        self.nonnegative_barrier = nonnegative.linear(self._opt_selector)
        self._response_selector = rr.selector(~opt_vars, (n + E,))

        self.set_parameter(mean_parameter, noise_variance)

        self.A_active = np.hstack([np.true_divide(-X[:, active].T, sigma), np.identity(E) * active_signs[None, :]])

        self.A_inactive = np.hstack([np.true_divide(-X[:, ~active].T, sigma), np.zeros((p - E, E))])

        self.offset_active = active_signs * threshold[active]
        self.offset_inactive = np.zeros(p - E)

        self.active_conj_loss = rr.affine_smooth(self.active_conjugate,
                                                 rr.affine_transform(self.A_active, self.offset_active))

        cube_obj = neg_log_cube_probability(self.q, threshold[~active], randomization_scale=1.)

        self.cube_loss = rr.affine_smooth(cube_obj, rr.affine_transform(self.A_inactive, self.offset_inactive))

        self.total_loss = rr.smooth_sum([self.active_conj_loss,
                                         self.cube_loss,
                                         self.likelihood_loss,
                                         self.nonnegative_barrier])

    def set_parameter(self, mean_parameter, noise_variance):
        """
        Set $\beta_E^*$.
        """
        mean_parameter = np.squeeze(mean_parameter)
        likelihood_loss = rr.signal_approximator(mean_parameter, coef=1. / noise_variance)
        self.likelihood_loss = rr.affine_smooth(likelihood_loss, self._response_selector)

    def smooth_objective(self, param, mode='both', check_feasibility=False):
        """
        Evaluate the smooth objective, computing its value, gradient or both.
        Parameters
        ----------
        mean_param : ndarray
            The current parameter values.
        mode : str
            One of ['func', 'grad', 'both'].
        check_feasibility : bool
            If True, return `np.inf` when
            point is not feasible, i.e. when `mean_param` is not
            in the domain.
        Returns
        -------
        If `mode` is 'func' returns just the objective value
        at `mean_param`, else if `mode` is 'grad' returns the gradient
        else returns both.
        """

        param = self.apply_offset(param)

        if mode == 'func':
            f = self.total_loss.smooth_objective(param, 'func')
            return self.scale(f)
        elif mode == 'grad':
            g = self.total_loss.smooth_objective(param, 'grad')
            return self.scale(g)
        elif mode == 'both':
            f = self.total_loss.smooth_objective(param, 'func')
            g = self.total_loss.smooth_objective(param, 'grad')
            return self.scale(f), self.scale(g)
        else:
            raise ValueError("mode incorrectly specified")

    def minimize2(self, step=1, nstep=100, tol=1.e-8):

        n, p = self._X.shape

        current = self.coefs
        current_value = np.inf

        objective = lambda u: self.smooth_objective(u, 'func')
        grad = lambda u: self.smooth_objective(u, 'grad')

        for itercount in range(nstep):
            newton_step = grad(current) * self.noise_variance

            # make sure proposal is feasible

            count = 0
            while True:
                count += 1
                proposal = current - step * newton_step
                if np.all(proposal[n:] > 0):
                    break
                step *= 0.5
                if count >= 40:
                    raise ValueError('not finding a feasible point')

            # make sure proposal is a descent

            count = 0
            while True:
                proposal = current - step * newton_step
                proposed_value = objective(proposal)
                # print(current_value, proposed_value, 'minimize')
                if proposed_value <= current_value:
                    break
                step *= 0.5

            # stop if relative decrease is small

            if np.fabs(current_value - proposed_value) < tol * np.fabs(current_value):
                current = proposal
                current_value = proposed_value
                break

            current = proposal
            current_value = proposed_value

            if itercount % 4 == 0:
                step *= 2

        # print('iter', itercount)
        value = objective(current)
        return current, value

class sel_prob_gradient_map_ms(rr.smooth_atom):
    def __init__(self,
                 X,
                 feasible_point,  # in R^{ |E|}
                 active,
                 active_signs,
                 threshold,  # in R^p
                 generative_X,  # in R^{p}\times R^{n}
                 noise_variance,
                 randomizer,
                 coef=1.,
                 offset=None,
                 quadratic=None):

        self.E = active.sum()
        self.n, self.p = X.shape
        self.dim = generative_X.shape[1]

        self.noise_variance = noise_variance

        (self.X, self.feasible_point, self.active, self.active_signs, self.threshold, self.generative_X, self.noise_variance,
         self.randomizer) = (X, feasible_point, active, active_signs, threshold, generative_X, noise_variance, randomizer)

        rr.smooth_atom.__init__(self,
                                (self.dim,),
                                offset=offset,
                                quadratic=quadratic,
                                coef=coef)

    def smooth_objective(self, true_param, mode='both', check_feasibility=False, tol=1.e-8):
        true_param = self.apply_offset(true_param)

        mean_parameter = np.squeeze(self.generative_X.dot(true_param))

        primal_sol = selection_probability_ms(self.X,
                                              self.feasible_point,
                                              self.active,
                                              self.active_signs,
                                              self.threshold,
                                              mean_parameter,
                                              self.noise_variance,
                                              self.randomizer)

        sel_prob_primal = primal_sol.minimize2(nstep=100)[::-1]
        optimal_primal = (sel_prob_primal[1])[:self.n]
        sel_prob_val = -sel_prob_primal[0]
        optimizer = self.generative_X.T.dot(np.true_divide(optimal_primal - mean_parameter, self.noise_variance))

        if mode == 'func':
            return sel_prob_val
        elif mode == 'grad':
            return optimizer
        elif mode == 'both':
            return sel_prob_val, optimizer
        else:
            raise ValueError('mode incorrectly specified')

class selective_inf_ms(rr.smooth_atom):
    def __init__(self,
                 y,
                 grad_map,
                 prior_variance,
                 coef=1.,
                 offset=None,
                 quadratic=None,
                 nstep=10):

        generative_X = grad_map.generative_X
        self.param_shape = generative_X.shape[1]

        y = np.squeeze(y)

        self.E = grad_map.E

        self.generative_X = grad_map.generative_X

        initial = np.zeros(self.E)

        rr.smooth_atom.__init__(self,
                                (self.param_shape,),
                                offset=offset,
                                quadratic=quadratic,
                                initial=initial,
                                coef=coef)

        self.coefs[:] = initial

        noise_variance = grad_map.noise_variance

        self.set_likelihood(y, noise_variance, generative_X)

        self.set_prior(prior_variance)

        self.initial_state = initial

        self.total_loss_0 = rr.smooth_sum([self.likelihood_loss,
                                           self.log_prior_loss,
                                           grad_map])

    def set_likelihood(self, y, noise_variance, generative_X):
        likelihood_loss = rr.signal_approximator(y, coef=1. / noise_variance)
        self.likelihood_loss = rr.affine_smooth(likelihood_loss, generative_X)

    def set_prior(self, prior_variance):
        self.log_prior_loss = rr.signal_approximator(np.zeros(self.param_shape), coef=1. / prior_variance)

    def smooth_objective(self, true_param, mode='both', check_feasibility=False):

        true_param = self.apply_offset(true_param)

        if mode == 'func':
            f = self.total_loss_0.smooth_objective(true_param, 'func')
            return self.scale(f)
        elif mode == 'grad':
            g = self.total_loss_0.smooth_objective(true_param, 'grad')
            return self.scale(g)
        elif mode == 'both':
            f, g = self.total_loss_0.smooth_objective(true_param, 'both')
            return self.scale(f), self.scale(g)
        else:
            raise ValueError("mode incorrectly specified")

    def map_solve(self, step=1, nstep=100, tol=1.e-8):

        current = self.coefs[:]
        current_value = np.inf

        objective = lambda u: self.smooth_objective(u, 'func')
        grad = lambda u: self.smooth_objective(u, 'grad')

        for itercount in range(nstep):

            newton_step = grad(current)
            # * self.noise_variance

            # make sure proposal is a descent
            count = 0
            while True:
                proposal = current - step * newton_step
                proposed_value = objective(proposal)

                if proposed_value <= current_value:
                    break
                step *= 0.5

            # stop if relative decrease is small

            if np.fabs(current_value - proposed_value) < tol * np.fabs(current_value):
                current = proposal
                current_value = proposed_value
                break

            current = proposal
            current_value = proposed_value

            if itercount % 4 == 0:
                step *= 2

        value = objective(current)
        return current, value

    def posterior_samples(self, langevin_steps=1500, burnin=50):
        state = self.initial_state
        sys.stderr.write("Number of selected variables by marginal screening: "+str(state.shape)+"\n")
        gradient_map = lambda x: -self.smooth_objective(x, 'grad')
        projection_map = lambda x: x
        stepsize = 1. / self.E
        sampler = projected_langevin(state, gradient_map, projection_map, stepsize)

        samples = []

        for i in range(langevin_steps):
            sampler.next()
            samples.append(sampler.state.copy())
            #print i, sampler.state.copy()
            sys.stderr.write("sample number: " + str(i)+"\n")

        samples = np.array(samples)
        return samples[burnin:, :]

    def posterior_risk(self, estimator_1, estimator_2, langevin_steps=2000, burnin=0):
        state = self.initial_state
        sys.stderr.write("Number of selected variables by randomized lasso: "+str(state.shape)+"\n")
        gradient_map = lambda x: -self.smooth_objective(x, 'grad')
        projection_map = lambda x: x
        stepsize = 1. / self.E
        sampler = projected_langevin(state, gradient_map, projection_map, stepsize)

        post_risk_1 = 0.
        post_risk_2 = 0.

        for i in range(langevin_steps):
            sampler.next()
            sample = sampler.state.copy()

            #print(sample)
            risk_1 = ((estimator_1-sample)**2).sum()
            print("adjusted risk", risk_1)
            post_risk_1 += risk_1

            risk_2 = ((estimator_2-sample) ** 2).sum()
            print("unadjusted risk", risk_2)
            post_risk_2 += risk_2

        return post_risk_1/langevin_steps, post_risk_2/langevin_steps

reason: import numpy,from scipy


text:
#!/usr/bin/env python
from collections import OrderedDict, defaultdict
from sense2vec import Sense2Vec
from sense2vec.util import split_key, cosine_similarity
from pathlib import Path
from wasabi import msg
import numpy
import typer

def main(
    # fmt: off
    in_file: str = typer.Argument(..., help="Vectors file (text-based)"),
    vocab_file: str = typer.Argument(..., help="Vocabulary file"),
    out_dir: str = typer.Argument(..., help="Path to output directory"),
    min_freq_ratio: float = typer.Option(0.0, "--min-freq-ratio", "-r", help="Frequency ratio threshold for discarding minority senses or casings"),
    min_distance: float = typer.Option(0.0, "--min-distance", "-s", help="Similarity threshold for discarding redundant keys"),
    # fmt: on
):
    """
    Step 5: Export a sense2vec component

    Expects a vectors.txt and a vocab file trained with GloVe and exports
    a component that can be loaded with Sense2vec.from_disk.
    """
    input_path = Path(in_file)
    vocab_path = Path(vocab_file)
    output_path = Path(out_dir)
    if not input_path.exists():
        msg.fail("Can't find input file", in_file, exits=1)
    if input_path.suffix == ".bin":
        msg.fail("Need text-based vectors file, not binary", in_file, exits=1)
    if not vocab_path.exists():
        msg.fail("Can't find vocab file", vocab_file, exits=1)
    if not output_path.exists():
        output_path.mkdir(parents=True)
        msg.good(f"Created output directory {out_dir}")
    with input_path.open("r", encoding="utf8") as f:
        (n_vectors, vector_size), f = _get_shape(f)
        vectors_data = f.readlines()
    with vocab_path.open("r", encoding="utf8") as f:
        vocab = read_vocab(f)
    vectors = {}
    all_senses = set()
    for item in vectors_data:
        item = item.rstrip().rsplit(" ", vector_size)
        key = item[0]
        try:
            _, sense = split_key(key)
        except ValueError:
            continue
        vec = item[1:]
        if len(vec) != vector_size:
            msg.fail(f"Wrong vector size: {len(vec)} (expected {vector_size})", exits=1)
        all_senses.add(sense)
        vectors[key] = numpy.asarray(vec, dtype=numpy.float32)
    discarded = set()
    discarded.update(get_minority_keys(vocab, min_freq_ratio))
    discarded.update(get_redundant_keys(vocab, vectors, min_distance))
    n_vectors = len(vectors) - len(discarded)
    s2v = Sense2Vec(shape=(n_vectors, vector_size), senses=list(all_senses))
    for key, vector in vectors.items():
        if key not in discarded:
            s2v.add(key, vector)
            s2v.set_freq(key, vocab[key])
    msg.good("Created the sense2vec model")
    msg.info(f"{n_vectors} vectors, {len(all_senses)} total senses")
    s2v.to_disk(output_path)
    msg.good("Saved model to directory", out_dir)

def _get_shape(file_):
    """Return a tuple with (number of entries, vector dimensions). Handle
    both word2vec/FastText format, which has a header with this, or GloVe's
    format, which doesn't."""
    first_line = next(file_).replace("\ufeff", "").split()
    if len(first_line) == 2:
        return tuple(int(size) for size in first_line), file_
    count = 1
    for line in file_:
        count += 1
    file_.seek(0)
    shape = (count, len(first_line) - 1)
    return shape, file_

def read_vocab(vocab_file):
    freqs = OrderedDict()
    for line in vocab_file:
        item = line.rstrip()
        if item.endswith(" word"):  # for fastText vocabs
            item = item[:-5]
        try:
            key, freq = item.rsplit(" ", 1)
        except ValueError:
            continue
        freqs[key] = int(freq)
    return freqs

def get_minority_keys(freqs, min_ratio):
    """Remove keys that are too infrequent relative to a main sense."""
    by_word = defaultdict(list)
    for key, freq in freqs.items():
        try:
            term, sense = split_key(key)
        except ValueError:
            continue
        if freq:
            by_word[term.lower()].append((freq, key))
    discarded = []
    for values in by_word.values():
        if len(values) >= 2:
            values.sort(reverse=True)
            freq1, key1 = values[0]
            for freq2, key2 in values[1:]:
                ratio = freq2 / freq1
                if ratio < min_ratio:
                    discarded.append(key2)
    return discarded

def get_redundant_keys(vocab, vectors, min_distance):
    if min_distance <= 0.0:
        return []
    by_word = defaultdict(list)
    for key, freq in vocab.items():
        try:
            term, sense = split_key(key)
        except ValueError:
            continue
        term = term.split("_")[-1]
        by_word[term.lower()].append((freq, key))
    too_similar = []
    for values in by_word.values():
        if len(values) >= 2:
            values.sort(reverse=True)
            freq1, key1 = values[0]
            vector1 = vectors[key1]
            for freq2, key2 in values[1:]:
                vector2 = vectors[key2]
                sim = cosine_similarity(vector1, vector2)
                if sim >= (1 - min_distance):
                    too_similar.append(key2)
    return too_similar

if __name__ == "__main__":
    typer.run(main)


text:
# This code is part of Qiskit.
#
# (C) Copyright IBM 2022.
#
# This code is licensed under the Apache License, Version 2.0. You may
# obtain a copy of this license in the LICENSE.txt file in the root directory
# of this source tree or at http://www.apache.org/licenses/LICENSE-2.0.
#
# Any modifications or derivative works of this code must retain this
# copyright notice, and modified files need to carry a notice indicating
# that they have been altered from the originals.

"""Tests for Sampler."""

from test import combine
import unittest

from ddt import ddt
import numpy as np

from qiskit import QuantumCircuit
from qiskit.circuit import Parameter
from qiskit.circuit.library import RealAmplitudes
from qiskit.exceptions import QiskitError
from qiskit.primitives import Sampler, SamplerResult
from qiskit.test import QiskitTestCase

@ddt
class TestSampler(QiskitTestCase):
    """Test Sampler"""

    def setUp(self):
        super().setUp()
        hadamard = QuantumCircuit(1, 1)
        hadamard.h(0)
        hadamard.measure(0, 0)
        bell = QuantumCircuit(2)
        bell.h(0)
        bell.cx(0, 1)
        bell.measure_all()
        self._circuit = [hadamard, bell]
        self._target = [
            {0: 0.5, 1: 0.5},
            {0: 0.5, 3: 0.5, 1: 0, 2: 0},
        ]
        self._pqc = RealAmplitudes(num_qubits=2, reps=2)
        self._pqc.measure_all()
        self._pqc_params = [[0.0] * 6, [1.0] * 6]
        self._pqc_target = [{0: 1}, {0: 0.0148, 1: 0.3449, 2: 0.0531, 3: 0.5872}]

    def _generate_circuits_target(self, indices):
        if isinstance(indices, list):
            circuits = [self._circuit[j] for j in indices]
            target = [self._target[j] for j in indices]
        else:
            raise ValueError(f"invalid index {indices}")
        return circuits, target

    def _generate_params_target(self, indices):
        if isinstance(indices, int):
            params = self._pqc_params[indices]
            target = self._pqc_target[indices]
        elif isinstance(indices, list):
            params = [self._pqc_params[j] for j in indices]
            target = [self._pqc_target[j] for j in indices]
        else:
            raise ValueError(f"invalid index {indices}")
        return params, target

    def _compare_probs(self, prob, target):
        if not isinstance(target, list):
            target = [target]
        self.assertEqual(len(prob), len(target))
        for p, targ in zip(prob, target):
            for key, t_val in targ.items():
                if key in p:
                    self.assertAlmostEqual(p[key], t_val, places=1)
                else:
                    self.assertAlmostEqual(t_val, 0, places=1)

    @combine(indices=[[0], [1], [0, 1]])
    def test_sampler(self, indices):
        """test for sampler"""
        circuits, target = self._generate_circuits_target(indices)
        with Sampler(circuits=circuits) as sampler:
            result = sampler(list(range(len(indices))), parameter_values=[[] for _ in indices])
            self._compare_probs(result.quasi_dists, target)

    @combine(indices=[[0], [1], [0, 1]])
    def test_sampler_pqc(self, indices):
        """test for sampler with a parametrized circuit"""
        params, target = self._generate_params_target(indices)
        with Sampler(circuits=self._pqc) as sampler:
            result = sampler([0] * len(params), params)
            self._compare_probs(result.quasi_dists, target)

    @combine(indices=[[0, 0], [0, 1], [1, 1]])
    def test_evaluate_two_pqcs(self, indices):
        """test for sampler with two parametrized circuits"""
        circs = [self._pqc, self._pqc]
        params, target = self._generate_params_target(indices)
        with Sampler(circuits=circs) as sampler:
            result = sampler(indices, parameter_values=params)
            self._compare_probs(result.quasi_dists, target)

    def test_sampler_example(self):
        """test for Sampler example"""

        bell = QuantumCircuit(2)
        bell.h(0)
        bell.cx(0, 1)
        bell.measure_all()

        # executes a Bell circuit
        with Sampler(circuits=[bell], parameters=[[]]) as sampler:
            result = sampler(parameter_values=[[]], circuits=[0])
            self.assertIsInstance(result, SamplerResult)
            self.assertEqual(len(result.quasi_dists), 1)
            keys, values = zip(*sorted(result.quasi_dists[0].items()))
            self.assertTupleEqual(keys, tuple(range(4)))
            np.testing.assert_allclose(values, [0.5, 0, 0, 0.5])
            self.assertEqual(len(result.metadata), 1)

        # executes three Bell circuits
        with Sampler([bell] * 3, [[]] * 3) as sampler:
            result = sampler([0, 1, 2], [[]] * 3)
            self.assertIsInstance(result, SamplerResult)
            self.assertEqual(len(result.quasi_dists), 3)
            self.assertEqual(len(result.metadata), 3)
            for dist in result.quasi_dists:
                keys, values = zip(*sorted(dist.items()))
                self.assertTupleEqual(keys, tuple(range(4)))
                np.testing.assert_allclose(values, [0.5, 0, 0, 0.5])

        with Sampler([bell]) as sampler:
            result = sampler([bell, bell, bell])
            self.assertIsInstance(result, SamplerResult)
            self.assertEqual(len(result.quasi_dists), 3)
            self.assertEqual(len(result.metadata), 3)
            for dist in result.quasi_dists:
                keys, values = zip(*sorted(dist.items()))
                self.assertTupleEqual(keys, tuple(range(4)))
                np.testing.assert_allclose(values, [0.5, 0, 0, 0.5])

        # parametrized circuit
        pqc = RealAmplitudes(num_qubits=2, reps=2)
        pqc.measure_all()
        pqc2 = RealAmplitudes(num_qubits=2, reps=3)
        pqc2.measure_all()

        theta1 = [0, 1, 1, 2, 3, 5]
        theta2 = [1, 2, 3, 4, 5, 6]
        theta3 = [0, 1, 2, 3, 4, 5, 6, 7]

        with Sampler(circuits=[pqc, pqc2], parameters=[pqc.parameters, pqc2.parameters]) as sampler:
            result = sampler([0, 0, 1], [theta1, theta2, theta3])
            self.assertIsInstance(result, SamplerResult)
            self.assertEqual(len(result.quasi_dists), 3)
            self.assertEqual(len(result.metadata), 3)

            keys, values = zip(*sorted(result.quasi_dists[0].items()))
            self.assertTupleEqual(keys, tuple(range(4)))
            np.testing.assert_allclose(
                values,
                [0.13092484629757767, 0.3608720796028449, 0.09324865232050054, 0.414954421779077],
            )

            keys, values = zip(*sorted(result.quasi_dists[1].items()))
            self.assertTupleEqual(keys, tuple(range(4)))
            np.testing.assert_allclose(
                values,
                [0.06282290651933871, 0.02877144385576703, 0.606654494132085, 0.3017511554928095],
            )

            keys, values = zip(*sorted(result.quasi_dists[2].items()))
            self.assertTupleEqual(keys, tuple(range(4)))
            np.testing.assert_allclose(
                values,
                [
                    0.18802639943804164,
                    0.6881971261189544,
                    0.09326232720582446,
                    0.030514147237179882,
                ],
            )

    def test_sampler_param_order(self):
        """test for sampler with different parameter orders"""
        x = Parameter("x")
        y = Parameter("y")

        qc = QuantumCircuit(3, 3)
        qc.rx(x, 0)
        qc.rx(y, 1)
        qc.x(2)
        qc.measure(0, 0)
        qc.measure(1, 1)
        qc.measure(2, 2)

        with Sampler([qc, qc], [[x, y], [y, x]]) as sampler:
            result = sampler([0, 1, 0, 1], [[0, 0], [0, 0], [np.pi / 2, 0], [np.pi / 2, 0]])
            self.assertIsInstance(result, SamplerResult)
            self.assertEqual(len(result.quasi_dists), 4)

            # qc({x: 0, y: 0})
            keys, values = zip(*sorted(result.quasi_dists[0].items()))
            self.assertTupleEqual(keys, tuple(range(8)))
            np.testing.assert_allclose(values, [0, 0, 0, 0, 1, 0, 0, 0])

            # qc({x: 0, y: 0})
            keys, values = zip(*sorted(result.quasi_dists[1].items()))
            self.assertTupleEqual(keys, tuple(range(8)))
            np.testing.assert_allclose(values, [0, 0, 0, 0, 1, 0, 0, 0])

            # qc({x: pi/2, y: 0})
            keys, values = zip(*sorted(result.quasi_dists[2].items()))
            self.assertTupleEqual(keys, tuple(range(8)))
            np.testing.assert_allclose(values, [0, 0, 0, 0, 0.5, 0.5, 0, 0])

            # qc({x: 0, y: pi/2})
            keys, values = zip(*sorted(result.quasi_dists[3].items()))
            self.assertTupleEqual(keys, tuple(range(8)))
            np.testing.assert_allclose(values, [0, 0, 0, 0, 0.5, 0, 0.5, 0])

    def test_sampler_reverse_meas_order(self):
        """test for sampler with reverse measurement order"""
        x = Parameter("x")
        y = Parameter("y")

        qc = QuantumCircuit(3, 3)
        qc.rx(x, 0)
        qc.rx(y, 1)
        qc.x(2)
        qc.measure(0, 2)
        qc.measure(1, 1)
        qc.measure(2, 0)

        with Sampler([qc, qc], [[x, y], [y, x]]) as sampler:
            result = sampler([0, 1, 0, 1], [[0, 0], [0, 0], [np.pi / 2, 0], [np.pi / 2, 0]])
            self.assertIsInstance(result, SamplerResult)
            self.assertEqual(len(result.quasi_dists), 4)

            # qc({x: 0, y: 0})
            keys, values = zip(*sorted(result.quasi_dists[0].items()))
            self.assertTupleEqual(keys, tuple(range(8)))
            np.testing.assert_allclose(values, [0, 1, 0, 0, 0, 0, 0, 0])

            # qc({x: 0, y: 0})
            keys, values = zip(*sorted(result.quasi_dists[1].items()))
            self.assertTupleEqual(keys, tuple(range(8)))
            np.testing.assert_allclose(values, [0, 1, 0, 0, 0, 0, 0, 0])

            # qc({x: pi/2, y: 0})
            keys, values = zip(*sorted(result.quasi_dists[2].items()))
            self.assertTupleEqual(keys, tuple(range(8)))
            np.testing.assert_allclose(values, [0, 0.5, 0, 0, 0, 0.5, 0, 0])

            # qc({x: 0, y: pi/2})
            keys, values = zip(*sorted(result.quasi_dists[3].items()))
            self.assertTupleEqual(keys, tuple(range(8)))
            np.testing.assert_allclose(values, [0, 0.5, 0, 0.5, 0, 0, 0, 0])

    def test_1qubit(self):
        """test for 1-qubit cases"""
        qc = QuantumCircuit(1)
        qc.measure_all()
        qc2 = QuantumCircuit(1)
        qc2.x(0)
        qc2.measure_all()

        with Sampler([qc, qc2], [qc.parameters, qc2.parameters]) as sampler:
            result = sampler([0, 1], [[]] * 2)
            self.assertIsInstance(result, SamplerResult)
            self.assertEqual(len(result.quasi_dists), 2)

            keys, values = zip(*sorted(result.quasi_dists[0].items()))
            self.assertTupleEqual(keys, tuple(range(2)))
            np.testing.assert_allclose(values, [1, 0])

            keys, values = zip(*sorted(result.quasi_dists[1].items()))
            self.assertTupleEqual(keys, tuple(range(2)))
            np.testing.assert_allclose(values, [0, 1])

    def test_2qubit(self):
        """test for 2-qubit cases"""
        qc0 = QuantumCircuit(2)
        qc0.measure_all()
        qc1 = QuantumCircuit(2)
        qc1.x(0)
        qc1.measure_all()
        qc2 = QuantumCircuit(2)
        qc2.x(1)
        qc2.measure_all()
        qc3 = QuantumCircuit(2)
        qc3.x([0, 1])
        qc3.measure_all()

        with Sampler(
            [qc0, qc1, qc2, qc3], [qc0.parameters, qc1.parameters, qc2.parameters, qc3.parameters]
        ) as sampler:
            result = sampler([0, 1, 2, 3], [[]] * 4)
            self.assertIsInstance(result, SamplerResult)
            self.assertEqual(len(result.quasi_dists), 4)

            keys, values = zip(*sorted(result.quasi_dists[0].items()))
            self.assertTupleEqual(keys, tuple(range(4)))
            np.testing.assert_allclose(values, [1, 0, 0, 0])

            keys, values = zip(*sorted(result.quasi_dists[1].items()))
            self.assertTupleEqual(keys, tuple(range(4)))
            np.testing.assert_allclose(values, [0, 1, 0, 0])

            keys, values = zip(*sorted(result.quasi_dists[2].items()))
            self.assertTupleEqual(keys, tuple(range(4)))
            np.testing.assert_allclose(values, [0, 0, 1, 0])

            keys, values = zip(*sorted(result.quasi_dists[3].items()))
            self.assertTupleEqual(keys, tuple(range(4)))
            np.testing.assert_allclose(values, [0, 0, 0, 1])

    def test_errors(self):
        """Test for errors"""
        qc1 = QuantumCircuit(1)
        qc1.measure_all()
        qc2 = RealAmplitudes(num_qubits=1, reps=1)
        qc2.measure_all()

        with Sampler([qc1, qc2], [qc1.parameters, qc2.parameters]) as sampler:
            with self.assertRaises(QiskitError):
                sampler([0], [[1e2]])
            with self.assertRaises(QiskitError):
                sampler([1], [[]])
            with self.assertRaises(QiskitError):
                sampler([1], [[1e2]])

    def test_empty_parameter(self):
        """Test for empty parameter"""
        n = 5
        qc = QuantumCircuit(n, n - 1)
        qc.measure(range(n - 1), range(n - 1))
        with Sampler(circuits=[qc] * 10) as sampler:
            with self.subTest("one circuit"):
                result = sampler([0], shots=1000)
                self.assertEqual(len(result.quasi_dists), 1)
                for q_d in result.quasi_dists:
                    quasi_dist = {k: v for k, v in q_d.items() if v != 0.0}
                    self.assertDictEqual(quasi_dist, {0: 1.0})
                self.assertEqual(len(result.metadata), 1)

            with self.subTest("two circuits"):
                result = sampler([2, 4], shots=1000)
                self.assertEqual(len(result.quasi_dists), 2)
                for q_d in result.quasi_dists:
                    quasi_dist = {k: v for k, v in q_d.items() if v != 0.0}
                    self.assertDictEqual(quasi_dist, {0: 1.0})
                self.assertEqual(len(result.metadata), 2)

    def test_numpy_params(self):
        """Test for numpy array as parameter values"""
        qc = RealAmplitudes(num_qubits=2, reps=2)
        qc.measure_all()
        k = 5
        params_array = np.random.rand(k, qc.num_parameters)
        params_list = params_array.tolist()
        params_list_array = list(params_array)
        with Sampler(circuits=qc) as sampler:
            target = sampler([0] * k, params_list)

            with self.subTest("ndarrary"):
                result = sampler([0] * k, params_array)
                self.assertEqual(len(result.metadata), k)
                for i in range(k):
                    self.assertDictEqual(result.quasi_dists[i], target.quasi_dists[i])

            with self.subTest("list of ndarray"):
                result = sampler([0] * k, params_list_array)
                self.assertEqual(len(result.metadata), k)
                for i in range(k):
                    self.assertDictEqual(result.quasi_dists[i], target.quasi_dists[i])

    def test_passing_objects(self):
        """Test passing objects for Sampler."""

        params, target = self._generate_params_target([0])

        with self.subTest("Valid test"):
            with Sampler(circuits=self._pqc) as sampler:
                result = sampler(circuits=[self._pqc], parameter_values=params)
                self._compare_probs(result.quasi_dists, target)

        with self.subTest("Invalid circuit test"):
            circuit = QuantumCircuit(2)
            with Sampler(circuits=self._pqc) as sampler:
                with self.assertRaises(QiskitError):
                    result = sampler(circuits=[circuit], parameter_values=params)

    @combine(indices=[[0], [1], [0, 1]])
    def test_deprecated_circuit_indices(self, indices):
        """Test for deprecated arguments"""
        circuits, target = self._generate_circuits_target(indices)
        with Sampler(circuits=circuits) as sampler:
            with self.assertWarns(DeprecationWarning):
                result = sampler(
                    circuit_indices=list(range(len(indices))),
                    parameter_values=[[] for _ in indices],
                )
            self._compare_probs(result.quasi_dists, target)

if __name__ == "__main__":
    unittest.main()

max_stars_repo_name: t-imamichi/qiskit-core

max_issues_repo_name: t-imamichi/qiskit-core

max_forks_repo_name: declanmillar/qiskit-terra


text:
__doc__ = """Week 5 exercise: implementing LDA with covariance shrinkage.

.. codeauthor:: Derek Huang <djh458@stern.nyu.edu>
"""

import numpy as np
import pytest
from sklearn.base import BaseEstimator
from sklearn.datasets import make_blobs
from sklearn.discriminant_analysis import LinearDiscriminantAnalysis as _LDA
from sklearn.model_selection import train_test_split
from sklearn.utils import check_array, check_X_y

class LinearDiscriminantAnalysis(BaseEstimator):
    """Student implementation of linear discriminant analysis.

    coef_, intercept_ computed by inversion of sample covariance matrix. The
    sample covariance matrix may have shrinkage applied. Without shrinkage, it
    is the maximum likelihood estimate of the true covariance matrix.

    Parameters
    ----------
    shrinkage : float, default=None
        Value in [0, 1] controlling the shrinkage of the covariance matrix
        estimation. If shrinkage is not None, the covariance matrix estimate
        will be shrinkage * tr(S) / n_features * I + (1 - shrinkage) * S.
        Here S is the maximum likelihood estimate for the covariance matrix,
        I is the identity matrix, tr(S) is the trace of S.

    Attributes
    ----------
    classes_ : numpy.ndarray
        Array of class labels known to the classifier, shape (n_classes,)
    coef_ : numpy.ndarray
        Model coefficients of the fitted model, shape (n_features,) in the case
        of two classes and shape (n_classes, n_features) with >2 classes.
    covariance_ : numpy.ndarray
        Covariance matrix shared by the classes, shape (n_features, n_features)
    intercept_ : numpy.ndarray
        Intercept (bias) terms for the model, shape (n_classes,).
    means_ : numpy.ndarray
        Class means, shape (n_classes, n_features)
    priors_ : numpy.ndarray
        Class priors, shape (n_classes,)

    Methods
    -------
    fit(X, y)
        Compute parameters to best fit the model to X, y.
    predict(X)
        Return predictions on given data X.
    score(X, y)
        Return the accuracy of the predictions given true labels y.
    """
    def __init__(self, shrinkage=None):
        # check that shrinkage is valid
        if shrinkage is None:
            pass
        elif isinstance(shrinkage, float):
            if shrinkage < 0 or shrinkage > 1:
                raise ValueError("shrinkage must be in [0, 1]")
        else:
            raise TypeError("shrinkage must be float in [0, 1]")
        # set shrinkage attribute
        self.shrinkage = shrinkage

    def fit(self, X, y):
        """Compute parameters to best fit the model to X, y.

        Parameters
        ----------
        X : numpy.ndarray
            Input matrix shape (n_samples, n_features)
        y : numpy.ndarray
            Response vector shape (n_samples,)

        Returns
        -------
        self
        """
        # validate input
        X, y = check_X_y(X, y)
        # get n_features
        _, n_features = X.shape
        # get number of unique labels in y. if labels.size < 2, raise error
        labels = np.unique(y)
        if labels.size < 2:
            raise ValueError(
                "Cannot fit coefficients on single-class classification task"
            )
        # number of classes
        n_classes = labels.size
        # compute class priors
        priors = np.empty(n_classes)

        ###########################
        ### your code goes here ###
        ###########################

        # compute the class means
        means = np.empty((n_classes, n_features))

        ###########################
        ### your code goes here ###
        ###########################

        # compute shared + shrunk covariance matrix. sample covariance matrix
        # is convex combination of class prior weighted covariance matrices.
        # note that self.shrinkage might be None, in which case no shrinkage!
        cov = np.zeros((n_features, n_features))

        ###########################
        ### your code goes here ###
        ###########################

        # inverse of covariance matrix (in practice, use np.linalg.lstsq).
        cov_i = np.linalg.inv(cov)
        # compute coefficients
        if n_classes == 2:
            self.coef_ = cov_i @ (means[1] - means[0])
        else:
            self.coef_ = (cov_i @ means.T).T
        # compute intercept(s)
        if n_classes == 2:
            self.intercept_ = (
                -0.5 *
                (means[1] @ cov_i @ means[1] - means[0] @ cov_i @ means[0]) +
                np.log(priors[1] / priors[0])
            )
        else:
            self.intercept_ = (
                -0.5 * np.diag(means @ self.coef_.T) + np.log(priors)
            )
        # set other attributes
        self.classes_ = labels
        self.priors_ = priors
        self.means_ = means
        self.covariance_ = cov
        return self

    def predict(self, X):
        """Return predictions on given data X.

        Parameters
        ----------
        X : numpy.ndarray
            Input matrix shape (n_samples, n_features)

        Returns
        -------
        numpy.ndarray
            Predicted labels, shape (n_samples,). These are the same labels as
            the labels passed during fitting and stored in self.classes_.
        """
        if not hasattr(self, "coef_") or not hasattr(self, "intercept_"):
            raise RuntimeError("cannot predict with unfitted model")
        # validate input matrix
        X = check_array(X)
        # two-class case
        if self.classes_.size == 2:
            if X.shape[1] != self.coef_.shape[0]:
                raise ValueError("n_features must match coef_.shape[0]")
        else:
            if X.shape[1] != self.coef_.shape[1]:
                raise ValueError("n_features must match coef_.shape[1]")
        # two-class. positive class if decision > 0, else negative
        if self.classes_.size == 2:
            return np.where(X @ self.coef_ + self.intercept_ > 0, 1, 0)
        # multi-class. get argmax of decision function values and return labels
        return self.classes_[
            np.argmax(X @ self.coef_.T + self.intercept_, axis=1)
        ]

    def score(self, X, y):
        """Return the accuracy of the predictions given true labels y.

        Parameters
        ----------
        X : numpy.ndarray
            Input matrix shape (n_samples, n_features)
        y : numpy.ndarray
            Response vector shape (n_samples,)

        Returns
        -------
        float
            Fraction of examples predicted correctly
        """
        ###########################
        ### your code goes here ###
        ###########################

@pytest.fixture(scope="session")
def blob_bin():
    """Generated two-class blob classification problem with train/test split.

    Returns
    -------
    X_train : numpy.ndarray
    X_test : numpy.ndarray
    y_train : numpy.ndarray
    y_test : numpy.ndarray
    """
    # seed value for reproducible results
    _seed = 7
    # generate noisy classification problem using isotropic Gaussian blobs
    # pylint: disable=unbalanced-tuple-unpacking
    X, y = make_blobs(
        n_samples=600,
        n_features=10,
        centers=2,
        random_state=_seed
    )
    # pylint: enable=unbalanced-tuple-unpacking
    # split the data with train_test_split and return
    X_train, X_test, y_train, y_test = train_test_split(
        X,
        y,
        test_size=0.2,
        random_state=_seed
    )
    # return split data, coef, and bias
    return X_train, X_test, y_train, y_test

@pytest.fixture(scope="session")
def blob_multi():
    """Generated multi-class blob classification problem with train/test split.

    Returns
    -------
    X_train : numpy.ndarray
    X_test : numpy.ndarray
    y_train : numpy.ndarray
    y_test : numpy.ndarray
    """
    # seed value for reproducible results
    _seed = 7
    # pylint: disable=unbalanced-tuple-unpacking
    X, y = make_blobs(
        n_samples=600,
        n_features=10,
        centers=4,
        random_state=_seed
    )
    # pylint: enable=unbalanced-tuple-unpacking
    X_train, X_test, y_train, y_test = train_test_split(
        X,
        y,
        test_size=0.2,
        random_state=_seed
    )
    return X_train, X_test, y_train, y_test

@pytest.mark.parametrize("shrinkage", [None, 0.2, 0.7])
def test_res_binary(blob_bin, shrinkage):
    """Check LinearDiscriminantAnalysis equals sklearn for two-class case.

    Checks the fit, predict, and score methods, checking solution and accuracy.
    Also checks the different attributes set during the fitting process.

    Parameters
    ----------
    blob_bin : tuple
        pytest fixture. See blob_bin.
    shrinkage : float
        Shrinkage parameter to pass to LinearDiscriminantAnalysis
    """
    # unpack data from fixture
    X_train, X_test, y_train, y_test = blob_bin
    # fit scikit-learn model and our model
    _lc = _LDA(solver="lsqr", shrinkage=shrinkage).fit(X_train, y_train)
    lc = LinearDiscriminantAnalysis(shrinkage=shrinkage).fit(X_train, y_train)
    # check class labels, priors, means, covariance
    np.testing.assert_allclose(_lc.classes_, lc.classes_)
    np.testing.assert_allclose(_lc.priors_, lc.priors_)
    np.testing.assert_allclose(_lc.means_, lc.means_)
    np.testing.assert_allclose(_lc.covariance_, lc.covariance_)
    # check that coefficients and intercepts are close. in the two-class case,
    # sklearn has an extra dimension that can be dropped
    np.testing.assert_allclose(_lc.coef_.ravel(), lc.coef_)
    np.testing.assert_allclose(_lc.intercept_, lc.intercept_)
    # check that predictions are close
    np.testing.assert_allclose(_lc.predict(X_test), lc.predict(X_test))
    # accuracy should be the same
    np.testing.assert_allclose(
        _lc.score(X_test, y_test),
        lc.score(X_test, y_test)
    )

@pytest.mark.parametrize("shrinkage", [None, 0.2, 0.7])
def test_res_multi(blob_multi, shrinkage):
    """Check LinearDiscriminantAnalysis equals sklearn for multi-class case.

    Checks the fit, predict, and score methods, checking solution and accuracy.
    Also checks the different attributes set during the fitting process.

    Parameters
    ----------
    blob_bin : tuple
        pytest fixture. See blob_multi.
    shrinkage : float
        Shrinkage parameter to pass to LinearDiscriminantAnalysis
    """
    # unpack data from fixture
    X_train, X_test, y_train, y_test = blob_multi
    # fit scikit-learn model and our model
    _lc = _LDA(solver="lsqr", shrinkage=shrinkage).fit(X_train, y_train)
    lc = LinearDiscriminantAnalysis(shrinkage=shrinkage).fit(X_train, y_train)
    # check class labels, priors, means, covariance
    np.testing.assert_allclose(_lc.classes_, lc.classes_)
    np.testing.assert_allclose(_lc.priors_, lc.priors_)
    np.testing.assert_allclose(_lc.means_, lc.means_)
    np.testing.assert_allclose(_lc.covariance_, lc.covariance_)
    # check that coefficients and intercepts are close
    np.testing.assert_allclose(_lc.coef_, lc.coef_)
    np.testing.assert_allclose(_lc.intercept_, lc.intercept_)
    # check that predictions are close
    np.testing.assert_allclose(_lc.predict(X_test), lc.predict(X_test))
    # accuracy should be the same
    np.testing.assert_allclose(
        _lc.score(X_test, y_test),
        lc.score(X_test, y_test)
    )

max_stars_repo_name: phetdam/bac-advanced-ml

max_issues_repo_name: phetdam/bac-advanced-ml

max_forks_repo_name: phetdam/bac-advanced-ml


text:
import argparse
import time
import math
import os
import torch
import torch.nn as nn
import torch.onnx
import numpy as np
import pickle
import discriminator_model as model
import data
import pandas as pd
from sklearn.metrics import f1_score

parser = argparse.ArgumentParser(description='PyTorch RNN/LSTM language Model')
parser.add_argument('--data', type=str,
                    help='location of the data corpus')
args = parser.parse_args()

device = torch.device("cuda" if torch.cuda.is_available() else "cpu")

###############################################################################
# Load data
###############################################################################
dic_path = os.path.join(args.data, 'action_dictionary.pkl')
dic_exists = os.path.isfile(dic_path)
if dic_exists:
    with open(dic_path, 'rb') as input:
        Corpus_Dic = pickle.load(input)
else:
    Corpus_Dic = data.Dictionary()

test_data_name = os.path.join(args.data, 'test.csv')

if args.data == 'ssec':
    import json
    # train_data = data.SSEC_DataSet('data_collection/ssec-aggregated/train-combined-0.33.csv')
    test_data = data.SSEC_DataSet('data_collection/ssec-aggregated/test-combined-0.33.csv')
    # train_data.load(dictionary=Corpus_Dic)
    # train_data.labels = json.load(open('ssec_new_label.json'))
    test_data.load(dictionary=Corpus_Dic, train_mode=False)
    test_data.labels = json.load(open('ssec_new_label_test.json'))
else:
    test_data = data.TEC_ISEAR_DataSet(args.data)
    test_data.load(dictionary=Corpus_Dic, train_mode=False, tokenize=True, level=0)

bitch_size = 128

print('The size of the dictionary is', len(Corpus_Dic))

def evaluate(model, test_loader):
    # Turn on evaluate mode which disables dropout.
    correct = 0
    total = 0
    model.eval()
    with torch.no_grad():
        gold = []
        pred = []
        for i_batch, sample_batched in enumerate(test_loader):
            token_seqs = torch.from_numpy(np.transpose(sample_batched[0])).to(device)
            labels = torch.from_numpy(np.array(sample_batched[3])).to(device)
            seq_lengths = np.transpose(sample_batched[4])
            hidden = model.init_hidden(token_seqs.shape[1])
            output, _ = model(token_seqs, hidden, seq_lengths)
            #print(output.size(), labels.size())
            if args.data == 'ssec':
                predict_class = (output > 0).float()
                pred.append(predict_class)
                correct += ((predict_class == labels).sum(axis=1).true_divide(8)).sum().item()
            else:
                _, predict_class = torch.max(output, 1)
                pred.append(predict_class)
                correct += (predict_class == labels).sum().item()
            total += labels.size(0)
            gold.append(labels)
        print('Accuracy of the classifier on the test data is : {:5.4f}'.format(
                100 * correct / total))
        y_true = torch.cat(gold).cpu().detach().numpy()
        y_pred = torch.cat(pred).cpu().detach().numpy()
        return correct / total, f1_score(y_true, y_pred, average='macro'), f1_score(y_true, y_pred, average='micro')

if  __name__ == '__main__':
    result = {
        'model': [],
        'acc': [],
        'micro': [],
        'macro': []
    }
    for filenames in os.walk('results/'+args.data):
        if ['classifier_model.pt'] in filenames:
            test_loader = torch.utils.data.DataLoader(dataset=test_data,
                                                    batch_size=bitch_size,
                                                    shuffle=True,
                                                    collate_fn=data.collate_fn)
            model_name = filenames[0] + '/classifier_model.pt'
            try:
                model = torch.load(model_name, map_location=device)
                acc, micro, macro = evaluate(model, test_loader)
                result['model'].append(filenames[0])
                result['acc'].append(acc)
                result['micro'].append(micro)
                result['macro'].append(macro)
            except Exception as e:
                print(model_name, e)
    df = pd.DataFrame(result)
    df.to_csv(args.data+'_result.csv')


text:
# coding:utf-8
__author__ = 'Ruslan N. Kosarev'

import numpy as np
from PIL import Image
import math

def image_processing(image, box, options):
    if not isinstance(image, Image.Image):
        raise ValueError('Input must be PIL.Image')

    w_margin = round(box.width * options.margin / 2)
    h_margin = round(box.height * options.margin / 2)

    cropped = image.crop((box.left - w_margin, box.top - h_margin, box.right + w_margin, box.bottom + h_margin))

    # compute size of the output image
    width = math.ceil(options.size + options.size * options.margin)
    height = math.ceil(options.size + options.size * options.margin)
    size = (width, height)

    # resize input image
    resized = cropped.resize(size, Image.ANTIALIAS)

    return resized

class BoundingBox:
    def __init__(self, left, top, width, height, confidence=None):
        self.left = int(np.round(left))
        self.right = int(np.round(left + width)) + 1

        self.top = int(np.round(top))
        self.bottom = int(np.round(top + height)) + 1

        self.width = self.right - self.left - 1
        self.height = self.bottom - self.top - 1
        self.confidence = confidence

    def info(self, mode=False):
        if mode is False:
            return '{}'.format([self.left, self.top, self.width, self.height, self.confidence])
        info = "left = {}, top = {}, width = {}, height = {}, confidence = {}".format(self.left, self.top, self.width, self.height, self.confidence)
        return info

    def __repr__(self):
        return self.info(mode=True)

    @property
    def left_upper(self):
        return self.left, self.top

    @property
    def right_lower(self):
        return self.right, self.bottom

    @property
    def confidence_as_string(self):
        return str(np.round(self.confidence, 3))

class MTCNN:
    def __init__(self):
        from mtcnn.mtcnn import MTCNN
        self.__detector = MTCNN().detect_faces
        self.mode = 'RGB'

    def detector(self, image):
        faces = self.__detector(image)
        bboxes = []

        for face in faces:
            box = face['box']
            bbox = BoundingBox(left=box[0], top=box[1], width=box[2], height=box[3], confidence=face['confidence'])
            bboxes.append(bbox)

        return bboxes

class FasterRCNNv3:
    def __init__(self, gpu_memory_fraction=1.0):
        from .frcnnv3 import detector
        self.__detector = detector.FaceDetector(gpu_memory_fraction=gpu_memory_fraction).get_faces
        self.mode = 'RGB'

    def detector(self, image):

        boxes, scores = self.__detector(image)
        bboxes = []

        for (y1, x1, y2, x2), score in zip(boxes, scores):
            bbox = BoundingBox(left=x1, top=y1, width=x2-x1, height=y2-y1, confidence=score)
            bboxes.append(bbox)

        return bboxes

class FaceDetector:
    def __init__(self, detector='frcnnv3', gpu_memory_fraction=1.0):
        self.detector = detector

        if self.detector == 'pypimtcnn':
            obj = MTCNN()
            self.mode = obj.mode
            self.__detector = obj.detector

        elif self.detector == 'frcnnv3':
            obj = FasterRCNNv3(gpu_memory_fraction=gpu_memory_fraction)
            self.mode = obj.mode
            self.__detector = obj.detector

        else:
            raise 'Undefined face detector type {}'.format(self.detector)

    def detect(self, image):
        return self.__detector(image)

    def __repr__(self):
        info = (f'class {self.__class__.__name__}\n' +
                f'detector type: {self.detector}')
        return info

max_stars_repo_name: sMedX/facenet

max_issues_repo_name: RuslanKosarev/FaceNet

max_forks_repo_name: RuslanKosarev/FaceNet