text:
import matplotlib.pyplot as plt
import numpy as np
import scipy
import scipy.linalg
from matplotlib import cm
from matplotlib.patches import FancyArrowPatch
from matplotlib.text import Annotation
from mpl_toolkits.mplot3d import Axes3D
from mpl_toolkits.mplot3d import proj3d
from scipy import integrate
from scipy import interpolate
from scipy.interpolate import griddata
from scipy.optimize import curve_fit
# from sympy import symbols, diff, solve, exp

import basis.robot_math as rm
import math

# import quadpy

PLOT_EDGE = 1

class Arrow3D(FancyArrowPatch):
    def __init__(self, x, y, z, dx, dy, dz, *args, **kwargs):
        super().__init__((0, 0), (0, 0), *args, **kwargs)
        self._xyz = (x, y, z)
        self._dxdydz = (dx, dy, dz)

    def draw(self, renderer):
        x1, y1, z1 = self._xyz
        dx, dy, dz = self._dxdydz
        x2, y2, z2 = (x1 + dx, y1 + dy, z1 + dz)

        xs, ys, zs = proj3d.proj_transform((x1, x2), (y1, y2), (z1, z2), renderer.M)
        self.set_positions((xs[0], ys[0]), (xs[1], ys[1]))
        super().draw(renderer)

class Annotation3D(Annotation):
    def __init__(self, text, xyz, *args, **kwargs):
        super().__init__(text, xy=(0, 0), *args, **kwargs)
        self._xyz = xyz

    def draw(self, renderer):
        x2, y2, z2 = proj3d.proj_transform(*self._xyz, renderer.M)
        self.xy = (x2, y2)
        super().draw(renderer)

def _arrow3D(ax, x, y, z, dx, dy, dz, *args, **kwargs):
    """
    Add an 3d arrow to an `Axes3D` instance.
    """

    arrow = Arrow3D(x, y, z, dx, dy, dz, *args, **kwargs)
    ax.add_artist(arrow)

def _annotate3D(ax, text, xyz, *args, **kwargs):
    """Add anotation `text` to an `Axes3d` instance."""

    annotation = Annotation3D(text, xyz, *args, **kwargs)
    ax.add_artist(annotation)

setattr(Axes3D, 'annotate3D', _annotate3D)
setattr(Axes3D, 'arrow3D', _arrow3D)

def get_pcd_center(pcd):
    return np.array((np.mean(pcd[:, 0]), np.mean(pcd[:, 1]), np.mean(pcd[:, 2])))

def linear_interp_2d(p1, p2, step=1.0):
    diff = np.array(p1) - np.array(p2)
    x = [p1[0], p2[0]]
    y = [p1[1], p2[1]]
    if abs(diff[0]) > abs(diff[1]):
        f = interpolate.interp1d(x, y, kind='linear')
        if x[0] < x[1]:
            x_new = list(np.arange(x[0], x[1], step)) + [x[1]]
        else:
            x_new = list(np.arange(x[1], x[0], step)) + [x[0]]
            x_new = x_new[::-1]
        y_new = list(f(x_new))
    else:
        f = interpolate.interp1d(y, x, kind='linear')
        if y[0] < y[1]:
            y_new = list(np.arange(y[0], y[1], step)) + [y[1]]
        else:
            y_new = list(np.arange(y[1], y[0], step)) + [y[0]]
            y_new = y_new[::-1]
        x_new = list(f(y_new))
    return list(zip(x_new, y_new))

def linear_interp_3d(p1, p2, step=1.0):
    def __find_2d(p1, p2, v, axis):
        p = []
        for current_axis in [0, 1, 2]:
            if current_axis != axis:
                p.append(np.interp(v, (p1[axis], p2[axis]), (p1[current_axis], p2[current_axis])))
            else:
                p.append(v)
        return p

    def __sort_p1_p2(p1, p2, axis=0):
        if p1[axis] > p2[axis]:
            return p2, p1
        return p1, p2

    result = []
    diff = np.abs(np.array(p1) - np.array(p2))
    if list(diff).count(0) == 2:
        axis = diff.argmax()
        p1, p2 = __sort_p1_p2(p1, p2, axis)
        for v in np.arange(p1[axis], p2[axis], step):
            p = __find_2d(p1, p2, v, axis)
            result.append(p)

    return result

def bilinear_interp_2d(target_point, pts, values):
    quad_xs = [p[0] for p in pts]
    quad_ys = [p[1] for p in pts]
    interp_f = interpolate.interp2d(quad_xs, quad_ys, values, kind='linear')
    result = interp_f(target_point[0], target_point[1])[0]
    return result

def interp_3d():
    # 3D example
    total_rad = 10
    z_factor = 3
    noise = 0.1

    num_true_pts = 200
    s_true = np.linspace(0, total_rad, num_true_pts)
    x_true = np.cos(s_true)
    y_true = np.sin(s_true)
    z_true = s_true / z_factor

    num_sample_pts = 80
    s_sample = np.linspace(0, total_rad, num_sample_pts)
    x_sample = np.cos(s_sample) + noise * np.random.randn(num_sample_pts)
    y_sample = np.sin(s_sample) + noise * np.random.randn(num_sample_pts)
    z_sample = s_sample / z_factor + noise * np.random.randn(num_sample_pts)

    tck, u = interpolate.splprep([x_sample, y_sample, z_sample], s=2)
    x_knots, y_knots, z_knots = interpolate.splev(tck[0], tck)
    u_fine = np.linspace(0, 1, num_true_pts)
    x_fine, y_fine, z_fine = interpolate.splev(u_fine, tck)

    fig2 = plt.figure(2)
    ax3d = fig2.add_subplot(111, projection='3d')
    ax3d.plot(x_true, y_true, z_true, 'b')
    ax3d.plot(x_sample, y_sample, z_sample, 'r*')
    ax3d.plot(x_knots, y_knots, z_knots, 'go')
    ax3d.plot(x_fine, y_fine, z_fine, 'g')
    fig2.show()
    plt.show()

def fit_qua_surface(data, axis=2, toggledebug=False):
    """
    :param data:
    :param toggledebug:
    :return:
    """
    # best-fit quadratic curve
    # l = [max(data[:, 0]) - min(data[:, 0]), max(data[:, 1]) - min(data[:, 1]), max(data[:, 2]) - min(data[:, 2])]
    # axis = l.index(min(l))
    x, y, z = symbols('x, y, z')
    if axis == 0:
        A = np.c_[np.ones(data.shape[0]), data[:, 1:], np.prod(data[:, 1:], axis=1), data[:, 1:] ** 2]
        coef, _, _, _ = scipy.linalg.lstsq(A, data[:, 0])
        f = coef[0] + coef[1] * y + coef[2] * z + coef[3] * y * z + coef[4] * y ** 2 + coef[5] * z ** 2 - x
    elif axis == 1:
        A = np.c_[np.ones(data.shape[0]), np.hstack((data[:, :1], data[:, 2:])),
                  np.prod(np.hstack((data[:, :1], data[:, 2:])), axis=1), np.hstack((data[:, :1], data[:, 2:])) ** 2]
        coef, _, _, _ = scipy.linalg.lstsq(A, data[:, 1])
        f = coef[0] + coef[1] * x + coef[2] * z + coef[3] * x * z + coef[4] * x ** 2 + coef[5] * z ** 2 - y
    else:
        A = np.c_[np.ones(data.shape[0]), data[:, :2], np.prod(data[:, :2], axis=1), data[:, :2] ** 2]
        coef, _, _, _ = scipy.linalg.lstsq(A, data[:, 2])
        f = coef[0] + coef[1] * x + coef[2] * y + coef[3] * x * y + coef[4] * x ** 2 + coef[5] * y ** 2 - z
    if toggledebug:
        x_range = (min(data[:, 0].flatten()) - PLOT_EDGE, max(data[:, 0].flatten()) + PLOT_EDGE)
        y_range = (min(data[:, 1].flatten()) - PLOT_EDGE, max(data[:, 1].flatten()) + PLOT_EDGE)
        z_range = (min(data[:, 2].flatten()) - PLOT_EDGE, max(data[:, 2].flatten()) + PLOT_EDGE)
        plot_surface_f(ax, f, x_range, y_range, z_range, dense=.5)
        # ax.scatter(data[:, 0], data[:, 1], data[:, 2], c='k', s=5, alpha=.5)
    return f

def trans_data_pcv(data, toggledebug=False, random_rot=True):
    pcv, pcaxmat = rm.compute_pca(data)
    inx = sorted(range(len(pcv)), key=lambda k: pcv[k])
    x_v = pcaxmat[:, inx[2]]
    y_v = pcaxmat[:, inx[1]]
    z_v = pcaxmat[:, inx[0]]
    pcaxmat = np.asarray([y_v, -x_v, -z_v]).T
    if random_rot:
        pcaxmat = np.dot(rm.rotmat_from_axangle([1, 0, 0], math.radians(5)), pcaxmat)
        pcaxmat = np.dot(rm.rotmat_from_axangle([0, 1, 0], math.radians(5)), pcaxmat)
        pcaxmat = np.dot(rm.rotmat_from_axangle([0, 0, 1], math.radians(5)), pcaxmat)
    data_tr = np.dot(pcaxmat.T, data.T).T
    if toggledebug:
        center = get_pcd_center(data)
        print('center:', center)
        scale = 2
        ax.arrow3D(center[0], center[1], center[2], scale * x_v[0], scale * x_v[1], scale * x_v[2],
                   mutation_scale=10, arrowstyle='->', color='r')
        ax.arrow3D(center[0], center[1], center[2], scale * y_v[0], scale * y_v[1], scale * y_v[2],
                   mutation_scale=10, arrowstyle='->', color='g')
        ax.arrow3D(center[0], center[1], center[2], scale * z_v[0], scale * z_v[1], scale * z_v[2],
                   mutation_scale=10, arrowstyle='->', color='b')
        ax.scatter(data[:, 0], data[:, 1], data[:, 2], c='r', s=5, alpha=.5)
        ax.scatter(data_tr[:, 0], data_tr[:, 1], data_tr[:, 2], c='g', s=5, alpha=.5)
    center = np.mean(data_tr, axis=0)
    data_tr = data_tr - center
    transmat = np.eye(4)
    transmat[:3,:3]=pcaxmat
    transmat[:3,3]=np.mean(data, axis=0)
    return data_tr, transmat

def gaussian(x, y, x0, y0, xalpha, yalpha, A):
    """
    Our function to fit is going to be a sum of two-dimensional Gaussians
    :param x:
    :param y:
    :param x0:
    :param y0:
    :param xalpha:
    :param yalpha:
    :param A:
    :return:
    """
    return A * np.exp(-((x - x0) / xalpha) ** 2 - ((y - y0) / yalpha) ** 2)

def fit_gss_surface(data, toggledebug=False):
    def _gaussian(M, *args):
        """
        This is the callable that is passed to curve_fit. M is a (2,N) array
        where N is the total number of data points in Z, which will be ravelled
        to one dimension.

        :param M:
        :param args:
        :return:
        """
        x, y = M
        arr = np.zeros(x.shape)
        for i in range(len(args) // 5):
            arr += gaussian(x, y, *args[i * 5:i * 5 + 5])
        return arr

    X, Y = np.meshgrid(np.unique(data[:, 0]), np.unique(data[:, 1]))
    Z = np.zeros(X.shape)
    for x, y, z in zip(data[:, 0], data[:, 1], data[:, 2]):
        Z[np.bitwise_and(y == Y, x == X)] = z

    Z = griddata(data[:, :2], data[:, 2], (X, Y), method='nearest')

    # Initial guesses to the fit parameters.
    guess_prms = [(np.mean(data[:, 0]), np.mean(data[:, 1]), 1, 1, 2)]
    p0 = [p for prms in guess_prms for p in prms]

    # Ravel the meshgrids of X, Y points to a pair of 1-D arrays.
    xdata = np.vstack((X.ravel(), Y.ravel()))
    # Do the fit, using _gaussian function which understands flattened (ravelled) ordering of the data points.
    popt, pcov = curve_fit(_gaussian, xdata, Z.ravel(), p0)
    fit = np.zeros(Z.shape)
    for i in range(len(popt) // 5):
        fit += gaussian(X, Y, *popt[i * 5:i * 5 + 5])
    print('Fitted parameters:\n', popt)

    rms = np.sqrt(np.mean((Z - fit) ** 2))
    print('RMS residual =', rms)

    x, y, z = symbols('x, y, z')
    f = popt[4] * exp(-((x - popt[0]) / popt[2]) ** 2 - ((y - popt[1]) / popt[3]) ** 2) - z

    if toggledebug:
        x_range = (min(data[:, 0].flatten()) - PLOT_EDGE, max(data[:, 0].flatten()) + PLOT_EDGE)
        y_range = (min(data[:, 1].flatten()) - PLOT_EDGE, max(data[:, 1].flatten()) + PLOT_EDGE)
        z_range = (min(data[:, 2].flatten()) - PLOT_EDGE, max(data[:, 2].flatten()) + PLOT_EDGE)
        plot_surface_f(ax, f, x_range, y_range, z_range, axis='z', dense=.5, c=cm.coolwarm)

        ax.scatter(data[:, 0], data[:, 1], data[:, 2], c='k')
        # ax.plot_surface(X, Y, Z)

        # ax.contourf(X, Y, Z - fit, zdir='z', offset=-4, cmap='plasma')
        # ax.set_zlim(-4, np.max(fit))

    return f

def fit_plane(data, toggledebug=False):
    # best-fit linear plane
    A = np.c_[data[:, 0], data[:, 1], np.ones(data.shape[0])]
    coef, _, _, _ = scipy.linalg.lstsq(A, data[:, 2])  # coefficients
    x, y, z = symbols('x, y, z')
    print(coef)
    f = coef[0] * x + coef[1] * y + coef[2] - z

    if toggledebug:
        x_range = (min(data[:, 0].flatten()), max(data[:, 0].flatten()))
        y_range = (min(data[:, 1].flatten()), max(data[:, 1].flatten()))
        z_range = (min(data[:, 2].flatten()), max(data[:, 2].flatten()))
        plot_surface_f(ax, f, x_range, y_range, z_range, dense=.5)

    return f

def plot_surface_f(ax, f, x_range, y_range, z_range, dense=.5, alpha=.8, c=None, axis=None):
    x, y, z = symbols('x, y, z')
    coef_dict = f.as_coefficients_dict()
    if axis is None:
        if z in coef_dict.keys():
            axis = 'z'
        elif y in coef_dict.keys():
            axis = 'y'
        else:
            axis = 'x'
    if axis == 'z':
        s = solve(f, z)[0]
        X, Y = np.meshgrid(np.arange(x_range[0] - PLOT_EDGE, x_range[1] + PLOT_EDGE, dense),
                           np.arange(y_range[0] - PLOT_EDGE, y_range[1] + PLOT_EDGE, dense))
        Z = []
        for y in Y[:, 0]:
            z_res = []
            for x in X[0]:
                z_res.append(s.subs({'x': x, 'y': y}))
            Z.append(z_res)
        Z = np.asarray(Z).astype(dtype=float)
    elif axis == 'y':
        s = solve(f, y)[0]
        X, Z = np.meshgrid(np.arange(x_range[0] - PLOT_EDGE, x_range[1] + PLOT_EDGE, dense),
                           np.arange(z_range[0] - PLOT_EDGE, z_range[1] + PLOT_EDGE, dense))
        Y = []
        for z in Z[:, 0]:
            y = []
            for x in X[0]:
                y.append(s.subs({'x': x, 'z': z}))
            Y.append(y)
        Y = np.asarray(Y).astype(dtype=float)
    else:
        s = solve(f, x)[0]
        Y, Z = np.meshgrid(np.arange(y_range[0] - PLOT_EDGE, y_range[1] + PLOT_EDGE, dense),
                           np.arange(z_range[0] - PLOT_EDGE, z_range[1] + PLOT_EDGE, dense))
        X = []
        for y in Y[:, 0]:
            x = []
            for z in Z[0]:
                x.append(s.subs({'y': y, 'z': z}))
            X.append(x)
        X = np.asarray(X).astype(dtype=float)
    if c is None:
        ax.plot_surface(X, Y, Z, rstride=1, cstride=1, alpha=alpha, color='k')
    elif type(c) == str:
        ax.plot_surface(X, Y, Z, rstride=1, cstride=1, alpha=alpha, color=c)
    else:
        ax.plot_surface(X, Y, Z, rstride=1, cstride=1, alpha=alpha, cmap=c)
    # ax.set_zlim(z_range[0] - PLOT_EDGE, z_range[1] + PLOT_EDGE)
    ax.set_xlabel('X')
    ax.set_ylabel('Y')
    ax.set_zlabel('Z')

def get_nrml(coef, p):
    if len(coef) == 6:
        nrml = np.asarray([coef[1] + coef[3] * p[1] + 2 * coef[4] * p[0],
                           coef[2] + coef[3] * p[0] + 2 * coef[5] * p[1],
                           -1.0], dtype=np.float64)
        nrml = nrml / np.linalg.norm(nrml)
        return nrml
    else:
        print('wrong coef length!')
        return None

def get_plane(v1, v2, p, transmat=None, toggledebug=False):
    if transmat is not None:
        v1 = np.dot(transmat, v1)
        v2 = np.dot(transmat, v2)
        p = np.dot(transmat, p)
    n = np.cross(v1, v2)
    coef = [n[0], n[1], n[2], -sum(n * p)]
    x, y, z = symbols('x, y, z')
    f = coef[0] * x + coef[1] * y + coef[2] * z + coef[3]
    if toggledebug:
        x_range = (p[0] - PLOT_EDGE, p[0] + PLOT_EDGE)
        y_range = (p[1] - PLOT_EDGE, p[1] + PLOT_EDGE)
        z_range = (p[2] - PLOT_EDGE, p[2] + PLOT_EDGE)
        ax.scatter([p[0]], [p[1]], [p[2]], c='g', s=10, alpha=1)
        ax.arrow3D(p[0], p[1], p[2], v1[0], v1[1], v1[2], mutation_scale=10, arrowstyle='->', color='g')
        ax.arrow3D(p[0], p[1], p[2], v2[0], v2[1], v2[2], mutation_scale=10, arrowstyle='->', color='g')
        plot_surface_f(ax, f, x_range, y_range, z_range, dense=.5)
        # plt.show()
    return f

def bisection(func, tgt, torlerance, lb, ub, mode='lb'):
    l = lb
    r = ub
    while True:
        v = l + (r - l) / 2
        tmpv, _ = func(v)
        # print(tmpv, v)
        if abs(tmpv - tgt) < torlerance:
            # print(v, tmpv)
            return v, tmpv
        else:
            if mode == 'lb':
                if tmpv < tgt:
                    r = v
                else:
                    l = v
            else:
                if tmpv < tgt:
                    l = v
                else:
                    r = v

def plot_intersc(ax, f, intersc, p, transmat=None, plot_edge=2, c='k', alpha=.5):
    xs = []
    ys = []
    zs = []
    x, y, z = symbols('x, y, z')

    if x in list(f.free_symbols):
        for x_value in np.arange(p[0] - plot_edge, p[0] + plot_edge, 0.1):
            y_value = f.subs({'x': x_value})
            z_value = solve(intersc.subs({'x': x_value, 'y': y_value}), z)[0]
            xs.append(x_value)
            ys.append(y_value)
            zs.append(z_value)
    else:
        for y_value in np.arange(p[1] - plot_edge, p[1] + plot_edge, 0.1):
            x_value = f.subs({'y': y_value})
            z_value = solve(intersc.subs({'x': x_value, 'y': y_value}), z)[0]
            xs.append(x_value)
            ys.append(y_value)
            zs.append(z_value)

    ps = np.hstack(([[x] for x in xs], [[y] for y in ys], [[z] for z in zs]))
    if transmat is not None:
        ps = np.dot(transmat, ps.T).T
    ax.plot(ps[:, 0], ps[:, 1], ps[:, 2], c=c, alpha=alpha)

def cal_surface_intersc(F, G, p, tgtlen, mode='ub', method='bisec', torlerance=1e-5, itg_axis='x', toggledebug=False):
    """

    :param F:
    :param G:
    :param p:
    :param tgtlen:
    :param mode: 'ub', 'lb'
    :param method: 'bisec', 'primitive'
    :param torlerance:
    :param itg_axis: 'x', 'y'
    :param toggledebug:
    :return:
    """

    def _func_L_x(x):
        return L_x.subs({'x': x})

    def _func_L_y(y):
        return L_y.subs({'y': y})

    def _elimination(F1, F2, p, itg_axis=itg_axis):
        if itg_axis == 'x':
            x_y1 = solve(F1, y)[0]
            x_y2 = solve(F1, y)[1]
            y_s1 = x_y1.subs({'x': p[0]})
            y_s2 = x_y2.subs({'x': p[0]})
            if abs(p[1] - y_s1) < abs(p[1] - y_s2):
                return x_y1, F2.subs({'y': x_y1})
            else:
                return x_y2, F2.subs({'y': x_y2})
        else:
            y_x1 = solve(F1, x)[0]
            y_x2 = solve(F1, x)[1]
            x_s1 = y_x1.subs({'y': p[1]})
            x_s2 = y_x2.subs({'y': p[1]})
            if abs(p[0] - x_s1) < abs(p[0] - x_s2):
                return y_x1, F2.subs({'x': y_x1})
            else:
                return y_x2, F2.subs({'x': y_x2})

    def _solve_bnd(tgt_func, tgtlen, mode, start_p, eliminated_func, itg_axis=itg_axis):
        if mode not in ['ub', 'lb']:
            print('wrong mode!')

        step = tgtlen * 1.5
        if itg_axis == 'x':
            if mode == 'ub':
                func = lambda x: integrate.quad(tgt_func, start_p[0], x)
                x_final, real_len = bisection(func, tgtlen, torlerance, start_p[0], start_p[0] + step, 'ub')
            else:
                func = lambda x: integrate.quad(tgt_func, x, start_p[0])
                x_final, real_len = bisection(func, tgtlen, torlerance, start_p[0] - step, start_p[0], 'lb')
            y_final = eliminated_func.subs({'x': x_final})
            z_final = float(F.subs({'x': x_final, 'y': y_final, 'z': 0}))

        else:
            if mode == 'ub':
                func = lambda y: integrate.quad(tgt_func, start_p[1], y)
                y_final, real_len = bisection(func, tgtlen, torlerance, start_p[1], start_p[1] + step, 'ub')
            else:
                func = lambda y: integrate.quad(tgt_func, y, start_p[1])
                y_final, real_len = bisection(func, tgtlen, torlerance, start_p[1] - step, start_p[1], 'lb')
            x_final = eliminated_func.subs({'y': y_final})
            z_final = float(F.subs({'x': x_final, 'y': y_final, 'z': 0}))

        return np.asarray([x_final, y_final, z_final], dtype=np.float64), real_len

    if itg_axis not in ['x', 'y']:
        print('wrong integration axis!')
        return None

    x, y, z = symbols('x, y, z')

    F_dx = diff(F, x)
    F_dy = diff(F, y)
    F_dz = diff(F, z)
    G_dx = diff(G, x)
    G_dy = diff(G, y)
    G_dz = diff(G, z)

    J1 = F_dy * G_dz - F_dz * G_dy
    J2 = F_dz * G_dx - F_dx * G_dz
    J3 = F_dx * G_dy - F_dy * G_dx

    if itg_axis == 'x':
        L = (1 + (J2 / J1) ** 2 + (J3 / J1) ** 2) ** (1 / 2)
    else:
        L = (1 + (J1 / J2) ** 2 + (J3 / J2) ** 2) ** (1 / 2)

    intersc = F.subs({'z': solve(G, z)[0]})

    # print('intersc', intersc)
    # print('f_l', G)

    if method == 'bisec':
        if itg_axis == 'x':
            f_x, L_x = _elimination(intersc, L, p)
            if toggledebug:
                plot_intersc(ax, f_x, F, p)
                plot_intersc(ax, f_x, F, p, transmat=transmat)
            result_p, result_len = _solve_bnd(_func_L_x, tgtlen, mode, p, f_x)
            return result_p, result_len, F, G, f_x
        else:
            f_y, L_y = _elimination(intersc, L, p)
            if toggledebug:
                plot_intersc(ax, f_y, F, p)
                plot_intersc(ax, f_y, F, p, transmat=transmat)
            result_p, result_len = _solve_bnd(_func_L_y, tgtlen, mode, p, f_y)
            return result_p, result_len, F, G, f_y

    # elif method == 'primitive':
    #     x_y, L = _elimination(intersc, L, p, obj='y')
    #     # Get origin and calculate the upper bound
    #     prim_L = itg(nsimplify(simplify(L)), x)
    #     print(f'origin L: {prim_L}')
    #     line_length = 5
    #     F_p0 = prim_L.subs({'x': p[0]})
    #     x_final = solve(prim_L - F_p0 - line_length, x)
    #     y_final = x_y.subs({'x': x_final})
    #     z_final = float(F.subs({'x': x_final, 'y': y_final, 'z': 0}))
    #     if toggledebug:
    #         plot_intersc(ax, x_y, F, p)
    #     return np.asarray([x_final, y_final, z_final]), tgtlen, F, G, x_y

def cal_surface_intersc_p2p(F, G, p1, p2, itg_axis='x'):
    """
    :param coef1:
    :param coef2:
    :param p:
    :param tgtlen:
    :param mode: 'ub', 'lb'
    :param method: 'bisec', 'primitive'
    :param torlerance:
    :param itg_axis: 'x', 'y'
    :param toggledebug:
    :return:
    """

    def _elimination(F1, F2, p, obj='y'):
        if obj == 'x':
            y_x1 = solve(F1, x)[0]
            y_x2 = solve(F1, x)[1]
            x_s1 = y_x1.subs({'y': p[1]})
            x_s2 = y_x2.subs({'y': p[1]})
            if abs(p[0] - x_s1) < abs(p[0] - x_s2):
                return y_x1, F2.subs({'x': y_x1})
            else:
                return y_x2, F2.subs({'x': y_x2})
        else:
            x_y1 = solve(F1, y)[0]
            x_y2 = solve(F1, y)[1]
            y_s1 = x_y1.subs({'x': p[0]})
            y_s2 = x_y2.subs({'x': p[0]})
            if abs(p[1] - y_s1) < abs(p[1] - y_s2):
                return x_y1, F2.subs({'y': x_y1})
            else:
                return x_y2, F2.subs({'y': x_y2})

    def _func_L_x(x):
        return L_x.subs({'x': x})

    def _func_L_y(y):
        return L_y.subs({'y': y})

    if itg_axis not in ['x', 'y']:
        print('wrong integration axis!')
        return None

    x, y, z = symbols('x, y, z')

    F_dx = diff(F, x)
    F_dy = diff(F, y)
    F_dz = diff(F, z)
    G_dx = diff(G, x)
    G_dy = diff(G, y)
    G_dz = diff(G, z)

    J1 = F_dy * G_dz - F_dz * G_dy
    J2 = F_dz * G_dx - F_dx * G_dz
    J3 = F_dx * G_dy - F_dy * G_dx

    intersc = F.subs({'z': solve(G, z)[0]})

    if itg_axis == 'x':
        L = (1 + (J2 / J1) ** 2 + (J3 / J1) ** 2) ** (1 / 2)
        x_y, L_x = _elimination(intersc, L, p1, obj='y')
        gd, _ = integrate.quad(_func_L_x, p1[0], p2[0])
        return abs(gd)
    else:
        L = (1 + (J1 / J2) ** 2 + (J3 / J2) ** 2) ** (1 / 2)
        y_x, L_y = _elimination(intersc, L, p1, obj='x')
        gd, _ = integrate.quad(_func_L_y, p1[1], p2[1])
        return abs(gd)

if __name__ == '__main__':
    TB = True
    fig = plt.figure()
    ax = fig.gca(projection='3d')
    # p1 = (-50, -50, 100)
    # p2 = (-50, -50, 0)
    # linear_interp_3d(p1, p2, 1)
    # interp_3d()

    """
    fit surface
    """
    mean = np.array([0.0, 0.0, 0.0])
    cov = np.array([[1.0, -0.5, 0.8], [-0.5, 1.1, 0.0], [0.8, 0.0, 1.0]])
    data = np.random.multivariate_normal(mean, cov, 50)

    # f_q_base = fit_qua_surface(data, toggledebug=TB)
    # print(f_q_base)

    data_tr, transmat = trans_data_pcv(data)
    f_q = fit_plane(data_tr, toggledebug=TB)
    print(f_q)

    f_gaussian = fit_gss_surface(data_tr, toggledebug=TB)
    print(f_gaussian)
    plt.show()

    v1 = (1, 0, 0)
    v2 = (0, 1, 0)
    p = data[0]
    ax.scatter([p[0]], [p[1]], [p[2]], c='r', s=10, alpha=1)
    ax.arrow3D(p[0], p[1], p[2], v1[0], v1[1], v1[2], mutation_scale=10, arrowstyle='->', color='r')
    ax.arrow3D(p[0], p[1], p[2], v2[0], v2[1], v2[2], mutation_scale=10, arrowstyle='->', color='r')

    f_l = get_plane(v1, v2, p, transmat=transmat.T, toggledebug=TB)
    print(f_l)

    f_l_base = get_plane(v1, v2, p, toggledebug=TB)
    print(f_l_base)

    p_ub_new, _, _, _, _ = \
        cal_surface_intersc(f_gaussian, f_l, np.dot(transmat.T, p), tgtlen=3, toggledebug=TB, mode='ub', itg_axis='y')
    print(p, p_ub_new)

    p_ub = np.dot(transmat, p_ub_new)
    p_new = np.dot(transmat.T, p)
    ax.scatter([p_ub[0]], [p_ub[1]], [p_ub[2]], c='m', s=10, alpha=.5)
    ax.scatter([p_ub_new[0]], [p_ub_new[1]], [p_ub_new[2]], c='y', s=10, alpha=1)

    length = cal_surface_intersc_p2p(f_gaussian, f_l, p_new, p_ub_new, itg_axis='y')
    length_base = cal_surface_intersc_p2p(f_q_base, f_l_base, p, p_ub, itg_axis='y')

    print('result:', length, length_base)
    plt.show()

    p_lb, _, _, _, _ = cal_surface_intersc(f_q, f_l, p, tgtlen=3, toggledebug=TB, mode='lb', itg_axis='y')
    print(p, p_lb)

    p_ub_new, _, _, _, _ = cal_surface_intersc(f_q, f_l, p, tgtlen=3, toggledebug=TB, mode='ub', itg_axis='x')
    print(p, p_ub_new)

    p_lb, _, _, _, _ = cal_surface_intersc(f_q, f_l, p, tgtlen=3, toggledebug=TB, mode='lb', itg_axis='x')
    print(p, p_lb)

max_stars_repo_name: takuya-ki/wrs

max_issues_repo_name: takuya-ki/wrs

max_forks_repo_name: takuya-ki/wrs

reason: import numpy,import scipy,from scipy,from sympy


text:
import dash
import dash_core_components as dcc
import dash_html_components as html
from dash.dependencies import Input, Output, State

import numpy as np
import pandas as pd

# custom import(s)
import helpers.gui_setup as gui_setup
from helpers.prediction_fn import prediction_func
from helpers.app_config import app_config_cls

# initialize gui helper class
app_config = app_config_cls()
gui_params = gui_setup.gui_params_cls(use_SQL_method=True)

# dash app config
app = dash.Dash(__name__)
app.config.update({
    'routes_pathname_prefix': app_config.dash_config.routes_pathname_prefix,
    'requests_pathname_prefix': app_config.dash_config.requests_pathname_prefix
})
# for serving as Flask app
flask_app = app.server

# app layout
app.layout = html.Div([
    html.H2(
        children='Interactive Support Vector Machine for Regression (SVR) Demo',
        style={'textAlign': 'center'}
    ),
    html.H4(
        children='Actual Function (Logarithmic Spiral): c(t) = <[alpha * e^(k * t) * cos(t)], [alpha * e^(k * t) * sin(t)]>',
        style={'textAlign': 'left'}
    ),
    html.Div([
        # plot
        html.Div(
            html.Div(id=gui_params.graph_container_id.comp_id_str),
        ),
        # determine if data should be random or fixed with a seed
        html.Div([
            html.Div(
                html.P(gui_params.shuffle_data.additional_value_str),
                style={'display': 'inline-block'}
            ),
            html.Div(
                dcc.RadioItems(
                    id=gui_params.shuffle_data.comp_id_str,
                    options=[
                        {'label': 'True', 'value': 'True'},
                        {'label': 'False', 'value': 'False'}
                    ], value=gui_params.shuffle_data.value_str #'False'
                ),
                style={'display': 'inline-block'}
            ),
        ]),
        # return array of interactive sliders as a tab list (horizontal)
        html.Div(
            dcc.Tabs(vertical=False, children=[
                gui_params.get_dcc_slider(obj) for obj in gui_params.tab_params_list
                # gui_setup.dcc_slider_wrapper(obj) for obj in gui_params.tab_params_list
            ]),
        ),
    ])
])

# update gui sliders
@app.callback(
    [Output(obj.container_id_str, 'children') for obj in gui_params.tab_params_list],
    [Input(obj.slider_id_str, 'value') for obj in gui_params.tab_params_list]
)
def slider_strs_update(random_sample_size, svr_c, svr_epsilon, actual_func_k, actual_func_alpha, noise_offset, noise_scale):
    return (
        'random sample size: {value}'.format(value=random_sample_size),
        'regularization value: {value}'.format(value=svr_c),
        'training loss no-penalty range: {value}'.format(value=svr_epsilon),
        'prediction function parameter "k" value: {value}'.format(value=actual_func_k),
        'prediction function parameter "alpha" value: {value}'.format(value=actual_func_alpha),
        'observation noise deviation offset value: {value}'.format(value=noise_offset),
        'observation noise deviation scale value: {value}'.format(value=noise_scale)
    )

# update plot
@app.callback(
    Output(gui_params.graph_container_id.comp_id_str, 'children'),
    (
        [Input(obj.slider_id_str, 'value') for obj in gui_params.tab_params_list] +
        [Input(gui_params.shuffle_data.comp_id_str, 'value')]
    )
)
def update_graph(random_sample_size, svr_c, svr_epsilon, actual_func_k, actual_func_alpha, noise_offset, noise_scale, shuffle_data):

    # train model and return predictions in a pandas dataframe
    dff = prediction_func(
        random_sample_size=random_sample_size,
        svr_c=svr_c,
        svr_epsilon=svr_epsilon,
        actual_func_k=actual_func_k,
        actual_func_alpha=actual_func_alpha,
        noise_offset=noise_offset,
        noise_scale=noise_scale,
        shuffle_random_data=True if shuffle_data == 'True' else False
    )

    # lists to reference from in dcc.Graph
    column_list = [
        ['X Actual Function', 'Y Actual Function'],
        ['X Prediction', 'Y Prediction'],
        ['X Sample Data', 'Y Sample Data'],
        ['X Support Vectors', 'Y Support Vectors']
    ]
    data_points_list = ['X Sample Data', 'X Support Vectors']

    # return plot in JSON format
    return [
        dcc.Graph(
            id='graph_id',
            animate=True,
            config={
                'responsive': True,
                'showTips': True
            },
            figure={
                'data': [
                    {
                        'x': dff[column_x],
                        'y': dff[column_y],
                        'name': '{col_name}'.format(col_name=column_x.replace('X ', '')),
                        'legend': 'legendonly',
                        'type': 'line' if column_x not in data_points_list else 'scatter',
                        'line': {'shape': 'linear' if column_x not in data_points_list else None},
                        'mode': 'markers' if column_x in data_points_list else None
                    } for column_x, column_y in column_list if column_x in dff.columns
                ],
                'layout': {
                    'xaxis': {'automargin': True},
                    'yaxis': {'automargin': True},
                    'autosize': True,
                    'automargin': True,
                    'width': '1500',
                    'height': '850',
                    'hovermode': 'closest',
                    'legend': {
                        'orientation': 'v'
                    }
                },
            },
        )
    ]

if __name__ == '__main__':
    app.run_server(
        debug=True,
        port=app_config.dash_config.port,
        host=app_config.dash_config.host
    )

ext: py

reason: import numpy


text:
import re
from nltk.corpus import stopwords
from string import punctuation
import torch
import numpy as np
from typing import List
import random

"""
Author: Anish Acharya <anishacharya@utexas.edu>
Adopted From: Greg Durret <gdurrett@cs.utexas.edu>
"""

class TextCleaning:
    def __init__(self):
        """ Define Your stop word List here"""
        stops = set(stopwords.words("english"))
        stops.update(set(punctuation))

        self.stops = stops

    def text_cleaning(self, text):
        """Define your Text Cleaning Rules here"""
        # *** Remove this if you are using capitalization as feature ex. NER **
        text = text.lower()

        text = re.sub(r"\. \. \.", "\.", text)
        text = re.sub(r"[^A-Za-z0-9(),!?\'\`\.]", " ", text)
        text = re.sub(r'[0-9]+', '', text)
        text = re.sub(r"\'s", " \'s", text)
        text = re.sub(r"\'ve", " \'ve", text)
        text = re.sub(r"n\'t", " n\'t", text)
        text = re.sub(r"\'re", " \'re", text)
        text = re.sub(r"\'d", " \'d", text)
        text = re.sub(r"\'ll", " \'ll", text)
        text = re.sub(r",", "", text)
        text = re.sub(r"!", "", text)
        text = re.sub(r"\(", "", text)
        text = re.sub(r"\)", "", text)
        text = re.sub(r"\?", "", text)
        text = re.sub(r"\s{2,}", " ", text)
        text = re.sub(r"<br />", " ", text)
        text = re.sub(r'[^\w\s]', '', text)
        text = text.split(" ")
        # text = [w for w in text if w not in self.stops]
        return text

# Print iterations progress
def print_progress_bar(iteration, total, prefix='', suffix='', decimals=1, length=100, fill='█'):
    """
    Call in a loop to create terminal progress bar
    @params:
        iteration   - Required  : current iteration (Int)
        total       - Required  : total iterations (Int)
        prefix      - Optional  : prefix string (Str)
        suffix      - Optional  : suffix string (Str)
        decimals    - Optional  : positive number of decimals in percent complete (Int)
        length      - Optional  : character length of bar (Int)
        fill        - Optional  : bar fill character (Str)
    """
    percent = ("{0:." + str(decimals) + "f}").format(100 * (iteration / float(total)))
    filled_length = int(length * iteration // total)
    bar = fill * filled_length + '-' * (length - filled_length)
    print('\r%s |%s| %s%% %s' % (prefix, bar, percent, suffix), end='\r')
    # Print New Line on Complete
    if iteration == total:
        print()

def flatten(lst):
    """Flattens a list of lists"""
    return [sub_elem for elem in lst
            for sub_elem in elem]

def argmax_from_onehot(vec):
    _, idx = torch.max(vec, 0)
    return idx

def pad_to_length(sentence: List, length: int, pad_ix: int) -> List:
    """
    Forces np_arr to length by either truncation (if longer) or zero-padding (if shorter)
    :param pad_ix:
    :param np_arr:
    :param length: Length to pad to
    :return: a new numpy array with the data from np_arr padded to be of length length. If length is less than the
    length of the base array, truncates instead.
    """
    result = [pad_ix] * length
    result[0:len(sentence)] = sentence
    return result

def get_batch(data: List, batch_size: int, start_ix: int) -> List:
    """ given a start ix and batch size this will return truncated data(list) as batch """
    if len(data) == 0: return []
    return data[start_ix:] if len(data[start_ix:]) <= batch_size else data[start_ix: start_ix + batch_size]

def word_dropout(dropout_prob: float) -> bool:
    """ Toss a biased coin and return bool if to drop this token"""
    if random.random() < dropout_prob:
        return True
    return False

def get_onehot_np(y: np.array, no_classes: int):
    # np.eye won't work for float which we use to get one hot
    y_onehot_np = np.eye(no_classes, dtype=np.int32)[y]
    y_onehot_np = np.squeeze(y_onehot_np, axis=0)
    return y_onehot_np

def count_parameters(model):
    return sum(p.numel() for p in model.parameters() if p.requires_grad)

max_stars_repo_name: anishacharya/nlp-wrapper

max_issues_repo_name: anishacharya/nlp-wrapper

max_forks_repo_name: anishacharya/nlp-wrapper


text:
import os
import h5py
import logging
from os import path
import pandas as pd
import numpy as np
from collections import defaultdict

logger = logging.getLogger(__name__)

class SegmentRecord:
    """Stores information needed to reconstruct a segment from chops"""
    def __init__(self, seg_id, clock_time, offset, n_chops, overlap):
        """Stores the information needed to reconstruct a segment

        Parameters
        ----------
        seg_id : int
            The ID of this segment.
        clock_time : pd.Series
            The TimeDeltaIndex of the original data from this segment
        offset : int
            The offset of the chops from the start of the segment
        n_chops : int
            The number of chops that make up this segment
        overlap : int
            The number of bins of overlap between adjacent chops
        """
        self.seg_id = seg_id
        self.clock_time = clock_time
        self.offset = offset
        self.n_chops = n_chops
        self.overlap = overlap

    def rebuild_segment(self, chops, smooth_pwr=2):
        """Reassembles a segment from its chops

        Parameters
        ----------
        chops : np.ndarray
            A 3D numpy array of shape n_chops x seg_len x data_dim that
            holds the data from all of the chops in this segment
        smooth_pwr : float, optional
            The power to use for smoothing. See `merge_chops`
            function for more details, by default 2

        Returns
        -------
        pd.DataFrame
            A DataFrame of reconstructed segment data, indexed by the
            clock_time of the original segment
        """
        # Merge the chops for this segment
        merged_array = merge_chops(
            chops,
            overlap=self.overlap,
            orig_len=len(self.clock_time) - self.offset,
            smooth_pwr=smooth_pwr)
        # Add NaNs for points that were not modeled due to offset
        data_dim = merged_array.shape[1]
        offset_nans = np.full((self.offset, data_dim), np.nan)
        merged_array = np.concatenate([offset_nans, merged_array])
        # Recreate segment DataFrame with the appropriate `clock_time`s
        try:
            segment_df = pd.DataFrame(merged_array, index=self.clock_time)
        except:
            import pdb; pdb.set_trace()
            segment_df = None
        return segment_df

class ChopInterface:
    """Chops data from NWBDatasets into segments with fixed overlap"""
    def __init__(self,
                 window,
                 overlap,
                 max_offset=0,
                 chop_margins=0,
                 random_seed=None):
        """Initializes a ChopInterface

        Parameters
        ----------
        window : int
            The length of chopped segments in ms
        overlap : int
            The overlap between chopped segments in ms
        max_offset : int, optional
            The maximum offset of the first chop from the beginning of
            each segment in ms. The actual offset will be chose
            randomly. By default, 0 adds no offset
        chop_margins : int, optional
            The size of extra margins to add to either end of each chop
            in bins, designed for use with the temporal_shift operation,
            by default 0
        random_seed : int, optional
            The random seed for generating the dataset, by default None
            does not use a random seed
        """
        def to_timedelta(ms):
            return pd.to_timedelta(ms, unit='ms')

        self.window = to_timedelta(window)
        self.overlap = to_timedelta(overlap)
        self.max_offset = to_timedelta(max_offset)
        self.chop_margins = chop_margins
        self.random_seed = random_seed

    def chop(self, neural_df, chop_fields):
        """Chops a trialized or continuous RDS DataFrame.

        Parameters
        ----------
        neural_df : pd.DataFrame
            A continuous or trialized DataFrame from RDS.
        chop_fields : str or list of str
            `signal_type` or list of `signal_type`s in neural_df to chop

        Returns
        -------
        dict of np.array
            A data_dict of the chopped data. Consists of a dictionary
            with data tags mapping to 3D numpy arrays with dimensions
            corresponding to samples x time x features.
        """
        # Set the random seed for the offset
        if self.random_seed is not None:
            np.random.seed(self.random_seed)

        if type(chop_fields) != list:
            chop_fields = [chop_fields]

        # Get info about the column groups to be chopped
        data_fields = sorted(chop_fields)
        get_field_dim = lambda field: getattr(neural_df, field).shape[1] if len(getattr(neural_df, field).shape) > 1 else 1
        data_dims = [get_field_dim(f) for f in data_fields]
        data_splits = np.cumsum(data_dims[:-1])
        # Report information about the fields that are being chopped
        logger.info(f'Chopping data field(s) {data_fields} with dimension(s) {data_dims}.')

        # Calculate bin widths and set up segments for chopping
        if 'trial_id' in neural_df:
            # Trialized data
            bin_width = neural_df.clock_time[1] - neural_df.clock_time[0]
            segments = neural_df.groupby('trial_id')
        else:
            # Continuous data
            bin_width = neural_df.index[1] - neural_df.index[0]
            if np.any(np.isnan(neural_df[chop_fields])):
                splits = np.where(neural_df[chop_fields].sum(axis=1, min_count=1).isna().diff())[0].tolist() + [len(neural_df)]
                segments = {n: neural_df.iloc[splits[n]:splits[n+1]].reset_index() for n in range(0, len(splits) - 1, 2)}
            else:
                segments = {1: neural_df.reset_index()}.items()

        # Calculate the number of bins to use for chopping parameters
        window = int(self.window / bin_width)
        overlap = int(self.overlap / bin_width)
        chop_margins_td = pd.to_timedelta(
            self.chop_margins * bin_width, unit='ms')

        # Get correct offset based on data type
        if 'trial_id' in neural_df:
            # Trialized data
            max_offset = int(self.max_offset / bin_width)
            max_offset_td = self.max_offset
            get_offset = lambda: np.random.randint(max_offset+1)
        else:
            # Continuous data
            max_offset = 0
            max_offset_td = pd.to_timedelta(max_offset)
            get_offset = lambda: 0
            if self.max_offset > pd.to_timedelta(0):
                # Doesn't make sense to use offset on continuous data
                logger.info("Ignoring offset for continuous data.")

        def to_ms(timedelta):
            return int(timedelta.total_seconds() * 1000)

        # Log information about the chopping to be performed
        chop_message = ' - '.join([
            'Chopping data',
            f'Window: {window} bins, {to_ms(self.window)} ms',
            f'Overlap: {overlap} bins, {to_ms(self.overlap)} ms',
            f'Max offset: {max_offset} bins, {to_ms(max_offset_td)} ms',
            f'Chop margins: {self.chop_margins} bins, {to_ms(chop_margins_td)} ms',
        ])
        logger.info(chop_message)

        # Iterate through segments, which can be trials or continuous data
        data_dict = defaultdict(list)
        segment_records = []
        for segment_id, segment_df in segments:
            # Get the data from all of the column groups to extract
            data_arrays = [getattr(segment_df, f).to_numpy() if len(getattr(segment_df, f).shape) > 1
                           else getattr(segment_df, f).to_numpy()[:, None] for f in data_fields]
            # Concatenate all data types into a single segment array
            segment_array = np.concatenate(data_arrays, axis=1)
            if self.chop_margins > 0:
                # Add padding to segment if we are using chop margins
                seg_dim = segment_array.shape[1]
                pad = np.full((self.chop_margins, seg_dim), 0.0001)
                segment_array = np.concatenate([pad, segment_array, pad])
            # Sample an offset for this segment
            offset = get_offset()
            # Chop all of the data in this segment
            chops = chop_data(
                segment_array,
                overlap + 2*self.chop_margins,
                window + 2*self.chop_margins,
                offset)
            # Split the chops back up into the original fields
            data_chops = np.split(chops, data_splits, axis=2)
            # Create the data_dict with LFADS input names
            for field, data_chop in zip(data_fields, data_chops):
                data_dict[field].append(data_chop)
            # Keep a record to represent each original segment
            seg_rec = SegmentRecord(
                segment_id,
                segment_df.clock_time,
                offset,
                len(chops),
                overlap)
            segment_records.append(seg_rec)
        # Store the information for reassembling segments
        self.segment_records = segment_records
        # Consolidate data from all segments into a single array
        data_dict = {name: np.concatenate(c) for name, c in data_dict.items()}
        # Report diagnostic info
        dict_key = list(data_dict.keys())[0]
        n_chops = len(data_dict[dict_key])
        n_segments = len(segment_records)
        logger.info(f'Created {n_chops} chops from {n_segments} segment(s).')

        return data_dict

    def merge(self, chopped_data, smooth_pwr=2):
        """Merges chopped data to reconstruct the original input
        sequence

        Parameters
        ----------
        chopped_data : dict
            Dict mapping the keys to chopped 3d numpy arrays
        smooth_pwr : float, optional
            The power to use for smoothing. See `merge_chops`
            function for more details, by default 2

        Returns
        -------
        pd.DataFrame
            A merged DataFrame indexed by the clock time of the original
            chops. Columns are multiindexed using `fields_map`.
            Unmodeled data is indicated by NaNs.
        """
        # Get the desired arrays from the output
        output_fields = sorted(chopped_data.keys())
        output_arrays = [chopped_data[f] for f in output_fields]
        # Keep track of boundaries between the different signals
        output_dims = [a.shape[-1] for a in output_arrays]
        # Concatenate the output arrays for more efficient merging
        output_full = np.concatenate(output_arrays, axis=-1)
        # Get info for separating the chops related to each segment
        seg_splits = np.cumsum([s.n_chops for s in self.segment_records])[:-1]
        # Separate out the chops for each segment
        seg_chops = np.split(output_full, seg_splits, axis=0)
        # Reconstruct the segment DataFrames
        segment_dfs = [record.rebuild_segment(chops, smooth_pwr) \
            for record, chops in zip(self.segment_records, seg_chops)]
        # Concatenate the segments with clock_time indices
        merged_df = pd.concat(segment_dfs)
        # Add multiindexed columns
        midx_tuples = [(sig, f'{i:04}') \
            for sig, dim in zip(output_fields, output_dims) \
                for i in range(dim)]
        merged_df.columns = pd.MultiIndex.from_tuples(midx_tuples)

        return merged_df

def chop_data(data, overlap, window, offset=0):
    """Rearranges an array of continuous data into overlapping segments.

    This low-level function takes a 2-D array of features measured
    continuously through time and breaks it up into a 3-D array of
    partially overlapping time segments.

    Parameters
    ----------
    data : np.ndarray
        A TxN numpy array of N features measured across T time points.
    overlap : int
        The number of points to overlap between subsequent segments.
    window : int
        The number of time points in each segment.
    Returns
    -------
    np.ndarray
        An SxTxN numpy array of S overlapping segments spanning
        T time points with N features.

    See Also
    --------
    chop.merge_chops : Performs the opposite of this operation.
    """
    # Random offset breaks temporal connection between trials and chops
    offset_data = data[offset:]
    shape = (
        int((offset_data.shape[0] - window)/(window - overlap)) + 1,
        window,
        offset_data.shape[-1],
    )
    strides = (
        offset_data.strides[0]*(window - overlap),
        offset_data.strides[0],
        offset_data.strides[1],
    )
    chopped = np.lib.stride_tricks.as_strided(
        offset_data, shape=shape, strides=strides).copy().astype('f')
    return chopped

def merge_chops(data, overlap, orig_len=None, smooth_pwr=2):
    """Merges an array of overlapping segments back into continuous data.
    This low-level function takes a 3-D array of partially overlapping
    time segments and merges it back into a 2-D array of features measured
    continuously through time.

    Parameters
    ----------
    data : np.ndarray
        An SxTxN numpy array of S overlapping segments spanning
        T time points with N features.
    overlap : int
        The number of overlapping points between subsequent segments.
    orig_len : int, optional
        The original length of the continuous data, by default None
        will cause the length to depend on the input data.
    smooth_pwr : float, optional
        The power of smoothing. To keep only the ends of chops and
        discard the beginnings, use np.inf. To linearly blend the
        chops, use 1. Raising above 1 will increasingly prefer the
        ends of chops and lowering towards 0 will increasingly
        prefer the beginnings of chops (not recommended). To use
        only the beginnings of chops, use 0 (not recommended). By
        default, 2 slightly prefers the ends of segments.
    Returns
    -------
    np.ndarray
        A TxN numpy array of N features measured across T time points.

    See Also
    --------
    chop.chop_data : Performs the opposite of this operation.
    """
    if smooth_pwr < 1:
        logger.warning('Using `smooth_pwr` < 1 for merging '
            'chops is not recommended.')

    merged = []
    full_weight_len = data.shape[1] - 2*overlap
    # Create x-values for the ramp
    x = np.linspace(1/overlap, 1 - 1/overlap, overlap) \
        if overlap != 0 else np.array([])
    # Compute a power-function ramp to transition
    ramp = 1 - x ** smooth_pwr
    ramp = np.expand_dims(ramp, axis=-1)
    # Compute the indices to split up each chop
    split_ixs = np.cumsum([overlap, full_weight_len])
    for i in range(len(data)):
        # Split the chop into overlapping and non-overlapping
        first, middle, last = np.split(data[i], split_ixs)
        # Ramp each chop and combine it with the previous chop
        if i == 0:
            last = last * ramp
        elif i == len(data) - 1:
            first = first * (1-ramp) + merged.pop(-1)
        else:
            first = first * (1-ramp) + merged.pop(-1)
            last = last * ramp
        # Track all of the chops in a list
        merged.extend([first, middle, last])

    merged = np.concatenate(merged)
    # Indicate unmodeled data with NaNs
    if orig_len is not None and len(merged) < orig_len:
        nans = np.full((orig_len-len(merged), merged.shape[1]), np.nan)
        merged = np.concatenate([merged, nans])

    return merged


text:
import json
import numpy as np

PREFIX = "("
SUFFIX = ")"

def compute_initial_prior(tokens, transition_weights):
    states = []
    weights = []
    for i, token in enumerate(tokens):
        if token.startswith(PREFIX):
            states.append(i)
            weights.append(sum(transition_weights[i].values()))
    return states, weights

class LoadedModel:
    def __init__(self, tokens, transition_weights):
        self._tokens = tokens
        self._token_size = len(tokens[0])

        initial_states, initial_weights = compute_initial_prior(
            tokens, transition_weights
        )
        self._initial_states = initial_states
        self._initial_weights = initial_weights

        self._transition_destinations = [
            list(transition.keys()) for transition in transition_weights
        ]
        self._transition_weights = [
            list(transition.values()) for transition in transition_weights
        ]

    @property
    def token_size(self):
        return self._token_size

    @property
    def tokens(self):
        return self._tokens

    @property
    def initial_prior(self):
        return self._initial_states, self._initial_weights

    def posterior(self, i):
        return self._transition_destinations[i], self._transition_weights[i]

    def save(self, store):
        self._save_metadata(store)
        self._save_initial_prior(store)
        self._save_posterior(store)

    def _save_metadata(self, store):
        metadata = {
            "prefix": PREFIX,
            "suffix": SUFFIX,
            "token_size": self._token_size,
        }
        store.create_dataset(
            "metadata",
            data=json.dumps(metadata),
        )

    def _save_initial_prior(self, store):
        # Save as a sparse vector.
        data = list(zip(self._initial_states, self._initial_weights))

        store.create_dataset(
            "initial_prior_data",
            data=np.array(data, dtype=np.int32),
            scaleoffset=0,
            compression=1,
        )

    def _save_posterior(self, store):
        # Save as a sparse matrix in a flattened LIL format.
        rows = []
        data = []
        for i in range(len(self._tokens)):
            destinations, weights = self.posterior(i)
            start = len(data)
            end = start + len(destinations)
            for j, weight in zip(destinations, weights):
                data.append((j, weight))
            rows.append((start, end))

        store.create_dataset(
            "tokens",
            data=np.array(self._tokens, dtype=f"S{self._token_size}"),
            compression=True,
        )
        store.create_dataset(
            "transition_rows",
            data=np.array(rows, dtype=np.int32),
            chunks=(4096, 2),
            scaleoffset=0,
            compression=1,
        )
        store.create_dataset(
            "transition_data",
            data=np.array(data, dtype=np.int32),
            chunks=(4096, 2),
            scaleoffset=0,
            compression=1,
        )

class StoredModel:
    def __init__(self, store):
        self._store = store
        self._metadata = json.loads(store["metadata"][()])
        self._token_size = self._metadata["token_size"]

        self._tokens = [token.decode("ascii") for token in self._store["tokens"]]
        self._transition_rows = self._store["transition_rows"][:]
        self._transition_data = self._store["transition_data"]

        initial_prior_data = self._store["initial_prior_data"]
        self._initial_states = initial_prior_data[:, 0]
        self._initial_weights = initial_prior_data[:, 1]

    @property
    def token_size(self):
        return self._token_size

    @property
    def tokens(self):
        return self._tokens

    @property
    def initial_prior(self):
        return self._initial_states, self._initial_weights

    def posterior(self, i):
        start, end = self._transition_rows[i]
        data = self._transition_data[start:end]
        destinations = data[:, 0]
        weights = data[:, 1]
        return destinations, weights

max_stars_repo_name: snsinfu/web-wordgen

max_issues_repo_name: snsinfu/web-wordgen

max_forks_repo_name: snsinfu/web-wordgen


text:
import os
import unittest

import pandas as pd
import numpy as np
from pandas.testing import assert_frame_equal, assert_series_equal

from mavedbconvert import empiric, constants

from tests import ProgramTestCase

class TestEmpiricInit(ProgramTestCase):
    def setUp(self):
        super().setUp()
        self.path = os.path.join(self.data_dir, "empiric", "empiric.xlsx")

    def test_offset_inframe(self):
        empiric.Empiric(src=self.path, wt_sequence="ATC", offset=3)

    def test_error_offset_not_inframe(self):
        with self.assertRaises(ValueError):
            empiric.Empiric(src=self.path, wt_sequence="ATC", offset=1)

    def test_error_noncoding(self):
        with self.assertRaises(ValueError):
            empiric.Empiric(src=self.path, wt_sequence="ATC", is_coding=False)

class TestInferProEvent(unittest.TestCase):
    def test_infers_equal_event(self):
        self.assertEqual(
            empiric.infer_pro_substitution(mut_aa="V", wt_aa="v", codon_pos=0),
            "p.Val1=",
        )

    def test_infers_sub_event_event(self):
        self.assertEqual(
            empiric.infer_pro_substitution(mut_aa="V", wt_aa="F", codon_pos=0),
            "p.Phe1Val",
        )

    def test_converts_triple_q_to_Xaa(self):
        self.assertEqual(
            empiric.infer_pro_substitution(mut_aa="?", wt_aa="V", codon_pos=0),
            "p.Val1Xaa",
        )

class TestInferNTEvent(unittest.TestCase):
    def test_infers_equal_event(self):
        self.assertEqual(
            empiric.infer_nt_substitution(wt_codon="aaa", mut_codon="AAA", codon_pos=0),
            "c.[1=;2=;3=]",
        )

    def test_infers_sub_event_event(self):
        self.assertEqual(
            empiric.infer_nt_substitution(wt_codon="ATC", mut_codon="GTA", codon_pos=0),
            "c.[1A>G;2=;3C>A]",
        )

    def test_adds_codon_pos_multiplied_by_3_to_position(self):
        self.assertEqual(
            empiric.infer_nt_substitution(wt_codon="ATC", mut_codon="GTA", codon_pos=1),
            "c.[4A>G;5=;6C>A]",
        )

class TestEmpiric(ProgramTestCase):
    def setUp(self):
        super().setUp()
        self.input = os.path.join(self.data_dir, "empiric", "empiric.xlsx")
        self.empiric = empiric.Empiric(
            src=self.input, wt_sequence="AAA", one_based=False
        )

    def test_error_missing_amino_acid(self):
        for nan in constants.extra_na:
            df = pd.DataFrame({"Position": [0], "Amino Acid": [nan], "row_num": [0]})
            self.empiric.validate_columns(df)
            with self.assertRaises(ValueError):
                self.empiric.parse_row(row=df.iloc[0, :])

    def test_value_error_codon_doesnt_match_aa_column(self):
        with self.assertRaises(ValueError):
            df = pd.DataFrame(
                {"Position": [0], "Amino Acid": ["V"], "Codon": ["AAT"], "row_num": [0]}
            )
            self.empiric.validate_columns(df)
            self.empiric.parse_row(row=df.iloc[0, :])

    def test_error_infer_nt_true_but_missing_codon_value(self):
        for nan in constants.extra_na:
            df = pd.DataFrame(
                {"Position": [0], "Amino Acid": ["N"], "row_num": [0], "Codon": [nan]}
            )
            self.empiric.validate_columns(df)
            with self.assertRaises(ValueError):
                self.empiric.parse_row(row=df.iloc[0, :])

    def test_index_error_negative_position(self):
        df = pd.DataFrame(
            {"Position": [0], "Amino Acid": ["K"], "row_num": [0], "Codon": ["AAA"]}
        )
        self.empiric.validate_columns(df)
        self.empiric.one_based = True
        with self.assertRaises(IndexError):
            self.empiric.parse_row(row=df.iloc[0, :])

    def test_index_error_out_of_codon_bounds(self):
        df = pd.DataFrame(
            {"Position": [56], "Amino Acid": ["K"], "row_num": [0], "Codon": ["AAA"]}
        )
        self.empiric.validate_columns(df)
        with self.assertRaises(IndexError):
            self.empiric.parse_row(row=df.iloc[0, :])

    def test_amino_acid_column_is_case_insensitive(self):
        df = pd.DataFrame(
            {"Position": [0], "Amino Acid": ["v"], "row_num": [0], "Codon": ["GTA"]}
        )
        self.empiric.validate_columns(df)
        _, hgvs_pro = self.empiric.parse_row(row=df.iloc[0, :])
        self.assertEqual(hgvs_pro, "p.Lys1Val")

    def test_infers_hgvs_pro_event_from_one_based_position(self):
        df = pd.DataFrame(
            {"Position": [1], "Amino Acid": ["V"], "row_num": [0], "Codon": ["GTA"]}
        )
        self.empiric.validate_columns(df)
        self.empiric.one_based = True
        _, hgvs_pro = self.empiric.parse_row(row=df.iloc[0, :])
        self.assertEqual(hgvs_pro, "p.Lys1Val")

    def test_infers_hgvs_pro_event_from_zero_based_position(self):
        df = pd.DataFrame(
            {"Position": [1], "Amino Acid": ["V"], "row_num": [0], "Codon": ["GTA"]}
        )
        self.empiric.validate_columns(df)
        self.empiric.wt_sequence = "GTAAAA"
        self.empiric.one_based = False
        _, hgvs_pro = self.empiric.parse_row(row=df.iloc[0, :])
        self.assertEqual(hgvs_pro, "p.Lys2Val")

    def test_protein_output_is_singular_when_inferring_nt(self):
        df = pd.DataFrame(
            {"Position": [0], "Amino Acid": ["V"], "row_num": [0], "Codon": ["GTA"]}
        )
        self.empiric.validate_columns(df)
        hgvs_nt, hgvs_pro = self.empiric.parse_row(row=df.iloc[0, :])
        self.assertEqual(hgvs_nt, "c.[1A>G;2A>T;3=]")
        self.assertEqual(hgvs_pro, "p.Lys1Val")

    def test_hgvs_nt_is_none_when_codon_is_not_in_axes(self):
        df = pd.DataFrame({"Position": [0], "Amino Acid": ["V"], "row_num": [0]})
        self.empiric.validate_columns(df)
        hgvs_nt, _ = self.empiric.parse_row(row=df.iloc[0, :])
        self.assertIsNone(hgvs_nt)

    def test_correctly_infers_hgvs_nt_positions_when_zero_based(self):
        df = pd.DataFrame(
            {"Position": [1], "Amino Acid": ["V"], "row_num": [0], "Codon": ["GTA"]}
        )
        self.empiric.validate_columns(df)
        self.empiric.one_based = False
        self.empiric.wt_sequence = "GGGAAT"
        hgvs_nt, _ = self.empiric.parse_row(row=df.iloc[0, :])
        self.assertEqual(hgvs_nt, "c.[4A>G;5A>T;6T>A]")

    def test_correctly_infers_hgvs_nt_positions_when_one_based(self):
        df = pd.DataFrame({"Position": [1], "Amino Acid": ["N"], "Codon": ["AAT"]})
        self.empiric.validate_columns(df)
        self.empiric.one_based = True
        self.empiric.wt_sequence = "GTAAAA"
        hgvs_nt, _ = self.empiric.parse_row(row=df.iloc[0, :])
        self.assertEqual(hgvs_nt, "c.[1G>A;2T>A;3A>T]")

class TestEmpiricValidateColumns(ProgramTestCase):
    def setUp(self):
        super().setUp()
        self.input = os.path.join(self.data_dir, "empiric", "empiric.xlsx")
        self.empiric = empiric.Empiric(
            src=self.input, wt_sequence="AAA", one_based=False
        )

    def test_error_cannot_find_case_insensitive_aa_column(self):
        df = pd.DataFrame({"Position": [1], "aa": ["N"], "Codon": ["AAT"]})
        with self.assertRaises(ValueError):
            self.empiric.validate_columns(df)

    def test_error_cannot_find_case_insensitive_position_column(self):
        df = pd.DataFrame({"pos": [1], "Amino Acid": ["N"], "Codon": ["AAT"]})
        with self.assertRaises(ValueError):
            self.empiric.validate_columns(df)

    def test_sets_codon_column_as_none_if_not_present(self):
        df = pd.DataFrame({"Position": [1], "Amino Acid": ["N"]})
        self.empiric.validate_columns(df)
        self.assertEqual(self.empiric.codon_column, None)

    def test_sets_codon_column_if_present(self):
        df = pd.DataFrame({"Position": [1], "Amino Acid": ["N"], "Codon": ["AAT"]})
        self.empiric.validate_columns(df)
        self.assertEqual(self.empiric.codon_column, "Codon")

    def test_sets_position_column(self):
        df = pd.DataFrame({"Position": [1], "Amino Acid": ["N"], "Codon": ["AAT"]})
        self.empiric.validate_columns(df)
        self.assertEqual(self.empiric.position_column, "Position")

    def test_sets_aa_column(self):
        df = pd.DataFrame({"Position": [1], "amino acid": ["N"], "Codon": ["AAT"]})
        self.empiric.validate_columns(df)
        self.assertEqual(self.empiric.aa_column, "amino acid")

class TestEmpiricParseScoresInput(ProgramTestCase):
    def setUp(self):
        super().setUp()
        self.input = os.path.join(self.data_dir, "empiric", "empiric.xlsx")
        self.empiric = empiric.Empiric(
            src=self.input,
            wt_sequence="AAA",
            one_based=False,
            input_type="scores",
            score_column="A",
        )

    def test_deletes_position_amino_acid_codon_row_num_columns(self):
        df = pd.DataFrame(
            {"Position": [0], "Amino Acid": ["N"], "Codon": ["AAT"], "A": [1.2]}
        )
        result = self.empiric.parse_input(df)
        self.assertNotIn("Position", result.columns)
        self.assertNotIn("Amino Acid", result.columns)
        self.assertNotIn("Codon", result.columns)
        self.assertNotIn("row_num", result.columns)

    def test_keeps_additional_non_score_columns(self):
        df = pd.DataFrame(
            {
                "Position": [0],
                "Amino Acid": ["N"],
                "Codon": ["AAT"],
                "A": [1.2],
                "B": [2.4],
            }
        )
        result = self.empiric.parse_input(df)
        self.assertIn("B", result.columns)

    def test_renames_score_column_to_score_and_drops_original(self):
        df = pd.DataFrame(
            {
                "Position": [0],
                "Amino Acid": ["N"],
                "Codon": ["AAT"],
                "A": [1.2],
                "B": [2.4],
            }
        )
        result = self.empiric.parse_input(df)
        self.assertListEqual(list(df["A"]), list(result["score"]))
        self.assertIn("B", result.columns)
        self.assertNotIn("A", result.columns)

    def test_sets_hgvs_pro_column(self):
        df = pd.DataFrame(
            {
                "Position": [0],
                "Amino Acid": ["N"],
                "Codon": ["AAT"],
                "A": [1.2],
                "B": [2.4],
            }
        )
        result = self.empiric.parse_input(df)
        self.assertEqual(result[constants.pro_variant_col].values[0], "p.Lys1Asn")

    def test_correctly_infers_hgvs_nt_column_when_codon_column_present(self):
        df = pd.DataFrame(
            {
                "Position": [1],
                "Amino Acid": ["N"],
                "Codon": ["AAT"],
                "A": [1.2],
                "B": [2.4],
            }
        )
        self.empiric.one_based = False
        self.empiric.wt_sequence = "GGGAAA"
        result = self.empiric.parse_input(df)
        self.assertEqual(result[constants.nt_variant_col].values[0], "c.[4=;5=;6A>T]")

    def test_orders_columns(self):
        df = pd.DataFrame(
            {
                "Position": [0],
                "Amino Acid": ["N"],
                "Codon": ["AAT"],
                "A": [1.2],
                "B": [2.4],
            }
        )
        result = self.empiric.parse_input(df)
        self.assertEqual(list(result.columns).index(constants.nt_variant_col), 0)
        self.assertEqual(list(result.columns).index(constants.pro_variant_col), 1)
        self.assertEqual(list(result.columns).index(constants.mavedb_score_column), 2)

    def test_removes_null_columns(self):
        df = pd.DataFrame(
            {
                "Position": [0],
                "Amino Acid": ["N"],
                "Codon": ["AAT"],
                "B": [None],
                "A": [2.4],
            }
        )
        result = self.empiric.parse_input(df)
        self.assertNotIn("B", result.columns)

    def test_drops_nt_when_codon_column_is_not_provided(self):
        df = pd.DataFrame(
            {"Position": [0], "Amino Acid": ["N"], "A": [1.2], "B": [2.4]}
        )
        result = self.empiric.parse_input(df)
        self.assertNotIn(constants.nt_variant_col, result.columns)

    def test_drops_non_numeric_columns(self):
        df = pd.DataFrame(
            {
                "Position": [0],
                "Amino Acid": ["N"],
                "Codon": ["AAT"],
                "A": [1.2],
                "B": ["a"],
            }
        )
        result = self.empiric.parse_input(df)
        self.assertNotIn("B", result.columns)

    def test_keeps_int_type_as_int(self):
        df = pd.DataFrame(
            {"Position": [0], "Amino Acid": ["N"], "Codon": ["AAT"], "A": [1]}
        )
        result = self.empiric.parse_input(df)
        self.assertTrue(
            np.issubdtype(
                result[constants.mavedb_score_column].values[0], np.signedinteger
            )
        )

class TestEmpiricParseCountsInput(ProgramTestCase):
    def setUp(self):
        super().setUp()
        self.input = os.path.join(self.data_dir, "empiric", "empiric.xlsx")
        self.empiric = empiric.Empiric(
            src=self.input,
            wt_sequence="AAA",
            one_based=False,
            input_type="counts",
            score_column="A",
        )

    def test_orders_columns(self):
        df = pd.DataFrame(
            {
                "Position": [0],
                "Amino Acid": ["N"],
                "Codon": ["AAT"],
                "A": [1.2],
                "B": [2.4],
            }
        )
        result = self.empiric.parse_input(df)
        self.assertEqual(list(result.columns).index(constants.nt_variant_col), 0)
        self.assertEqual(list(result.columns).index(constants.pro_variant_col), 1)
        self.assertEqual(list(result.columns).index("A"), 2)
        self.assertEqual(list(result.columns).index("B"), 3)

class TestEmpiricLoadInput(ProgramTestCase):
    def setUp(self):
        super().setUp()
        self.excel_path = os.path.join(self.data_dir, "empiric", "empiric.xlsx")
        self.excel_header_footer_path = os.path.join(
            self.data_dir, "empiric", "empiric_header_footer.xlsx"
        )
        self.csv_path = os.path.join(self.data_dir, "empiric", "tmp.csv")
        self.tsv_path = os.path.join(self.data_dir, "empiric", "tmp.tsv")
        self.excel_multisheet_path = os.path.join(
            self.data_dir, "empiric", "empiric_multisheet.xlsx"
        )

    def test_extra_na_load_as_nan(self):
        for value in constants.extra_na:
            df = pd.read_excel(self.excel_path, engine="openpyxl")
            df["A"] = [value] * len(df)
            df.to_csv(self.csv_path, index=False)
            e = empiric.Empiric(
                src=self.csv_path,
                wt_sequence="TTTTCTTATTGT",
                score_column="col_A",
                input_type=constants.score_type,
                one_based=False,
            )
            result = e.load_input_file()
            expected = pd.Series([np.NaN] * len(df), index=df.index, name="A")
            assert_series_equal(result["A"], expected)

    def test_loads_first_sheet_by_default(self):
        p = empiric.Empiric(
            src=self.excel_multisheet_path,
            wt_sequence="TTTTCTTATTGT",
            score_column="score",
            input_type=constants.score_type,
        )
        result = p.load_input_file()
        expected = pd.read_excel(
            self.excel_multisheet_path, na_values=constants.extra_na, engine="openpyxl"
        )
        assert_frame_equal(result, expected)

    def test_loads_correct_sheet(self):
        p = empiric.Empiric(
            src=self.excel_multisheet_path,
            wt_sequence="TTTTCTTATTGT",
            score_column="col_A",
            input_type=constants.score_type,
            one_based=False,
            sheet_name="Sheet3",
        )
        result = p.load_input_file()
        expected = pd.read_excel(
            self.excel_multisheet_path,
            na_values=constants.extra_na,
            sheet_name="Sheet3",
            engine="openpyxl",
        )
        assert_frame_equal(result, expected)

    def test_error_missing_sheet(self):
        p = empiric.Empiric(
            src=self.excel_multisheet_path,
            wt_sequence="TTTTCTTATTGT",
            score_column="col_A",
            input_type=constants.score_type,
            one_based=False,
            sheet_name="BadSheet",
        )
        with self.assertRaises(KeyError):
            p.load_input_file()

    def test_handles_csv(self):
        df = pd.read_excel(self.excel_path, engine="openpyxl")
        df.to_csv(self.csv_path, index=False, sep=",")
        e = empiric.Empiric(
            src=self.csv_path,
            wt_sequence="TTTTCTTATTGT",
            score_column="col_A",
            input_type=constants.score_type,
            one_based=False,
        )
        result = e.load_input_file()
        assert_frame_equal(result, df)

    def test_loads_with_skipped_rows(self):
        p = empiric.Empiric(
            src=self.excel_header_footer_path,
            wt_sequence="TTTTCTTATTGT",
            score_column="col_A",
            input_type=constants.score_type,
            one_based=False,
            skip_header_rows=2,
            skip_footer_rows=2,
        )
        result = p.load_input_file()
        df = pd.read_excel(self.excel_path, engine="openpyxl")
        assert_frame_equal(result, df)

    def test_handles_tsv(self):
        df = pd.read_excel(self.excel_path, engine="openpyxl")
        df.to_csv(self.tsv_path, index=False, sep="\t")
        e = empiric.Empiric(
            src=self.tsv_path,
            wt_sequence="TTTTCTTATTGT",
            score_column="col_A",
            input_type=constants.score_type,
            one_based=False,
        )
        result = e.load_input_file()
        assert_frame_equal(result, df)

    def test_error_position_not_in_columns(self):
        df = pd.read_excel(self.excel_path, engine="openpyxl")
        df = df.drop(columns=["Position"])
        df.to_csv(self.csv_path, index=False, sep="\t")
        with self.assertRaises(ValueError):
            e = empiric.Empiric(
                src=self.csv_path,
                wt_sequence="TTTTCTTATTGT",
                score_column="col_A",
                input_type=constants.score_type,
                one_based=False,
            )
            e.load_input_file()

    def test_error_amino_acid_not_in_columns(self):
        df = pd.read_excel(self.excel_path, engine="openpyxl")
        df = df.drop(columns=["Amino Acid"])
        df.to_csv(self.csv_path, index=False, sep="\t")
        with self.assertRaises(ValueError):
            e = empiric.Empiric(
                src=self.csv_path,
                wt_sequence="TTTTCTTATTGT",
                score_column="col_A",
                input_type=constants.score_type,
                one_based=False,
            )
            e.load_input_file()

    def test_not_scores_column_but_input_type_is_scores(self):
        with self.assertRaises(ValueError):
            empiric.Empiric(
                src=self.csv_path,
                wt_sequence="TTTTCTTATTGT",
                score_column=None,
                input_type=constants.score_type,
                one_based=False,
            )

    def test_applies_offset_to_position_column(self):
        e = empiric.Empiric(
            src=self.excel_path,
            wt_sequence="TTTTCTTATTGT",
            score_column="col_A",
            input_type=constants.score_type,
            one_based=False,
            offset=-9,
        )
        result = e.load_input_file()
        self.assertListEqual(list(result[e.position_column]), [3, 4, 5])

class TestEmpiricConvert(ProgramTestCase):
    def setUp(self):
        super().setUp()
        self.excel_path = os.path.join(self.data_dir, "empiric", "empiric.xlsx")
        self.expected = os.path.join(self.data_dir, "empiric", "empiric_expected.csv")
        self.empiric = empiric.Empiric(
            src=self.excel_path,
            wt_sequence="TTTTCTTATTGT",
            score_column="col_A",
            input_type=constants.score_type,
            one_based=False,
        )

    def test_saves_to_dst(self):
        self.empiric.convert()
        self.assertTrue(os.path.isfile(self.empiric.output_file))

    def test_integration(self):
        self.empiric = empiric.Empiric(
            src=self.excel_path,
            wt_sequence="TCTTATTGT",
            score_column="col_A",
            input_type=constants.score_type,
            one_based=False,
        )
        self.empiric.convert()
        assert_frame_equal(
            pd.read_csv(self.empiric.output_file, delimiter=","),
            pd.read_csv(self.expected, delimiter=","),
        )

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

max_stars_repo_name: VariantEffect/mavedbconvert

max_issues_repo_name: VariantEffect/mavedbconvert

max_forks_repo_name: VariantEffect/mavedbconvert


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

tossCount = int(input("Please give me the number of tosses to perform:"))
experimentsCount = [30,100,250,1000,5000]

randomTab = np.array([])
for x in range(experimentsCount[4]):
    randomTab = np.append(randomTab,np.random.randint(0, 2, tossCount))

randomTab = randomTab.reshape((-1, tossCount))
randomTab = np.sum(randomTab,axis=1)
print(randomTab)

firstExperimentTab = randomTab[:experimentsCount[0]]
secondExperimentTab = randomTab[:experimentsCount[1]]
thirdExperimentTab = randomTab[:experimentsCount[2]]
fourthExperimentTab = randomTab[:experimentsCount[3]]
fifthExperimentTab = randomTab

bins = np.arange(tossCount+1)
plt.figure(1)
plt.subplot(221)
plt.hist(firstExperimentTab,bins=bins, histtype='bar')
plt.subplot(222)
plt.hist(secondExperimentTab,bins=bins, histtype='bar')
plt.subplot(223)
plt.hist(thirdExperimentTab,bins=bins, histtype='bar')
plt.subplot(224)
plt.hist(fourthExperimentTab,bins=bins, histtype='bar')
plt.figure(2)
plt.xlabel("Ile reszek w " + str(tossCount) + " rzutach.")
plt.ylabel("Liczba eksperymentów")
plt.hist(fifthExperimentTab,bins=bins, histtype='bar')
plt.show()


text:
import json
import os
import xml.etree.ElementTree as ET
from collections import defaultdict
from pathlib import Path
from typing import Sequence

import numpy as np
from skimage.measure import points_in_poly

class Annotation:
    '''Represents an annotation using a coordinates array of shape (n, 2).'''

    def __init__(self, name: str, coords: np.ndarray) -> None:
        '''Initialize a Annotation object.

        - Args
            name: Name of the Annotation object
            coords: Numpy array containing coordinates;
                    [(x1,y1), (x2,y2), ... ,(xn,yn)] of shape (n, 2)

        - Returns
            None
        '''
        self.name = name
        self.coords = coords

    def coordinates(self) -> None:
        '''Returns annotation coordinates; vertices'''
        return self.coords

    def __repr__(self) -> str:
        return self.name

    def does_contain(self, points: list) -> Sequence[bool]:
        '''Check if the Annotation object contains the given coordinate.

        - Args
            points: A list of points to perform the test

        - Returns
            True if the given coorindate is inside the Annotation object,
            False otherwise
        '''
        return points_in_poly(points, self.coords)

class LesionAnnotations:
    '''Represents an annotation drawn on a whole slide image.'''

    def __init__(self, annot_path: str) -> None:
        '''Init LesionAnnotations.'''

        self.annot_path = annot_path
        self.annot_fname = Path(self.annot_path).name
        self.patient_id = self.annot_fname.rstrip('.json')

        with open(annot_path, 'r', encoding='utf-8') as f:
            lesion_annots = json.load(f)

        # Parse positive annotations
        self.pos_annots = []
        for annot in lesion_annots['pos']:
            annot_name = annot['name']
            coords = annot['coords'] # annot['vertices'], originally in ncrf code
            pos_annot = Annotation(name=annot_name, coords=coords)
            self.pos_annots.append(pos_annot)

        # Parse negative annotations
        self.neg_annots = []
        for annot in lesion_annots['neg']:
            annot_name = annot['name']
            coords = annot['coords']
            neg_annot = Annotation(name=annot_name, coords=coords)
            self.neg_annots.append(neg_annot)

        self.annots_dict = {
            'pos': self.pos_annots,
            'neg': self.neg_annots,
        }

    def filter_tumor_coords(self, save_path: str,
                            points: np.ndarray, is_pos: bool = True) -> np.ndarray:
        '''Filter tumor coords from the given points.

        - Args
            save_dir: Path to json file to save the tumor coordinates
            points: Roi points to check if each of them is a tumor
            is_pos: True if to filter positive tumor, False otherwise

        - Returns
            A numpy array which contains tumor coordinates; points inside annotations
        '''
        if is_pos:
            annots = self.pos_annots
        else:
            annots = self.neg_annots

        tumor_coords_dict = defaultdict(list)
        tumor_coords_dict['patient_id'] = self.patient_id
        tumor_coords_dict['num_coords'] = 0

        for annot in annots:
            tumor_coords_mask = annot.does_contain(points)
            tumor_coords = points[tumor_coords_mask]
            tumor_coords = tumor_coords.tolist()
            tumor_coords_dict['tumor_coords'].extend(tumor_coords)
            tumor_coords_dict['num_coords'] += len(tumor_coords)

        with open(save_path, 'w+', encoding='utf-8') as f:
            json.dump(tumor_coords_dict, f, indent=4)

        return tumor_coords_dict['tumor_coords']

    def does_contain(self, points: Sequence[tuple], is_pos: bool = True) -> bool:
        '''Check if the Annotation object contains the given coordinate.

        - Args
            points: A list of points to perform the test
            is_pos: True if to check positive tumor, False otherwise

        - Returns
            True if the given coorindate is inside the Annotation object,
            False otherwise
        '''
        num_points = len(points)

        coords_mask_dict = defaultdict(list)
        if num_points > 1:
            if is_pos:
                annots = self.pos_annots
            else:
                annots = self.neg_annots

            for annot in annots:
                coords_mask = annot.does_contain(points=points)
                coords_mask_dict[annot.name] = coords_mask

            return coords_mask_dict

        if is_pos:
            annots = self.pos_annots
        else:
            annots = self.neg_annots

        for annot in annots:
            if annot.does_contain(points=points): # Warn: repair to coord=coord
                return True

        return False

    def __repr__(self):
        return self.annot_fname

    @property
    def coords(self) -> dict:
        '''Return the entire coordinates of shape (num_annotations, 2)

        - Args
            is_pos: True if the target annotations are positive, False otherwise

        - Returns
            A dict containing every positive/negative coordinates of each annotation
        '''
        pos_coords = [annot.coords for annot in self.pos_annots]
        neg_coords = [annot.coords for annot in self.neg_annots]
        coords_dict = {
            'pos': pos_coords,
            'neg': neg_coords,
        }

        return coords_dict

    @property
    def annots(self) -> dict:
        '''Return the entire annotations.

        - Returns
            A dict contaning every positive/negative annotations
        '''
        return self.annots_dict

def xml_to_json(xml_path_in: str, json_path_out: str) -> None:
    '''Convert xml annotation to json

    - Args
        xml_in_path: Path to the input xml annotation file
        json_out_path: Path to save the output json annotation file

    - Returns
        None
    '''

    xml_annot = ET.parse(xml_path_in)
    xml_annot_root = xml_annot.getroot()

    tumor_annot_route = './Annotations/Annotation[@PartOfGroup="Tumor"]'
    tumor_annots = xml_annot_root.findall(tumor_annot_route)

    cls_0_annot_route = './Annotations/Annotation[@PartOfGroup="_0"]'
    cls_0_annots = xml_annot_root.findall(cls_0_annot_route)

    cls_1_annot_route = './Annotations/Annotation[@PartOfGroup="_1"]'
    cls_1_annots = xml_annot_root.findall(cls_1_annot_route)

    cls_2_annot_route = './Annotations/Annotation[@PartOfGroup="_2"]'
    cls_2_annots = xml_annot_root.findall(cls_2_annot_route)

    pos_annots = tumor_annots + cls_0_annots + cls_1_annots
    neg_annots = cls_2_annots

    json_annot = defaultdict(list)
    json_annot['pos'] = []
    json_annot['neg'] = []
    # Parse positive annotations
    for annot in pos_annots:
        coords = annot.findall('./Coordinates/Coordinate')
        x_coords = []
        y_coords = []
        for coord in coords:
            # Parse x coordinates
            x_coord = coord.get('X') # type(x_coord) == 'str'
            x_coord = float(x_coord) # type(x_coord) == 'float'
            x_coord = int(x_coord) # type(x_coord) == 'int'
            x_coords.append(x_coord)

            # Parse y coordinates
            y_coord = coord.get('Y') # type(x_coord) == 'str'
            y_coord = float(y_coord) # type(x_coord) == 'float'
            y_coord = int(y_coord) # type(x_coord) == 'int'
            y_coords.append(y_coord)

        # Concat x=[x1, x2, ...], y=[y1, y2, ...] to coords=[(x1,y1), (x2,y2), ...]
        coords = np.c_[x_coords, y_coords]
        coords = coords.tolist() # because np.ndarray is not json serializable
        annot_name = annot.attrib['Name']
        json_annot['pos'].append({
            'name': f'Annotation{annot_name}',
            'coords': coords,
        })

    # Parse negative annotations
    for annot in neg_annots:
        coords = annot.findall('./Coordinates/Coordinate')
        x_coords = []
        y_coords = []
        for coord in coords:
            # Parse x coordinates
            x_coord = coord.get('X') # type(x_coord) == 'str'
            x_coord = float(x_coord) # type(x_coord) == 'float'
            x_coord = int(x_coord) # type(x_coord) == 'int'
            x_coords.append(x_coord)

            # Parse y coordinates
            y_coord = coord.get('Y') # type(y_coord) == 'str'
            y_coord = float(y_coord) # type(y_coord) == 'float'
            y_coord = int(y_coord) # type(y_coord) == 'int'
            y_coords.append(y_coord)

        # Concat x=[x1, x2, ...], y=[y1, y2, ...] to coords=[(x1,y1), (x2,y2), ...]
        coords = np.c_[x_coords, y_coords]
        coords = coords.tolist() # because np.ndarray is not json serializable
        annot_name = annot.attrib['Name']
        if 'Annotation' not in annot_name:
            annot_name = f'Annotation{annot_name}'

        json_annot['neg'].append({
            'name': annot_name,
            'coords': coords,
        })

    with open(json_path_out, 'w', encoding='utf-8') as f:
        json.dump(json_annot, f, indent=1)

if __name__ == '__main__':
    ROOT_DIR = os.path.abspath('.')
    ANNOTS_DIR = os.path.join(ROOT_DIR, 'annots')
    TRAIN_ANNOTS_DIR = os.path.join(ANNOTS_DIR, 'train')
    TRAIN_XML_ANNOTS_DIR = os.path.join(TRAIN_ANNOTS_DIR, 'xml')
    TRAIN_JSON_ANNOTS_DIR = os.path.join(TRAIN_ANNOTS_DIR, 'json')

    TEST_ANNOTS_DIR = os.path.join(ANNOTS_DIR, 'test')
    TEST_XML_ANNOTS_DIR = os.path.join(TEST_ANNOTS_DIR, 'xml')
    TEST_JSON_ANNOTS_DIR = os.path.join(TEST_ANNOTS_DIR, 'json')

    # Convert training annotations
    os.makedirs(TRAIN_JSON_ANNOTS_DIR, exist_ok=True)
    train_json_annot_fnames = os.listdir(TRAIN_JSON_ANNOTS_DIR)

    train_xml_annot_fnames = os.listdir(TRAIN_XML_ANNOTS_DIR)
    train_xml_annot_fnames = [fname.strip('.xml') for fname in train_xml_annot_fnames]
    train_xml_annot_fnames = sorted(train_xml_annot_fnames)
    for fname in train_xml_annot_fnames:
        if f'{fname}.json' not in train_json_annot_fnames:
            xml_path = os.path.join(TRAIN_XML_ANNOTS_DIR, f'{fname}.xml')
            json_path = os.path.join(TRAIN_JSON_ANNOTS_DIR, f'{fname}.json')
            xml_to_json(xml_path_in=xml_path, json_path_out=json_path)
            print(f'Converted {fname}.xml to {fname}.json')

    # Convert test annotations
    os.makedirs(TEST_JSON_ANNOTS_DIR, exist_ok=True)
    test_json_annot_fnames = os.listdir(TEST_JSON_ANNOTS_DIR)

    test_xml_annot_fnames = os.listdir(TEST_XML_ANNOTS_DIR)
    test_xml_annot_fnames = [fname.strip('.xml') for fname in test_xml_annot_fnames]
    test_xml_annot_fnames = sorted(test_xml_annot_fnames)
    for fname in test_xml_annot_fnames:
        if f'{fname}.json' not in test_json_annot_fnames:
            xml_path = os.path.join(TEST_XML_ANNOTS_DIR, f'{fname}.xml')
            json_path = os.path.join(TEST_JSON_ANNOTS_DIR, f'{fname}.json')
            xml_to_json(xml_path_in=xml_path, json_path_out=json_path)
            print(f'Converted {fname}.xml to {fname}.json')

max_stars_repo_name: timecostslives/fake-doctors

max_issues_repo_name: timecostslives/fake-doctors

max_forks_repo_name: timecostslives/fake-doctors


text:
import torch
import pretrainedmodels
import torch.nn as nn
from torch.nn import init
import torchvision
from efficientnet_pytorch import EfficientNet
import torch.nn.functional as F
import numpy as np
import math

def get_efficientnet(model_name='efficientnet-b0', num_classes=2, pretrained=True):
    if pretrained:
        net = EfficientNet.from_pretrained(model_name)
    else:
        net = EfficientNet.from_name(model_name)
    in_features = net._fc.in_features
    net._fc = nn.Linear(in_features=in_features, out_features=num_classes, bias=True)

    return net

if __name__ == '__main__':
    model, image_size = get_efficientnet(model_name='efficientnet-b0', num_classes=2, pretrained=True), 224

    model = model.to(torch.device('cpu'))
    from torchsummary import summary
    input_s = (3, image_size, image_size)
    print(summary(model, input_s, device='cpu'))

    print(model._modules.items())

    pass